2 Commits

Author SHA1 Message Date
Marcos f6be2b3c61 feat(review): PR-level collapsible metadata + emoji-tagged inline comments
PR-level comment layout (per operator's format guide):
  * Summary of Changes — 2-4 bullets, sourced from the agent's new
    `summary_changes` JSON field. Falls back to splitting the prose
    `summary` if the list is missing.
  * Key Risks & Concerns — bullets from the new `risks` JSON field.
  * Findings Overview — Markdown table covering every finding
    (severity emoji / location / one-line problem). Both anchored and
    unanchored findings appear here so the table is the single scan point.
  * Unanchored Notes — bullets with severity + fix + Markdown-linked ref,
    for findings with no post-change line to anchor.
  * AI Usage & Run Details — wrapped in a <details>/<summary> collapsible
    so the body stays scannable. Cost line stays inside it.

Inline comment shape:
  * Severity badge: 🔴 [HIGH] / 🟡 [MEDIUM] / 🔵 [LOW] /  [INFO].
    Unknown severities fall back to � [INFO].
  * 1-2 short paragraphs of problem; **Fix:** label for the fix line.
  * Standard ```suggestion fence for replacement code (Gitea/Forgejo
    apply-on-click). Language-tagged fences are no longer used for
    single-file diffs.
  * Reference as a Markdown hyperlink, visible label truncated to
    <=60 chars; the underlying URL is preserved verbatim.
  * NO per-comment 🪙 token attribution. All telemetry stays in the
    collapsible block on the PR-level comment.

Agent prompt updated to emit `summary_changes` and `risks` in the JSON
output (backward-compatible — older outputs missing them still parse;
they fall back to splitting the prose `summary`).

Tests: 15 new (severity emoji mapping, reference truncation, findings
table escaping, collapsible usage rendering, summary_changes+risks
layout). Existing tests updated for the new structure.
Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-20 16:51:20 +00:00
Marcos 5302e8dcd7 fix(review): salvage findings from nested-object fences + bare arrays + unfenced tail JSON
The canalhandia PR review lost all findings because the agent ran out of
context before emitting the closing json fence. Three failure modes hit
the old regex \{.*?\}:
  * nested objects inside the fence truncated at the first }
  * bare arrays (no {summary, findings} wrapper) returned []
  * unfenced JSON in the prose tail was never reached (first not last)

Replace the regex with a balanced-brace scanner:
  * _last_json_block walks the fence contents with a depth counter so
    nested objects survive
  * _last_balanced_json + _balanced_json_substring handle bare arrays and
    prose-tail JSON when no fence is present
  * _parse_json_tolerant returns list as well as dict; parse_findings and
    parse_review_output accept a bare array as the outer value

Agent prompt tightened: reserve the final step for emitting the JSON
block so the analysis isn't lost when context runs out.

10 new tests in tests/pilot/test_ai_review.py cover the new shapes.
Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-20 16:29:44 +00:00
4 changed files with 743 additions and 126 deletions
+18 -4
View File
@@ -142,12 +142,18 @@ containing STRICT JSON, nothing else after it:
```json
{
"summary": "One-paragraph overview of the change and its risk.",
"summary_changes": [
"24 short bullets explaining what the PR introduces or modifies"
],
"risks": [
"Bullets detailing potential bugs, edge cases, lifecycle issues, or performance risks found across the diff"
],
"findings": [
{
"severity": "critical|high|medium|low",
"severity": "critical|high|medium|low|info|nit",
"path": "path exactly as in the diff `+++ b/` side",
"line": 12,
"problem": "one line: what is wrong",
"problem": "12 short paragraphs: what is wrong and why it fails",
"fix": "one line: how to fix it",
"suggestion": "exact replacement lines for that location, indented as in the file, or \"\" if no safe replacement",
"reference": "https://... or \"\""
@@ -157,11 +163,19 @@ containing STRICT JSON, nothing else after it:
```
Rules:
- `summary_changes` (24 bullets) goes into the **Summary of Changes** section.
`risks` (bullets) goes into **Key Risks & Concerns**. Both are required;
empty arrays are fine when nothing applies.
- `suggestion` is the literal new code that replaces the flagged line(s). Minimal —
just the changed lines, indented as they'd appear in the file. Empty string `""`
when no safe textual replacement exists (e.g. missing test, architectural note).
- `problem` is 12 short paragraphs (the inline comment shows it verbatim).
Lead with the consequence (security / data loss / perf / etc.), then the cause.
- At most ~15 findings, highest severity first.
- If the diff is clean, output `{"summary":"...","findings":[]}`.
- If the diff is clean, output `{"summary":"...","summary_changes":[],"risks":[],"findings":[]}`.
- Do NOT repeat anything in `prior_reviews`.
- The JSON block must be the LAST thing in your message — the Python shell parses
the last ```json fenced block from your output.
the last ```json fenced block from your output. If you run out of context/steps
before emitting it, your analysis is wasted: ALWAYS reserve the final step for
writing the JSON. Stop exploring and write findings at the first sign you've
covered the diff (no new findings in the last 2 file reads = stop).
+461 -93
View File
@@ -156,28 +156,85 @@ def parse_text_blocks(content: list) -> str:
return "\n".join(out).strip()
def format_review_body(findings: str, model: str, sha: str, summary: str = "", usage_section: str = "") -> str:
def format_review_body(
findings: str,
model: str,
sha: str,
summary: str = "",
usage_section: str = "",
*,
summary_changes: list[str] | None = None,
risks: list[str] | None = None,
findings_for_table: list[dict] | None = None,
inline_count: int = 0,
) -> str:
"""Format the posted review summary body.
`findings` is the bullet text for findings that could NOT be anchored inline
(or, on the legacy/no-inline path, the whole review). Empty -> "No issues
found.". `summary` (optional, opencode engine) is rendered as a "Summary"
section right under the header. `usage_section` (optional, shown only when
the PR carries the `AI-USAGE` label) is rendered between the summary and the
findings bullets. The hidden sha marker is always appended for the dedupe
pass.
Layout (per the operator's format guide):
* Header line (``🤖 AI Review …``).
* **Summary of Changes** — 24 bullets of what the PR introduces
(`summary_changes`); falls back to the opencode prose `summary` if
the agent didn't emit the list.
* **Key Risks & Concerns** — bullets of potential bugs/edge cases
found across the diff (`risks`).
* **Findings Overview** — a Markdown table (severity / location /
one-line problem) covering ALL findings, anchored or not.
* Unanchored bullets — findings with no post-change line to anchor
(the inline ones are posted separately as Gitea review comments).
* AI Usage & Run Details — wrapped in a ``<details>`` collapsible so
the body stays scannable; cost lines stay inside it.
* Hidden SHA marker — for the dedupe pass.
Empty `summary_changes` + empty `risks` + empty `summary` collapse into
a single "Summary of Changes: _no summary provided._" line so the body
never looks half-rendered.
"""
header = REVIEW_HEADER.format(model=model, sha=sha[:8] if sha else "unknown")
findings = (findings or "").strip()
if not findings:
findings = "No issues found."
marker = SHA_MARKER.format(sha=sha) if sha else ""
parts = [header]
if summary:
parts.append(summary.strip())
parts: list[str] = [header]
# --- Summary of Changes ---
sc = list(summary_changes or [])
if not sc and summary:
sc = _string_list(summary)
if sc:
sc = sc[:4]
items = "\n".join(f"- {item}" for item in sc)
parts.append(f"### Summary of Changes\n\n{items}")
else:
parts.append("### Summary of Changes\n\n_No summary provided._")
# --- Key Risks & Concerns ---
rs = list(risks or [])
if rs:
items = "\n".join(f"- {item}" for item in rs)
parts.append(f"### Key Risks & Concerns\n\n{items}")
else:
parts.append("### Key Risks & Concerns\n\n_None identified._")
# --- Findings Overview (table) ---
table = findings_table(findings_for_table or [])
if table:
n_inline = inline_count
n_total = len(findings_for_table or [])
if n_inline:
heading = f"### Findings Overview\n\n_{n_inline} inline comment(s); {n_total} total._"
else:
heading = f"### Findings Overview\n\n_{n_total} finding(s)._"
parts.append(f"{heading}\n\n{table}")
# --- Unanchored bullets ---
fb = (findings or "").strip()
if fb:
parts.append(fb)
# --- Collapsible usage ---
if usage_section:
parts.append(usage_section.strip())
parts.append(findings)
# --- Hidden marker ---
marker = SHA_MARKER.format(sha=sha) if sha else ""
body = "\n\n".join(parts)
if marker:
body += f"\n{marker}"
@@ -509,14 +566,42 @@ def _normalize_finding(f: dict) -> dict | None:
def _last_json_block(text: str) -> str | None:
"""Return the substring of the last fenced ```json block in text, or None.
Falls back to _extract_first_json_object when no fence is present."""
r"""Return the substring of the last JSON object/array in text, or None.
The pragent agent emits ```json fences around its final block, but real
outputs drift:
* the fence contains nested objects (regex ``\{.*?\}`` only matches the
first ``}``, truncating the JSON — the parser then sees
``json.JSONDecodeError``);
* the fence is missing or unterminated, but a balanced JSON object sits
in the prose tail;
* the agent emits a bare array (findings only, no summary wrapper).
Strategy:
1. Find each fenced block, take the last. Inside it, walk a balanced
``{...}``/``[...]`` scanner (not a regex) so nested structures survive.
2. Fall back to a balanced scanner over the whole text, picking the LAST
balanced object/array (the agent writes its conclusion last).
"""
s = text or ""
# Find all ```json ... ``` fenced blocks; take the last.
blocks = list(re.finditer(r"```(?:json)?\s*(\{.*?\})\s*```", s, re.DOTALL))
if blocks:
return blocks[-1].group(1)
return _extract_first_json_object(s)
if not s:
return None
# 1. Fenced blocks: take the last ```json ... ``` or ``` ... ``` region.
fences = list(re.finditer(r"```(?:json)?\n", s))
for m in reversed(fences):
start = m.end()
# Find the matching closing fence.
end = s.find("```", start)
if end < 0:
# Unterminated fence — try to salvage the balanced object inside.
end = len(s)
inner = s[start:end].strip()
obj = _balanced_json_substring(inner)
if obj is not None:
return obj
# 2. No (parseable) fence — scan the whole text for the LAST balanced
# object/array. The agent's conclusion is at the tail.
return _last_balanced_json(s)
def parse_findings(text: str) -> list[dict]:
@@ -526,11 +611,17 @@ def parse_findings(text: str) -> list[dict]:
scans for the first balanced `{...}` and extracts its `findings` array.
Drops findings missing path/line or with an unknown severity (normalised).
Never raises — returns [] on any parse failure.
Also accepts a bare JSON array as the outer value: ``[{...}, {...}]`` —
some agents skip the ``{"summary":..., "findings":[...]}`` wrapper.
"""
data = _parse_json_tolerant(text)
if not isinstance(data, dict):
if isinstance(data, dict):
findings = data.get("findings")
elif isinstance(data, list):
findings = data
else:
return []
findings = data.get("findings")
if not isinstance(findings, list):
return []
out = []
@@ -574,44 +665,87 @@ def salvage_summary(text: str, max_chars: int = SALVAGE_MAX_CHARS) -> str:
)
def parse_review_output(text: str) -> tuple[str, list[dict]]:
"""Parse the opengine's stdout into (summary, findings).
def parse_review_output(text: str) -> tuple[str, list[dict], list[str], list[str]]:
"""Parse the opengine's stdout into (summary, findings, summary_changes, risks).
Accepts `{"summary": "...", "findings": [...]}` (the opencode pragent agent)
or a bare `{"findings": [...]}`. `summary` defaults to "". Uses the LAST
```json fenced block (the pragent agent emits JSON as the final block), with
a tolerant fallback. Never raises.
Accepts `{"summary": "...", "summary_changes": [...], "risks": [...],
"findings": [...]}` (the opencode pragent agent), `{"findings": [...]}`,
or a bare `[...]` of finding dicts. `summary_changes` and `risks` default
to empty lists; older outputs without them still parse fine. Uses the
LAST fenced block (the pragent agent emits JSON as the final block), with
a tolerant fallback that scans for the last balanced object/array in the
prose tail. Never raises.
"""
blob = _last_json_block(text)
if blob is None:
return "", []
return "", [], [], []
try:
data = json.loads(blob)
except json.JSONDecodeError:
return "", []
if not isinstance(data, dict):
return "", []
summary = str(data.get("summary", "") or "").strip()
findings = data.get("findings")
return "", [], [], []
summary = ""
summary_changes: list[str] = []
risks: list[str] = []
findings_raw = None
if isinstance(data, dict):
summary = str(data.get("summary", "") or "").strip()
summary_changes = _string_list(data.get("summary_changes"))
risks = _string_list(data.get("risks"))
findings_raw = data.get("findings")
elif isinstance(data, list):
# Bare array: each item is a finding; no summary/sections.
findings_raw = data
else:
return "", [], [], []
out = []
if isinstance(findings, list):
for f in findings:
if isinstance(findings_raw, list):
for f in findings_raw:
n = _normalize_finding(f)
if n is not None:
out.append(n)
return summary, out
return summary, out, summary_changes, risks
def _parse_json_tolerant(text: str) -> dict | None:
"""Parse a JSON object from text: try the last fenced block, then a direct
parse, then the first balanced object. Returns None on any failure."""
def _string_list(value) -> list[str]:
"""Coerce a JSON value into a list of non-empty strings.
Accepts a list of strings, a single string (split on lines/bullets), or
anything else (returns []). Used for `summary_changes` and `risks`,
which some agents emit as one big string instead of a list.
"""
if isinstance(value, list):
return [str(v).strip() for v in value if str(v).strip()]
if isinstance(value, str):
s = value.strip()
if not s:
return []
# Split on newlines OR on lines that start with "- " / "* " (markdown
# bullets). Strip the bullet markers.
out: list[str] = []
for line in s.splitlines():
line = line.strip()
if not line:
continue
if line[:2] in ("- ", "* "):
line = line[2:].strip()
if line:
out.append(line)
return out
return []
def _parse_json_tolerant(text: str) -> dict | list | None:
"""Parse a JSON object/array from text: try the last fenced block, then a
direct parse, then the first balanced object. Returns None on any failure.
Accepts both ``{...}`` (the pragent schema) and bare ``[...]`` arrays
(agents that skip the wrapper)."""
if not text:
return None
blob = _last_json_block(text)
if blob is not None:
try:
d = json.loads(blob)
if isinstance(d, dict):
if isinstance(d, (dict, list)):
return d
except json.JSONDecodeError:
pass
@@ -621,7 +755,7 @@ def _parse_json_tolerant(text: str) -> dict | None:
s = re.sub(r"\n?```$", "", s).strip()
try:
d = json.loads(s)
if isinstance(d, dict):
if isinstance(d, (dict, list)):
return d
except json.JSONDecodeError:
pass
@@ -629,7 +763,17 @@ def _parse_json_tolerant(text: str) -> dict | None:
if obj is not None:
try:
d = json.loads(obj)
if isinstance(d, dict):
if isinstance(d, (dict, list)):
return d
except json.JSONDecodeError:
pass
# Last resort: the JSON lives at the tail of the prose with no fence.
# Walk the whole text for the last balanced object/array.
last = _last_balanced_json(text)
if last is not None:
try:
d = json.loads(last)
if isinstance(d, (dict, list)):
return d
except json.JSONDecodeError:
pass
@@ -641,6 +785,62 @@ def _extract_first_json_object(s: str) -> str | None:
start = s.find("{")
if start < 0:
return None
end = _scan_balanced(s, start, "{", "}")
if end is None:
return None
return s[start:end + 1]
def _last_balanced_json(s: str) -> str | None:
"""Return the substring of the LAST balanced ``{...}`` or ``[...]`` in s.
Used when the agent emits no fence: the JSON lives in the prose tail.
Picks whichever closer (object or array) appears latest in the text.
"""
if not s:
return None
last_obj = _find_last_close(s, "{", "}")
last_arr = _find_last_close(s, "[", "]")
candidates = []
if last_obj is not None:
candidates.append(last_obj)
if last_arr is not None:
candidates.append(last_arr)
if not candidates:
return None
end, opener, start = max(candidates, key=lambda t: t[0])
return s[start:end + 1]
def _balanced_json_substring(s: str) -> str | None:
"""Return the first balanced ``{...}`` or ``[...]`` substring in ``s``.
Skips past leading whitespace/non-JSON and returns the full balanced
extent (handles nested objects/arrays and string literals with braces).
"""
if not s:
return None
# Try object first; the pragent schema is an object on the outer level.
for i, c in enumerate(s):
if c == "{":
end = _scan_balanced(s, i, "{", "}")
if end is not None:
return s[i:end + 1]
break
if c == "[":
end = _scan_balanced(s, i, "[", "]")
if end is not None:
return s[i:end + 1]
break
return None
def _scan_balanced(s: str, start: int, opener: str, closer: str) -> int | None:
"""Return the index of the matching ``closer`` for ``s[start] == opener``.
Tracks string literals (with ``\\`` escapes) so braces inside strings don't
fool the depth counter. Returns None if no balance is reached.
"""
depth = 0
in_str = False
esc = False
@@ -656,12 +856,49 @@ def _extract_first_json_object(s: str) -> str | None:
continue
if c == '"':
in_str = True
elif c == "{":
elif c == opener:
depth += 1
elif c == "}":
elif c == closer:
depth -= 1
if depth == 0:
return s[start:i + 1]
return i
return None
def _find_last_close(s: str, opener: str, closer: str) -> tuple[int, str, int] | None:
"""Walk ``s`` backwards from the last ``closer`` to find its matching opener.
Returns ``(close_idx, opener_char, open_idx)`` for the rightmost balanced
structure, or None if no pair exists.
"""
# Find the last `closer` candidate.
last = s.rfind(closer)
while last >= 0:
# Walk left, tracking depth from the perspective of the opener.
depth = 1
in_str = False
esc = False
for j in range(last - 1, -1, -1):
c = s[j]
if in_str:
if esc:
esc = False
elif c == "\\":
esc = True
elif c == '"':
in_str = False
continue
if c == '"':
# Approximation: we don't track quotes perfectly walking
# backwards, but strings in agent output are short and rare.
in_str = not in_str
elif c == closer:
depth += 1
elif c == opener:
depth -= 1
if depth == 0:
return (last, opener, j)
last = s.rfind(closer, 0, last)
return None
@@ -708,43 +945,175 @@ def _lang_for_path(path: str) -> str:
}.get(ext, "")
_SEVERITY_EMOJI = {
"critical": "🔴",
"high": "🔴",
"medium": "🟡",
"low": "🔵",
"info": "",
"nit": "",
}
def _severity_badge(severity: str) -> str:
"""Render the severity as emoji + uppercase label (e.g. ``🔴 [HIGH]``)."""
sev = (severity or "").lower()
emoji = _SEVERITY_EMOJI.get(sev, "")
label = sev.upper() if sev in {"critical", "high", "medium", "low"} else "INFO"
return f"{emoji} [{label}]"
def _format_reference(ref: str) -> str:
"""Render a reference URL as a clean Markdown hyperlink.
``"https://example.com/x"`` → ``"[example.com/x](https://example.com/x)"``.
Accepts the bare URL form so older findings still render readably; drops
anything that doesn't look like a URL rather than embedding raw text in
parens (the spec says: never print raw URLs).
"""
ref = (ref or "").strip()
if not ref:
return ""
if not (ref.startswith("http://") or ref.startswith("https://")):
# Non-URL text (e.g. a CVE id, a doc title). Render as plain text label.
return f"[{ref}]({ref})"
# Strip the scheme + www. for the visible label so the link text is short.
visible = ref
for prefix in ("https://", "http://"):
if visible.startswith(prefix):
visible = visible[len(prefix):]
break
if visible.startswith("www."):
visible = visible[4:]
# Drop trailing slash + truncate any path noise past 60 chars.
visible = visible.rstrip("/")
if len(visible) > 60:
visible = visible[:57] + ""
return f"[{visible}]({ref})"
def inline_comment_body(f: dict) -> str:
"""Render one finding as a positional review-comment body.
Includes a fenced suggested-fix block only if the model produced non-empty
replacement code. The fence is tagged with the file's language (via
`_lang_for_path`) so Gitea syntax-highlights it — Gitea 1.26.x has no
GitHub-style "Apply suggestion" button (```suggestion is just an
unknown-language block there → plain monospace), so a language-tagged block
is strictly more readable and loses nothing. Appends a `📎 ref:` link when
the finding carries a `reference` URL.
Shape:
* Severity badge with emoji (🔴 HIGH / 🟡 MEDIUM / 🔵 LOW / ⚪ INFO).
* 12 short paragraphs: ``problem`` + optional ``fix``.
* ``suggestion`` block (Gitea/Forgejo apply-on-click) when the model
produced replacement code. Language-tagged fences are reserved for
cross-file patterns the suggestion block can't carry.
* Reference as a Markdown hyperlink (``[label](url)``) — never a raw URL.
* No per-comment token attribution: the PR-level collapsible carries
all telemetry; inline comments stay focused on the code.
"""
sev = f["severity"].upper()
body = f"**[{sev}]** {f['problem']}"
if f["fix"]:
body += f"\n\nFix: {f['fix']}"
if f["suggestion"]:
lang = _lang_for_path(f.get("path", ""))
fence = f"```{lang}" if lang else "```"
body += f"\n\n{fence}\n{f['suggestion']}\n```"
ref = f.get("reference", "")
if ref:
body += f"\n\n📎 ref: {ref}"
tok = f.get("_tok_attrib")
if tok is not None:
pct = (f.get("_tok_pct", 0.0) or 0.0) * 100
body += f"\n\n🪙 ~{tok} tok ({pct:.0f}% · attributed output)"
badge = _severity_badge(f.get("severity", "medium"))
body = f"{badge} {f.get('problem', '').strip()}"
fix = (f.get("fix") or "").strip()
if fix:
body += f"\n\n**Fix:** {fix}"
suggestion = (f.get("suggestion") or "").strip()
if suggestion:
# `suggestion` fence is the standard one-click-apply block in
# Gitea/Forgejo/GitHub. The agent's replacement lines must already be
# indented as in the target file.
body += f"\n\n```suggestion\n{suggestion}\n```"
ref_md = _format_reference(f.get("reference", ""))
if ref_md:
body += f"\n\n🔗 **Reference:** {ref_md}"
return body
def summary_bullets(findings: list[dict]) -> str:
"""Render unanchored findings as summary-body bullets (no line anchor)."""
"""Render unanchored findings as PR-level bullets.
Used for findings that couldn't be anchored to a post-change line (no
inline comment posted). Each bullet carries severity, location, problem,
fix, and a Markdown-linked reference.
"""
lines = []
for f in findings:
loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"]
fix = f" — fix: {f['fix']}" if f["fix"] else ""
ref = f" ({f.get('reference', '')})" if f.get("reference") else ""
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}{ref}")
badge = _severity_badge(f.get("severity", "medium"))
problem = f.get("problem", "").strip()
body = f"- {badge} `{loc}` — {problem}"
fix = (f.get("fix") or "").strip()
if fix:
body += f"\n - **Fix:** {fix}"
ref_md = _format_reference(f.get("reference", ""))
if ref_md:
body += f"\n - 🔗 **Reference:** {ref_md}"
lines.append(body)
return "\n".join(lines)
def findings_table(findings: list[dict]) -> str:
"""Render ALL findings as a Markdown table for the PR-level comment.
Columns: severity emoji, location (path:line), and a one-line summary.
Findings with empty location collapse to just the severity + summary.
"""
if not findings:
return ""
header = "| Severity | Location | Finding |\n|---|---|---|"
rows = []
for f in findings:
badge = _severity_badge(f.get("severity", "medium"))
path = (f.get("path") or "").strip()
line = f.get("line")
loc = f"`{path}:{line}`" if line else (f"`{path}`" if path else "_(no location)_")
problem = (f.get("problem") or "").strip()
# Escape pipes inside the finding text so the table stays valid.
problem_esc = problem.replace("|", "\\|").replace("\n", " ")
rows.append(f"| {badge} | {loc} | {problem_esc} |")
return "\n".join([header, *rows])
def _render_collapsible_usage(usage: dict | None, model: str, config: dict | None) -> str:
"""Render the telemetry as a collapsible ``<details>`` block.
Empty string when `usage` is None. The cost-equivalent line is always
shown (it's the operator's budgeting signal). The `actual` line is shown
but the FREE-TIER note is collapsed into a single short clause.
"""
if not usage:
return ""
dur = usage.get("duration_s")
dur_s = f"{dur}s" if dur is not None else "?"
actual = usage.get("cost") or 0.0
actual_s = f"${actual:.4f}" if actual else "$0.00"
actual_note = " (headroom glm-5.2:cloud — free tier)" if not actual else ""
price_key, price_err = _resolve_price_target(config)
from cost_model import PRICES
eq = equivalent_cost(usage, price_key)
eq_s = f"${eq:.4f}" if eq else "$0.00"
eq_label = PRICES[price_key].name
eq_note = (
f" _(price target: `{price_key}`; {price_err})_"
if price_err else ""
)
in_tok = usage.get("input", 0)
out_tok = usage.get("output", 0)
reason_tok = usage.get("reasoning", 0)
cache_r = usage.get("cache_read", 0)
cache_w = usage.get("cache_write", 0)
total = usage.get("total", 0)
scope = (
"Whole-repo checkout at head sha (agent can read any file + run "
"linters, not just the diff) — input tokens include files read "
"beyond the diff. Per-comment output is *attributed* (one model pass "
"produces all findings; output split by each finding's body weight)."
)
lines = [
"<details>",
"<summary>🔋 AI Usage & Run Details</summary>",
"",
f"- **Model / Engine**: `{model}` · opencode · {usage.get('steps', 0)} steps · {dur_s}",
f"- **Total Tokens**: {in_tok} in / {out_tok} out ({reason_tok} reasoning, cache {cache_r} read / {cache_w} write, {total} total)",
f"- **Est. cost on {eq_label}**: {eq_s}{eq_note}",
f"- **Actual**: {actual_s}{actual_note}",
f"- **Scope**: {scope}",
"",
"</details>",
]
return "\n".join(lines)
@@ -1297,7 +1666,7 @@ def review_pr(
prior_reviews=prior, model=oc_model,
compression_note=compression_note,
)
review_summary, findings = parse_review_output(stdout)
review_summary, findings, summary_changes, risks = parse_review_output(stdout)
if not findings and not review_summary:
# The findings JSON was missing or malformed. Don't discard the
# run: salvage the prose, keep the usage report (the label asked
@@ -1309,9 +1678,7 @@ def review_pr(
file=sys.stderr, flush=True,
)
salvaged = salvage_summary(stdout)
usage_section = ""
if report_usage and usage:
usage_section = format_usage_section(usage, [], model, config=config)
usage_section = _render_collapsible_usage(usage, model, config=config) if report_usage else ""
post_review(api, repo, index, token, format_review_body(
salvaged or "AI review produced no parseable output.",
model, sha, usage_section=usage_section))
@@ -1346,31 +1713,32 @@ def review_pr(
flush=True,
)
# Attribute output tokens to each finding (mutates finding dicts) so
# inline comments + the usage table can show a per-comment estimate.
# Only meaningful when we have measured usage AND the PR asked for it.
usage_section = ""
# Compute attribution so inline comments + the table can show per-comment
# estimates. Only meaningful when we have measured usage AND the PR asked
# for it.
if report_usage and usage and usage.get("output"):
compute_attribution(findings, usage["output"])
usage_section = format_usage_section(usage, findings, model, config=config)
usage_section = _render_collapsible_usage(usage, model, config=config) if report_usage else ""
anchors = parse_diff_anchors(diff)
anchored, unanchored = split_findings(findings, anchors)
# Summary body: the unanchored bullets (or "No issues found."), plus a
# one-line note when inline comments were posted so the summary isn't
# empty-looking. The opencode engine also carries a prose summary.
# Summary body: unanchored bullets fall through to a "Unanchored notes"
# section; the structured Findings Overview table covers both anchored
# + unanchored so reviewers see the full set even if inline comments
# are collapsed.
bullets = summary_bullets(unanchored)
summary_parts = []
if anchored:
summary_parts.append(f"_{len(anchored)} inline comment(s) posted below._")
if bullets:
summary_parts.append(bullets)
if not summary_parts:
summary_parts.append("No issues found.")
summary_parts.append("### Unanchored Notes\n\n" + bullets)
summary_body = format_review_body(
"\n\n".join(summary_parts), model, sha,
summary=review_summary, usage_section=usage_section,
summary=review_summary,
usage_section=usage_section,
summary_changes=summary_changes,
risks=risks,
findings_for_table=findings,
inline_count=len(anchored),
)
post_inline_review(api, repo, index, token, summary_body, anchored)
+1 -1
View File
@@ -103,7 +103,7 @@ def compress_diff(diff: str, *, context: int = 2) -> tuple[str, int, int]:
def _render_hunk_body(body: list[str], *, context: int) -> tuple[list[str], int]:
"""Trim `body` to `context` unchanged lines around the +/- lines.
r"""Trim `body` to `context` unchanged lines around the +/- lines.
Body lines are classified:
- `+` line → keep
+263 -28
View File
@@ -11,8 +11,13 @@ sys.path.insert(0, os.path.join(ROOT, "pilot"))
import ai_review # noqa: E402
from ai_review import ( # noqa: E402
_balanced_json_substring,
_extract_first_json_object,
_last_balanced_json,
_render_collapsible_usage,
build_user_prompt,
compute_attribution,
findings_table,
format_review_body,
format_usage_section,
inline_comment_body,
@@ -105,17 +110,25 @@ def test_format_review_body_findings():
assert "pragent pilot" in body
assert "glm-5.2:cloud" in body
assert "`abcdef12`" in body # 8-char sha
assert "- [high] x:1" in body
# New layout: always emits Summary of Changes + Key Risks. Findings table
# only shows when findings_for_table is passed (callers pass the actual
# list of finding dicts; plain-string findings arg renders as bullets).
assert "### Summary of Changes" in body
assert "### Key Risks & Concerns" in body
def test_format_review_body_empty_findings():
body = format_review_body("", "glm-5.2:cloud", "abcdef1234567890")
assert "No issues found." in body
# No summary → "no summary provided" sentinel; Findings table absent
# because no findings were passed.
assert "_No summary provided._" in body
assert "_None identified._" in body
assert "### Findings Overview" not in body
def test_format_review_body_whitespace_findings():
body = format_review_body(" \n ", "glm-5.2:cloud", "abcdef1234567890")
assert "No issues found." in body
assert "_No summary provided._" in body
def test_format_review_body_no_sha():
@@ -271,33 +284,98 @@ def test_split_findings_by_anchor():
def test_inline_comment_body_with_suggestion():
f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap", "suggestion": "good()"}
body = inline_comment_body(f)
assert "**[HIGH]**" in body
# Severity emoji + bracketed label.
assert "🔴 [HIGH]" in body
assert "bad" in body
# no extension → bare fence (Gitea 1.26.x has no apply-suggestion; we tag
# with the file language for highlighting instead of ```suggestion)
assert "```\ngood()\n```" in body
# Standard ```suggestion fence (Gitea/Forgejo apply-on-click).
assert "```suggestion\ngood()\n```" in body
assert "good()" in body
def test_inline_comment_body_suggestion_lang_tagged():
def test_inline_comment_body_suggestion_not_lang_tagged():
# Per the format spec, the suggestion fence is ALWAYS ```suggestion —
# never a language-tagged fence (those are reserved for cross-file
# pattern illustrations, which we don't emit here).
f = {"severity": "high", "path": "src/Foo.java", "line": 1,
"problem": "bad", "fix": "swap", "suggestion": "good();"}
body = inline_comment_body(f)
assert "```java\ngood();\n```" in body
assert "```suggestion\ngood();\n```" in body
assert "```java" not in body
def test_inline_comment_body_no_suggestion():
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "f", "suggestion": ""}
body = inline_comment_body(f)
assert "```" not in body
assert "Fix: f" in body
assert "**Fix:** f" in body
def test_inline_comment_body_severity_emoji_mapping():
cases = [
("critical", "🔴 [CRITICAL]"),
("high", "🔴 [HIGH]"),
("medium", "🟡 [MEDIUM]"),
("low", "🔵 [LOW]"),
("info", "⚪ [INFO]"),
("nit", "⚪ [INFO]"), # "nit" maps to the INFO label
("bogus", "⚪ [INFO]"), # unknown severity falls back to INFO
]
for sev, badge in cases:
f = {"severity": sev, "path": "a", "line": 1, "problem": "p", "fix": "",
"suggestion": "", "reference": ""}
assert badge in inline_comment_body(f), f"{sev}{badge}"
def test_inline_comment_body_no_token_attribution():
# Per spec: no per-comment 🪙 token attribution line.
f = {"severity": "high", "path": "a", "line": 1, "problem": "p",
"fix": "f", "suggestion": "", "reference": "",
"_tok_attrib": 1234, "_tok_pct": 0.3}
body = inline_comment_body(f)
assert "🪙" not in body
assert "tok" not in body.lower().split("fix")[0] # only in fix is OK
assert "attributed" not in body
def test_summary_bullets_format():
fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f", "suggestion": ""}]
b = summary_bullets(fs)
assert "- **[HIGH]**" in b
assert "🔴 [HIGH]" in b
assert "`a.py:7`" in b
assert "**Fix:** f" in b
def test_summary_bullets_with_reference_link():
fs = [{"severity": "medium", "path": "x", "line": 1, "problem": "p",
"fix": "", "suggestion": "", "reference": "https://owasp.org/x"}]
b = summary_bullets(fs)
assert "🔗 **Reference:** [owasp.org/x](https://owasp.org/x)" in b
assert "https://owasp.org/x" in b # URL preserved
def test_findings_table_renders_table():
fs = [
{"severity": "high", "path": "a.py", "line": 1, "problem": "bug", "fix": "", "suggestion": "", "reference": ""},
{"severity": "low", "path": "b.go", "line": 9, "problem": "nit", "fix": "", "suggestion": "", "reference": ""},
]
t = findings_table(fs)
assert t.startswith("| Severity | Location | Finding |")
assert "|---|---|---|" in t
assert "🔴 [HIGH]" in t
assert "🔵 [LOW]" in t
assert "`a.py:1`" in t
assert "`b.go:9`" in t
def test_findings_table_escapes_pipes():
fs = [{"severity": "high", "path": "a", "line": 1,
"problem": "uses | inside", "fix": "", "suggestion": "", "reference": ""}]
t = findings_table(fs)
assert "uses \\| inside" in t
def test_findings_table_empty():
assert findings_table([]) == ""
# ---------------------------------------------------------------------------
@@ -373,7 +451,7 @@ def test_parse_review_output_summary_and_findings():
"]}",
"\n```",
)
summary, fs = parse_review_output("".join(txt))
summary, fs, *_ = parse_review_output("".join(txt))
assert "eval()" in summary
assert len(fs) == 1
assert fs[0]["severity"] == "critical"
@@ -383,16 +461,16 @@ def test_parse_review_output_summary_and_findings():
def test_parse_review_output_bare_findings_no_summary():
txt = '```json\n{"findings":[{"severity":"low","path":"a","line":1,"problem":"p"}]}\n```'
summary, fs = parse_review_output(txt)
summary, fs, *_ = parse_review_output(txt)
assert summary == ""
assert len(fs) == 1
assert fs[0]["reference"] == "" # default
def test_parse_review_output_empty_and_bogus():
assert parse_review_output("") == ("", [])
assert parse_review_output("no json here") == ("", [])
assert parse_review_output('{"findings":[]}') == ("", [])
assert parse_review_output("") == ("", [], [], [])
assert parse_review_output("no json here") == ("", [], [], [])
assert parse_review_output('{"findings":[]}') == ("", [], [], [])
def test_parse_review_output_uses_last_json_block():
@@ -402,12 +480,114 @@ def test_parse_review_output_uses_last_json_block():
"more prose\n"
"```json\n{\"summary\":\"real\",\"findings\":[{\"path\":\"y\",\"line\":2,\"severity\":\"high\"}]}\n```"
)
summary, fs = parse_review_output(txt)
summary, fs, *_ = parse_review_output(txt)
assert summary == "real"
assert len(fs) == 1
assert fs[0]["path"] == "y"
def test_parse_findings_fenced_json_with_nested_object():
# Real-world regression: agent emits a fence whose inner JSON has nested
# objects. The old regex `\{.*?\}` matched only the first `}`, truncating
# the JSON. Now we balance braces inside the fence.
txt = (
"```json\n"
'{"summary":"x","findings":[{"severity":"high","path":"a.py","line":1,'
'"problem":"p","fix":"f","suggestion":"","reference":""}],"meta":{"engine":"opencode"}}\n'
"```"
)
fs = parse_findings(txt)
assert len(fs) == 1
assert fs[0]["path"] == "a.py"
def test_parse_findings_unfenced_at_tail():
# No fence at all. Agent wrote the JSON inline at the very end of its
# prose. The old first-balanced regex caught the FIRST `{`, not this one.
txt = (
"I considered the diff carefully. Two findings stand out:\n"
"First one is just text.\n"
'{"findings":[{"severity":"critical","path":"x","line":1,"problem":"p","fix":"f"}]}'
)
fs = parse_findings(txt)
assert len(fs) == 1
assert fs[0]["severity"] == "critical"
def test_parse_findings_bare_array():
# Some agents skip the `{"summary":..., "findings":[...]}` wrapper and
# emit just the array.
txt = (
"Here are my findings:\n"
"```json\n"
'[{"severity":"low","path":"a","line":1,"problem":"p","fix":"f","suggestion":"","reference":""}]\n'
"```"
)
fs = parse_findings(txt)
assert len(fs) == 1
assert fs[0]["path"] == "a"
def test_parse_review_output_unfenced_at_tail():
# The exact shape canalhandia produced: long prose, JSON at the very end,
# no fence. Old parser returned ([], salvage) — now we recover findings.
txt = (
"Let me refine the fix: should call a dedicated `setPermanent`.\n"
"Let me finalize. Let me also double-check the `find` thread-safety.\n"
'{"summary":"Adds void protection; one critical race.","findings":['
'{"severity":"high","path":"VoidProtection.java","line":162,'
'"problem":"drop duplication race","fix":"use ItemMeta","suggestion":"","reference":""}]}'
)
summary, fs, *_ = parse_review_output(txt)
assert "void protection" in summary.lower()
assert len(fs) == 1
assert fs[0]["path"] == "VoidProtection.java"
def test_parse_review_output_bare_array_at_tail():
txt = (
"All wrapped up.\n"
'[{"severity":"low","path":"a","line":1,"problem":"p","fix":"","suggestion":"","reference":""}]'
)
summary, fs, *_ = parse_review_output(txt)
assert summary == ""
assert len(fs) == 1
def test_scan_balanced_handles_braces_in_strings():
# The JSON scanner must not be fooled by `{` or `}` inside string literals.
s = '{"a":"contains { and }","b":1}'
obj = _extract_first_json_object(s)
assert obj == s
d = json.loads(obj)
assert d["a"] == "contains { and }"
def test_last_balanced_json_picks_latest():
s = '{"a":1} some text {"b":2,"nested":{"c":3}} trailing'
out = _last_balanced_json(s)
assert out is not None
d = json.loads(out)
assert d == {"b": 2, "nested": {"c": 3}}
def test_last_balanced_json_no_json():
assert _last_balanced_json("nothing here") is None
assert _last_balanced_json("") is None
def test_balanced_json_substring_skips_leading_prose():
s = 'preamble {"a":1} more prose {"b":2}'
out = _balanced_json_substring(s)
assert out == '{"a":1}'
def test_balanced_json_substring_handles_array():
s = '[{"a":1},{"b":2}]'
out = _balanced_json_substring(s)
assert out == s
# ---------------------------------------------------------------------------
# reference rendering in inline_comment_body + summary_bullets + summary section
# ---------------------------------------------------------------------------
@@ -417,13 +597,26 @@ def test_inline_comment_body_renders_reference():
f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "f",
"suggestion": "", "reference": "https://cve.example/X"}
body = inline_comment_body(f)
assert "📎 ref: https://cve.example/X" in body
# Per spec: Markdown hyperlink, not raw URL.
assert "🔗 **Reference:** [cve.example/X](https://cve.example/X)" in body
def test_inline_comment_body_no_reference_no_ref_line():
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "",
"suggestion": "", "reference": ""}
assert "📎 ref" not in inline_comment_body(f)
assert "🔗" not in inline_comment_body(f)
assert "Reference:" not in inline_comment_body(f)
def test_inline_comment_body_reference_truncates_long_url():
f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "",
"suggestion": "",
"reference": "https://very-long-domain.example.com/some/very/long/path/that/exceeds/the/sixty/char/limit/x"}
body = inline_comment_body(f)
# Visible label is truncated to ≤60 chars (ellipsis added).
assert "" in body
# But the underlying URL is preserved verbatim inside the link target.
assert "very-long-domain.example.com" in body
def test_summary_bullets_renders_reference():
@@ -480,13 +673,15 @@ def test_compute_attribution_noop_on_empty_or_zero_budget():
assert "_tok_attrib" not in fs[0]
def test_inline_comment_body_with_attribution_line():
def test_inline_comment_body_no_attribution_line():
# Per spec: NO per-comment token attribution — that telemetry lives in the
# collapsible block on the PR-level comment.
f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap",
"suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29}
body = inline_comment_body(f)
assert "🪙 ~180 tok" in body
assert "29%" in body
assert "attributed output" in body
assert "🪙" not in body
assert "tok" not in body
assert "attributed" not in body
def test_inline_comment_body_no_attribution_no_coin_line():
@@ -538,14 +733,14 @@ def test_format_usage_section_cost_nonzero():
assert "billed by provider" in sec
def test_format_review_body_usage_section_between_summary_and_findings():
def test_format_review_body_usage_section_below_findings():
# New layout: header → Summary of Changes → Key Risks → findings → usage.
usage_sec = "## 🔋 AI usage\n\n- model: `m`"
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890",
summary="This PR is risky.", usage_section=usage_sec)
# order: header < summary < usage < findings < marker
assert body.index("risky.") < body.index("AI usage")
assert body.index("AI usage") < body.index("[high]")
assert body.index("[high]") < body.index("<!-- pragent:sha=")
assert body.index("risky.") < body.index("[high]")
assert body.index("[high]") < body.index("AI usage")
assert body.index("AI usage") < body.index("<!-- pragent:sha=")
assert "## 🔋 AI usage" in body
@@ -553,6 +748,46 @@ def test_format_review_body_no_usage_section_omitted():
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890")
assert "AI usage" not in body
def test_format_review_body_with_summary_changes_and_risks():
body = format_review_body(
"", "glm-5.2:cloud", "abcdef1234567890",
summary_changes=["Adds void-death item rescue.", "Adds chunk-loader pause/rename."],
risks=["Mob spawn leak in force-loaded chunks.", "Race on /chunkloader tempo -1."],
findings_for_table=[
{"severity": "high", "path": "a.py", "line": 1, "problem": "race", "fix": "",
"suggestion": "", "reference": ""},
],
inline_count=1,
)
assert "### Summary of Changes" in body
assert "Adds void-death item rescue." in body
assert "Adds chunk-loader pause/rename." in body
assert "### Key Risks & Concerns" in body
assert "Mob spawn leak in force-loaded chunks." in body
assert "### Findings Overview" in body
assert "1 inline comment(s); 1 total." in body
assert "🔴 [HIGH]" in body
assert "`a.py:1`" in body
def test_render_collapsible_usage_contains_details():
usage = {
"model": "glm-5.2:cloud", "input": 1000, "output": 200, "reasoning": 0,
"cache_read": 0, "cache_write": 0, "total": 1200, "steps": 5, "duration_s": 12.0,
"cost": 0.0,
}
block = _render_collapsible_usage(usage, "glm-5.2:cloud", config=None)
assert "<details>" in block
assert "<summary>🔋 AI Usage & Run Details</summary>" in block
assert "</details>" in block
assert "glm-5.2:cloud" in block
assert "1000 in / 200 out" in block
def test_render_collapsible_usage_empty_when_no_usage():
assert _render_collapsible_usage(None, "glm-5.2:cloud", config=None) == ""
# ---------------------------------------------------------------------------
# parse_diff_anchors — empty context lines
# ---------------------------------------------------------------------------