pr-reviewer/app/handbook.py

"""Ground reviews in the Sea Haven engineering-handbook.

Keeps an app-managed shallow clone of the (private) handbook repo, distills the
review-relevant pages into a compact conventions digest via the Fireworks model
once a day, caches it, and hands it to the reviewer to inject into every review.

Everything degrades gracefully: if the clone/pull or distillation fails, the last
good digest is kept (or none), reviews continue on the base prompt, and failed
refreshes back off so a handbook outage never turns into a per-cycle retry storm.
"""

from __future__ import annotations

import base64
import json
import logging
import os
import shutil
import subprocess
import threading
import time
from pathlib import Path

from .config import Config
from .github_client import GitHubError, _resolve_token
from .reviewer import fireworks_complete

_log = logging.getLogger("pr_reviewer.handbook")

# Review-relevant pages. Missing ones are warned about, not fatal (handbook may
# be restructured); zero found is an error (an empty digest is worse than none).
CURATED_PAGES = (
    "code-review-rubric.md",
    "code-review.md",
    "naming-conventions.md",
    "commit-messages.md",
    "git-workflow.md",
    "pull-requests.md",
    "secrets-and-config.md",
    "github-standards.md",
)

DISTILL_MAX_BYTES = 48_000  # cap handbook input to leave the model room to write
# Room for a reasoning model's preamble plus the tagged ~4 KB checklist.
DISTILL_MAX_TOKENS = 3_000
FAIL_RETRY_SECONDS = 3_600  # after a failed refresh, wait an hour before retrying

DISTILL_SYSTEM = (
    "You compile a concise code-review checklist from an engineering handbook. "
    "From the handbook pages, extract the conventions a reviewer should enforce "
    "on a pull request: naming, commit messages, PR structure, the code-review "
    "rubric and its severities, secrets/config placement, IaC/Lambda defaults, "
    "and the dependency-update (Dependabot) merge policy.\n\n"
    "Wrap the finished checklist between <CHECKLIST> and </CHECKLIST> tags and "
    "put nothing after the closing tag. Inside the tags use short ALL-CAPS or "
    "Title-Case headings with imperative bullet points, no markdown code fences. "
    "Keep it under 4000 characters. Omit anything not actionable while reading a "
    "diff. Any reasoning must stay outside the tags."
)


def _extract_checklist(text: str) -> str:
    """Pull the checklist out of the delimiters, tolerating a reasoning preamble
    that some models emit before the tagged answer."""
    start, end = "<CHECKLIST>", "</CHECKLIST>"
    i, j = text.find(start), text.rfind(end)
    if i != -1 and j != -1 and j > i:
        return text[i + len(start) : j].strip()
    return text.strip()  # no tags: fall back to the whole response


class HandbookError(RuntimeError):
    pass


def _git(args: list[str], *, token: str | None, timeout: int = 120):
    """Run git with a non-interactive environment. When a token is given it is
    passed via an in-memory Basic-auth http.extraHeader (GitHub git-over-HTTPS
    wants Basic, not Bearer), never written to remote config or the URL."""
    cmd = ["git"]
    if token:
        creds = base64.b64encode(f"x-access-token:{token}".encode()).decode()
        cmd += ["-c", f"http.extraHeader=Authorization: Basic {creds}"]
    cmd += args
    return subprocess.run(
        cmd,
        env={
            "GIT_TERMINAL_PROMPT": "0",
            "PATH": os.environ.get("PATH", "/usr/bin:/bin:/usr/local/bin"),
            "HOME": os.environ.get("HOME", ""),
        },
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=timeout,
    )


def ensure_checkout(cache_dir: Path, repo_url: str, token: str | None) -> str:
    """Clone the handbook (shallow) if absent, else fetch + hard-reset to
    origin/main. Returns the checked-out HEAD sha. Raises HandbookError on
    failure."""
    git_dir = cache_dir / ".git"
    if git_dir.exists():
        # Clear a lock left by a killed prior run (safe: single-writer tool).
        (git_dir / "index.lock").unlink(missing_ok=True)
        f = _git(
            ["-C", str(cache_dir), "fetch", "--depth", "1", "origin", "main"],
            token=token,
        )
        if f.returncode != 0:
            raise HandbookError(f"git fetch failed: {f.stderr.strip()[:300]}")
        r = _git(["-C", str(cache_dir), "reset", "--hard", "origin/main"], token=token)
        if r.returncode != 0:
            raise HandbookError(f"git reset failed: {r.stderr.strip()[:300]}")
    else:
        # Remove any stale non-git contents so clone can proceed.
        if cache_dir.exists() and any(cache_dir.iterdir()):
            shutil.rmtree(cache_dir)
        cache_dir.parent.mkdir(parents=True, exist_ok=True)
        c = _git(
            ["clone", "--depth", "1", "--branch", "main", repo_url, str(cache_dir)],
            token=token,
            timeout=180,
        )
        if c.returncode != 0:
            raise HandbookError(f"git clone failed: {c.stderr.strip()[:300]}")

    head = _git(["-C", str(cache_dir), "rev-parse", "HEAD"], token=token)
    if head.returncode != 0:
        raise HandbookError("git rev-parse HEAD failed")
    return head.stdout.strip()


