pragent/pilot/opencode_review.py

#!/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 <workdir> --model <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 re
import shutil
import subprocess
import tarfile
import tempfile
import time
import urllib.error
import urllib.request

# Where the factory lives (opencode.json + .opencode/). Default: the pragent
# repo root (this file is at <root>/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 and os.path.isfile(b):
        return b
    found = shutil.which("opencode")
    if found:
        return found
    # last resort: the known linuxbrew path on the dev host.
    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"

# Matches unified-diff new-file path headers: `+++ b/path` (and `+++ /dev/null`
# for deletions, which we skip). Captures the path after the `b/` prefix.
_NEW_FILE_HEADER_RE = re.compile(r"^\+\+\+ b/(.+?)\s*$")


def changed_files(diff: str) -> list[str]:
    """Extract the sorted list of changed file paths from a unified diff.

    Pulled from `+++ b/<path>` headers (the post-change side). Deletions
    (`+++ /dev/null`) are excluded. Used to give the agent a clean focus list
    for context research, so it reads callers/imports of the actually-changed
    files instead of re-deriving them from the raw diff.
    """
    out = []
    seen = set()
    for line in (diff or "").splitlines():
        if not line.startswith("+++ b/"):
            continue
        m = _NEW_FILE_HEADER_RE.match(line)
        if not m:
            continue
        path = m.group(1).strip()
        if path and path not in seen:
            seen.add(path)
            out.append(path)
    return sorted(out)


_BRIEF_TEMPLATE = """\
# pragent review brief

- **repo:** {repo}
- **pr:** #{index}
- **head_sha:** `{sha}`

## Title
{title}

## Description
{description}

## Changed files (focus your context research here)
{changed_files}

For each changed file, read its callers, imports, sibling functions, and type
definitions so findings reflect how the change is actually used — don't flag a
hunk in isolation. Stop once a finding is grounded (1–3 related files per
finding; avoid runaway whole-repo walks).

## 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]"
    files = changed_files(diff)
    files_block = "\n".join(f"- `{p}`" for p in files) if files else "_(none)_"
    content = _BRIEF_TEMPLATE.format(
        repo=repo or "?",
        index=index or "?",
        sha=sha or "?",
        title=title or "(none)",
        description=description.strip() or "_(none)_",
        changed_files=files_block,
        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 <workdir>` 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
# ---------------------------------------------------------------------------

def _new_usage() -> dict:
    return {
        "input": 0, "output": 0, "reasoning": 0,
        "cache_read": 0, "cache_write": 0, "total": 0,
        "cost": 0.0, "steps": 0,
    }


def parse_opencode_events(stdout: str) -> tuple[str, dict | None]:
    """Parse `opencode run --format json` NDJSON stdout into (text, usage).

    - assistant text: concatenation of every `{"type":"text","part":{"text":…}}`
      event, in order → the agent's full message (prose + the findings ```json
      block). This is what `ai_review.parse_review_output` then extracts the
      findings JSON from.
    - usage: summed across every `{"type":"step_finish","part":{"tokens":…,
      "cost":…}}` event (one per model turn). Returns a dict with input/output/
      reasoning/cache_read/cache_write/total/cost/steps, or None if no
      step_finish was seen (e.g. empty/failed run).

    Tolerant: non-JSON lines, missing fields, or non-dict events are skipped
    (warm-up / log noise / tool events we don't care about). Never raises.
    """
    text_parts: list[str] = []
    usage = _new_usage()
    saw_step = False
    for line in (stdout or "").splitlines():
        line = line.strip()
        if not line or not line.startswith("{"):
            continue
        try:
            ev = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(ev, dict):
            continue
        etype = ev.get("type")
        part = ev.get("part") or {}
        if etype == "text" and isinstance(part, dict):
            t = part.get("text")
            if isinstance(t, str):
                text_parts.append(t)
        elif etype == "step_finish" and isinstance(part, dict):
            tok = part.get("tokens") or {}
            if isinstance(tok, dict):
                saw_step = True
                usage["steps"] += 1
                usage["input"] += int(tok.get("input") or 0)
                usage["output"] += int(tok.get("output") or 0)
                usage["reasoning"] += int(tok.get("reasoning") or 0)
                cache = tok.get("cache") or {}
                if isinstance(cache, dict):
                    usage["cache_read"] += int(cache.get("read") or 0)
                    usage["cache_write"] += int(cache.get("write") or 0)
                usage["total"] += int(tok.get("total") or 0)
                cost = part.get("cost")
                if isinstance(cost, (int, float)):
                    usage["cost"] += float(cost)
    return "".join(text_parts), (usage if saw_step else None)

_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) -> tuple[str, dict | None]:
    """Run the pragent agent headlessly in `workdir`. Returns `(text, usage)`.

    `text` is the reconstructed assistant message (prose + findings JSON) from
    the `--format json` event stream; `usage` is the summed token/cost usage
    across all model turns (or None if no step_finish event was seen).

    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 text.

    `--format json` makes opencode emit NDJSON events (text + step_finish with
    token usage) instead of formatted stdout — `parse_opencode_events` turns
    that into the assistant text + a usage dict.

    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",
        "--format", "json",
        "--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
        text, usage = parse_opencode_events(proc.stdout or "")
        if text.strip():
            return text, usage
        last_err = (
            f"opencode empty text (rc={proc.returncode}); "
            f"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,
) -> tuple[str, dict | None]:
    """End-to-end: checkout archive → brief → drop factory → opencode → (text, usage).

    Returns the reconstructed opencode assistant text (summary + findings JSON)
    and a usage dict (token/cost totals + `duration_s`), or `(text, None)` when
    no usage events were seen. 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"))
    t0 = time.monotonic()
    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)
        text, usage = run_opencode(workdir, model)
        if not text.strip():
            raise RuntimeError("opencode produced no output")
        if usage is not None:
            usage["duration_s"] = round(time.monotonic() - t0, 1)
        return text, usage
    finally:
        if not keep:
            shutil.rmtree(workdir, ignore_errors=True)