pragent/pilot/diff_compress.py

#!/usr/bin/env python3
r"""pragent pilot — diff compression + prior-review compaction.

Two pure helpers that shrink what lands in the model prompt without losing
signal:

  * ``compress_diff(diff, *, context=2)`` — re-renders a unified diff so each
    hunk keeps only ``context`` unchanged lines on either side of its +/- lines.
    The default 2 matches what most reviewers see on GitHub/Gitea, and is
    enough to anchor every ``+``/``-`` line and give the reviewer the enclosing
    statement. Wider context = more reading; narrower = less. Set
    ``context=0`` for +/- only, ``context=-1`` to disable entirely.

  * ``extract_finding_bullets(review_body)`` — pulls the lines of a prior
    review that look like a pragent finding (``- **[SEVERITY]** `path:line` — …``)
    and drops everything else. The model already has the diff — repeating the
    prose ("this PR adds eval() — risky") is just token burn. Bullet-only
    priors cut ~75% off prior-review bytes on a typical 4-finding review.

Stdlib only. No I/O. Tolerant of malformed input — never raises.
"""

from __future__ import annotations

import re

# Diff line types. Order matters: `+++`/ `---` headers and `@@` hunk headers
# are caught before the per-line prefix check.
_FILE_HEADER = re.compile(r"^(diff --git|Index:|---|\+\+\+|@@)")

# Captures `- <n>[,<m>]` AND `+ <n>[,<m>]` from `@@ -a,b +c,d @@`. We use the
# `+` side to reset the new-line counter; old-side is ignored.
_HUNK_RE = re.compile(r"^@@\s+-\d+(?:,\d+)?\s+\+(\d+)(?:,\d+)?\s+@@")

# Match a pragent summary-bullet line: `- **[SEVERITY]** \`path:line\` — …`.
# Severity is uppercased critical|high|medium|low per the findings schema.
# We also accept the lower-case form (`- [high]`) used by summary_bullets.
_FINDING_BULLET_RE = re.compile(
    r"^\s*-\s*\*?\*?\[(?P<sev>critical|high|medium|low|CRITICAL|HIGH|MEDIUM|LOW)\]"
    r"\*?\*?\s+(?P<rest>.+)$"
)


def compress_diff(diff: str, *, context: int = 2) -> tuple[str, int, int]:
    """Re-render `diff` keeping at most `context` unchanged lines around +/-.

    Args:
        diff: unified-diff text (what `gitea .../pulls/{n}.diff` returns).
        context: max unchanged lines to keep on each side of a hunk. Use 0
            for +/- only, -1 to disable compression (raw passthrough).

    Returns:
        `(text, original_chars, kept_chars)`. `original_chars` is the byte
        length of `diff` as given; `kept_chars` is the byte length of `text`.
        On parse failure the original is returned unchanged so the worst case
        is no improvement, never corruption.
    """
    if not diff:
        return diff or "", len(diff or ""), len(diff or "")
    if context < 0:
        return diff, len(diff), len(diff)

    orig = len(diff)
    lines = diff.splitlines()
    out: list[str] = []

    # State for the per-file walk.
    i = 0
    n = len(lines)
    while i < n:
        # Copy file headers verbatim until we hit the first `@@` hunk header.
        hunk_start = i
        while hunk_start < n and not lines[hunk_start].startswith("@@"):
            out.append(lines[hunk_start])
            hunk_start += 1
        i = hunk_start

        # Walk hunks, copying headers verbatim and trimming the inside.
        while i < n and lines[i].startswith("@@"):
            hunk_header = lines[i]
            i += 1

            # Collect the hunk body: every line until the next `@@` / file
            # header / EOF. Within the body, classify each line.
            body_start = i
            while i < n and not _FILE_HEADER.match(lines[i]):
                i += 1
            body = lines[body_start:i]

            # Render the body, collapsing long runs of context lines to a
            # `@@ … @@` marker so the reviewer still sees that there IS more
            # code there, just not in this window.
            rendered, _ = _render_hunk_body(body, context=context)
            if rendered:
                out.append(hunk_header)
                out.extend(rendered)

    text = "\n".join(out) + ("\n" if diff.endswith("\n") else "")
    if not text:
        # splitlines() dropped nothing-but-newlines; fall back to original.
        return diff, orig, orig
    return text, orig, len(text)


def _render_hunk_body(body: list[str], *, context: int) -> tuple[list[str], int]:
    """Trim `body` to `context` unchanged lines around the +/- lines.

    Body lines are classified:
      - `+` line → keep
      - `-` line → keep (paired with a `+` on the new side when both exist)
      - `` `` line (or empty context) → keep only within ``context`` of a +/- line
      - ``\ No newline at end of file`` → drop (no signal for the reviewer)

    Collapsed gaps of ≥ 5 lines get a single ``@@ … N context line(s) omitted … @@``
    marker so the reviewer knows code was elided. Smaller gaps (1–4 lines)
    stay silent — the marker would be longer than the elision.
    """
    if context == 0:
        # Keep only +/- lines; drop all context.
        out = [ln for ln in body if ln.startswith("+") or ln.startswith("-")]
        return out, 0

    # Find the index of every +/- line; a context line is kept if its
    # distance to the nearest +/- line is ≤ context.
    plus_minus_idx = [
        j for j, ln in enumerate(body)
        if ln.startswith("+") or ln.startswith("-")
    ]
    if not plus_minus_idx:
        # No +/- at all (rare — pure-context hunk): drop entirely.
        return [], 0

    keep = set()
    for k in plus_minus_idx:
        lo = max(0, k - context)
        hi = min(len(body) - 1, k + context)
        for j in range(lo, hi + 1):
            keep.add(j)

    out: list[str] = []
    last_kept = -2  # sentinel: a gap of ≥ 5 between consecutive kept lines triggers a marker
    for j, ln in enumerate(body):
        if ln.startswith("\\ No newline"):
            continue
        if j in keep:
            if j - last_kept > 5 and last_kept >= 0:
                out.append(f"@@ … {j - last_kept - 1} context line(s) omitted … @@")
            out.append(ln)
            last_kept = j

    return out, len(out)


def extract_finding_bullets(review_body: str) -> list[str]:
    """Pull the finding-bullet lines out of a prior review body.

    Returns the matching lines verbatim (with their original indentation +
    any continuation text), preserving the ``**[SEV]** `path:line` — problem
    …`` shape the model emitted. Lines that look like bullets but lack the
    severity tag are dropped — the reviewer synthesizes from the matched ones.
    """
    if not review_body:
        return []
    out = []
    for line in review_body.splitlines():
        m = _FINDING_BULLET_RE.match(line)
        if m:
            out.append(line.strip())
    return out