def read_pages(cache_dir: Path) -> str:
    """Concatenate the curated review-relevant pages, capped in size."""
    parts: list[str] = []
    found = 0
    for name in CURATED_PAGES:
        p = cache_dir / name
        if not p.exists():
            _log.warning("handbook page missing, skipping: %s", name)
            continue
        found += 1
        parts.append(
            f"===== {name} =====\n{p.read_text(encoding='utf-8', errors='ignore')}"
        )
    if found == 0:
        raise HandbookError("no curated handbook pages found in checkout")
    text = "\n\n".join(parts)
    raw = text.encode("utf-8", "ignore")
    if len(raw) > DISTILL_MAX_BYTES:
        text = (
            raw[:DISTILL_MAX_BYTES].decode("utf-8", "ignore")
            + "\n\n[handbook truncated]"
        )
    return text


class HandbookProvider:
    """Owns the cached conventions digest and its daily refresh. Takes only
    ``cfg`` (no Reviewer reference) so there is no circular object graph; the
    reviewer receives ``current_digest`` as its guidance callable."""

    def __init__(self, cfg: Config) -> None:
        self.cfg = cfg
        self._lock = threading.Lock()
        self._digest: str | None = None
        self._head_sha: str | None = None
        self._distilled_at: float | None = None
        self._last_attempt: float = 0.0
        self._load_cache()

    def _load_cache(self) -> None:
        p = Path(self.cfg.HANDBOOK_DIGEST_PATH)
        if not p.exists():
            return
        try:
            d = json.loads(p.read_text(encoding="utf-8"))
            self._digest = d.get("digest")
            self._head_sha = d.get("head_sha")
            self._distilled_at = d.get("distilled_at")
        except (json.JSONDecodeError, OSError):
            _log.warning("could not read handbook digest cache at %s", p)

    def _write_cache(self) -> None:
        p = Path(self.cfg.HANDBOOK_DIGEST_PATH)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(
            json.dumps(
                {
                    "digest": self._digest,
                    "head_sha": self._head_sha,
                    "distilled_at": self._distilled_at,
                }
            ),
            encoding="utf-8",
        )

    def current_digest(self) -> str | None:
        return self._digest  # plain atomic read; safe without the lock

    def _is_stale(self) -> bool:
        if self._digest is None or self._distilled_at is None:
            return True
        return (
            time.time() - self._distilled_at
        ) > self.cfg.HANDBOOK_REFRESH_HOURS * 3600

    def refresh_if_stale(self) -> bool:
        """Pull + re-distill when the digest is missing or older than the refresh
        window. Returns True only when a new digest was produced. Never raises."""
        if not self.cfg.HANDBOOK_ENABLED:
            return False
        with self._lock:
            if not self._is_stale():
                return False
            if (time.time() - self._last_attempt) < FAIL_RETRY_SECONDS:
                return False  # backing off from a recent (failed) attempt
            self._last_attempt = time.time()

        # Network + LLM work happens OUTSIDE the lock so current_digest() never
        # blocks behind a slow distillation.
        try:
            try:
                token = _resolve_token(self.cfg)
            except GitHubError:
                token = None
            cache_dir = Path(self.cfg.HANDBOOK_CACHE_DIR)
            head = ensure_checkout(cache_dir, self.cfg.HANDBOOK_REPO_URL, token)
            pages = read_pages(cache_dir)
            raw = fireworks_complete(
                self.cfg,
                DISTILL_SYSTEM,
                pages,
                max_tokens=DISTILL_MAX_TOKENS,
                temperature=0,
            )
            digest = _extract_checklist(raw)
            if not digest:
                raise HandbookError("model returned an empty digest")
            # Atomic swaps (single assignments) — no lock needed for readers.
            self._digest = digest
            self._head_sha = head
            self._distilled_at = time.time()
            self._write_cache()
            _log.info(
                "handbook digest refreshed: %d chars, head %s", len(digest), head[:8]
            )
            return True
        except Exception as e:  # noqa: BLE001 - keep last good digest, never crash
            _log.warning("handbook refresh failed, keeping last digest: %s", e)
            return False

    def status(self) -> dict:
        return {
            "enabled": self.cfg.HANDBOOK_ENABLED,
            "head_sha": self._head_sha,
            "distilled_at": self._distilled_at,
            "digest_chars": len(self._digest) if self._digest else 0,
        }