feat(input): add diff_compress module + prior-review compaction helpers #9

Merged
gitea_admin merged 8 commits from feat/cost-display-compress-config into main 2026-08-20 23:05:32 +00:00
3 changed files with 474 additions and 110 deletions
Showing only changes of commit f6be2b3c61 - Show all commits
+14 -3
View File
@@ -142,12 +142,18 @@ containing STRICT JSON, nothing else after it:
```json ```json
{ {
"summary": "One-paragraph overview of the change and its risk.", "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": [ "findings": [
{ {
"severity": "critical|high|medium|low", "severity": "critical|high|medium|low|info|nit",
"path": "path exactly as in the diff `+++ b/` side", "path": "path exactly as in the diff `+++ b/` side",
"line": 12, "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", "fix": "one line: how to fix it",
"suggestion": "exact replacement lines for that location, indented as in the file, or \"\" if no safe replacement", "suggestion": "exact replacement lines for that location, indented as in the file, or \"\" if no safe replacement",
"reference": "https://... or \"\"" "reference": "https://... or \"\""
@@ -157,11 +163,16 @@ containing STRICT JSON, nothing else after it:
``` ```
Rules: 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 — - `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 `""` 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). 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. - 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`. - Do NOT repeat anything in `prior_reviews`.
- The JSON block must be the LAST thing in your message — the Python shell parses - The JSON block must be the LAST thing in your message — the Python shell parses
the last ```json fenced block from your output. If you run out of context/steps the last ```json fenced block from your output. If you run out of context/steps
+300 -77
View File
@@ -156,28 +156,85 @@ def parse_text_blocks(content: list) -> str:
return "\n".join(out).strip() 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. """Format the posted review summary body.
`findings` is the bullet text for findings that could NOT be anchored inline Layout (per the operator's format guide):
(or, on the legacy/no-inline path, the whole review). Empty -> "No issues
found.". `summary` (optional, opencode engine) is rendered as a "Summary" * Header line (``🤖 AI Review …``).
section right under the header. `usage_section` (optional, shown only when * **Summary of Changes** — 24 bullets of what the PR introduces
the PR carries the `AI-USAGE` label) is rendered between the summary and the (`summary_changes`); falls back to the opencode prose `summary` if
findings bullets. The hidden sha marker is always appended for the dedupe the agent didn't emit the list.
pass. * **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") header = REVIEW_HEADER.format(model=model, sha=sha[:8] if sha else "unknown")
findings = (findings or "").strip() parts: list[str] = [header]
if not findings:
findings = "No issues found." # --- Summary of Changes ---
marker = SHA_MARKER.format(sha=sha) if sha else "" sc = list(summary_changes or [])
parts = [header] if not sc and summary:
if summary: sc = _string_list(summary)
parts.append(summary.strip()) 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: if usage_section:
parts.append(usage_section.strip()) parts.append(usage_section.strip())
parts.append(findings)
# --- Hidden marker ---
marker = SHA_MARKER.format(sha=sha) if sha else ""
body = "\n\n".join(parts) body = "\n\n".join(parts)
if marker: if marker:
body += f"\n{marker}" body += f"\n{marker}"
@@ -608,38 +665,73 @@ def salvage_summary(text: str, max_chars: int = SALVAGE_MAX_CHARS) -> str:
) )
def parse_review_output(text: str) -> tuple[str, list[dict]]: def parse_review_output(text: str) -> tuple[str, list[dict], list[str], list[str]]:
"""Parse the opengine's stdout into (summary, findings). """Parse the opengine's stdout into (summary, findings, summary_changes, risks).
Accepts `{"summary": "...", "findings": [...]}` (the opencode pragent agent), Accepts `{"summary": "...", "summary_changes": [...], "risks": [...],
`{"findings": [...]}`, or a bare `[...]` of finding dicts. `summary` defaults "findings": [...]}` (the opencode pragent agent), `{"findings": [...]}`,
to "". Uses the LAST fenced block (the pragent agent emits JSON as the final or a bare `[...]` of finding dicts. `summary_changes` and `risks` default
block), with a tolerant fallback that scans for the last balanced to empty lists; older outputs without them still parse fine. Uses the
object/array in the prose tail. Never raises. 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) blob = _last_json_block(text)
if blob is None: if blob is None:
return "", [] return "", [], [], []
try: try:
data = json.loads(blob) data = json.loads(blob)
except json.JSONDecodeError: except json.JSONDecodeError:
return "", [] return "", [], [], []
summary = ""
summary_changes: list[str] = []
risks: list[str] = []
findings_raw = None
if isinstance(data, dict): if isinstance(data, dict):
summary = str(data.get("summary", "") or "").strip() summary = str(data.get("summary", "") or "").strip()
findings = data.get("findings") summary_changes = _string_list(data.get("summary_changes"))
risks = _string_list(data.get("risks"))
findings_raw = data.get("findings")
elif isinstance(data, list): elif isinstance(data, list):
# Bare array: each item is a finding; no summary. # Bare array: each item is a finding; no summary/sections.
summary = "" findings_raw = data
findings = data
else: else:
return "", [] return "", [], [], []
out = [] out = []
if isinstance(findings, list): if isinstance(findings_raw, list):
for f in findings: for f in findings_raw:
n = _normalize_finding(f) n = _normalize_finding(f)
if n is not None: if n is not None:
out.append(n) out.append(n)
return summary, out return summary, out, summary_changes, risks
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: def _parse_json_tolerant(text: str) -> dict | list | None:
@@ -853,43 +945,175 @@ def _lang_for_path(path: str) -> str:
}.get(ext, "") }.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: def inline_comment_body(f: dict) -> str:
"""Render one finding as a positional review-comment body. """Render one finding as a positional review-comment body.
Includes a fenced suggested-fix block only if the model produced non-empty Shape:
replacement code. The fence is tagged with the file's language (via * Severity badge with emoji (🔴 HIGH / 🟡 MEDIUM / 🔵 LOW / ⚪ INFO).
`_lang_for_path`) so Gitea syntax-highlights it — Gitea 1.26.x has no * 12 short paragraphs: ``problem`` + optional ``fix``.
GitHub-style "Apply suggestion" button (```suggestion is just an * ``suggestion`` block (Gitea/Forgejo apply-on-click) when the model
unknown-language block there → plain monospace), so a language-tagged block produced replacement code. Language-tagged fences are reserved for
is strictly more readable and loses nothing. Appends a `📎 ref:` link when cross-file patterns the suggestion block can't carry.
the finding carries a `reference` URL. * 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() badge = _severity_badge(f.get("severity", "medium"))
body = f"**[{sev}]** {f['problem']}" body = f"{badge} {f.get('problem', '').strip()}"
if f["fix"]: fix = (f.get("fix") or "").strip()
body += f"\n\nFix: {f['fix']}" if fix:
if f["suggestion"]: body += f"\n\n**Fix:** {fix}"
lang = _lang_for_path(f.get("path", "")) suggestion = (f.get("suggestion") or "").strip()
fence = f"```{lang}" if lang else "```" if suggestion:
body += f"\n\n{fence}\n{f['suggestion']}\n```" # `suggestion` fence is the standard one-click-apply block in
ref = f.get("reference", "") # Gitea/Forgejo/GitHub. The agent's replacement lines must already be
if ref: # indented as in the target file.
body += f"\n\n📎 ref: {ref}" body += f"\n\n```suggestion\n{suggestion}\n```"
tok = f.get("_tok_attrib") ref_md = _format_reference(f.get("reference", ""))
if tok is not None: if ref_md:
pct = (f.get("_tok_pct", 0.0) or 0.0) * 100 body += f"\n\n🔗 **Reference:** {ref_md}"
body += f"\n\n🪙 ~{tok} tok ({pct:.0f}% · attributed output)"
return body return body
def summary_bullets(findings: list[dict]) -> str: 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 = [] lines = []
for f in findings: for f in findings:
loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"] loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"]
fix = f" — fix: {f['fix']}" if f["fix"] else "" badge = _severity_badge(f.get("severity", "medium"))
ref = f" ({f.get('reference', '')})" if f.get("reference") else "" problem = f.get("problem", "").strip()
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}{ref}") 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) return "\n".join(lines)
@@ -1442,7 +1666,7 @@ def review_pr(
prior_reviews=prior, model=oc_model, prior_reviews=prior, model=oc_model,
compression_note=compression_note, 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: if not findings and not review_summary:
# The findings JSON was missing or malformed. Don't discard the # The findings JSON was missing or malformed. Don't discard the
# run: salvage the prose, keep the usage report (the label asked # run: salvage the prose, keep the usage report (the label asked
@@ -1454,9 +1678,7 @@ def review_pr(
file=sys.stderr, flush=True, file=sys.stderr, flush=True,
) )
salvaged = salvage_summary(stdout) salvaged = salvage_summary(stdout)
usage_section = "" usage_section = _render_collapsible_usage(usage, model, config=config) if report_usage else ""
if report_usage and usage:
usage_section = format_usage_section(usage, [], model, config=config)
post_review(api, repo, index, token, format_review_body( post_review(api, repo, index, token, format_review_body(
salvaged or "AI review produced no parseable output.", salvaged or "AI review produced no parseable output.",
model, sha, usage_section=usage_section)) model, sha, usage_section=usage_section))
@@ -1491,31 +1713,32 @@ def review_pr(
flush=True, flush=True,
) )
# Attribute output tokens to each finding (mutates finding dicts) so # Compute attribution so inline comments + the table can show per-comment
# inline comments + the usage table can show a per-comment estimate. # estimates. Only meaningful when we have measured usage AND the PR asked
# Only meaningful when we have measured usage AND the PR asked for it. # for it.
usage_section = ""
if report_usage and usage and usage.get("output"): if report_usage and usage and usage.get("output"):
compute_attribution(findings, usage["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) anchors = parse_diff_anchors(diff)
anchored, unanchored = split_findings(findings, anchors) anchored, unanchored = split_findings(findings, anchors)
# Summary body: the unanchored bullets (or "No issues found."), plus a # Summary body: unanchored bullets fall through to a "Unanchored notes"
# one-line note when inline comments were posted so the summary isn't # section; the structured Findings Overview table covers both anchored
# empty-looking. The opencode engine also carries a prose summary. # + unanchored so reviewers see the full set even if inline comments
# are collapsed.
bullets = summary_bullets(unanchored) bullets = summary_bullets(unanchored)
summary_parts = [] summary_parts = []
if anchored:
summary_parts.append(f"_{len(anchored)} inline comment(s) posted below._")
if bullets: if bullets:
summary_parts.append(bullets) summary_parts.append("### Unanchored Notes\n\n" + bullets)
if not summary_parts:
summary_parts.append("No issues found.")
summary_body = format_review_body( summary_body = format_review_body(
"\n\n".join(summary_parts), model, sha, "\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) post_inline_review(api, repo, index, token, summary_body, anchored)
+160 -30
View File
@@ -14,8 +14,10 @@ from ai_review import ( # noqa: E402
_balanced_json_substring, _balanced_json_substring,
_extract_first_json_object, _extract_first_json_object,
_last_balanced_json, _last_balanced_json,
_render_collapsible_usage,
build_user_prompt, build_user_prompt,
compute_attribution, compute_attribution,
findings_table,
format_review_body, format_review_body,
format_usage_section, format_usage_section,
inline_comment_body, inline_comment_body,
@@ -108,17 +110,25 @@ def test_format_review_body_findings():
assert "pragent pilot" in body assert "pragent pilot" in body
assert "glm-5.2:cloud" in body assert "glm-5.2:cloud" in body
assert "`abcdef12`" in body # 8-char sha 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(): def test_format_review_body_empty_findings():
body = format_review_body("", "glm-5.2:cloud", "abcdef1234567890") 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(): def test_format_review_body_whitespace_findings():
body = format_review_body(" \n ", "glm-5.2:cloud", "abcdef1234567890") 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(): def test_format_review_body_no_sha():
@@ -274,33 +284,98 @@ def test_split_findings_by_anchor():
def test_inline_comment_body_with_suggestion(): def test_inline_comment_body_with_suggestion():
f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap", "suggestion": "good()"} f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap", "suggestion": "good()"}
body = inline_comment_body(f) body = inline_comment_body(f)
assert "**[HIGH]**" in body # Severity emoji + bracketed label.
assert "🔴 [HIGH]" in body
assert "bad" in body assert "bad" in body
# no extension → bare fence (Gitea 1.26.x has no apply-suggestion; we tag # Standard ```suggestion fence (Gitea/Forgejo apply-on-click).
# with the file language for highlighting instead of ```suggestion) assert "```suggestion\ngood()\n```" in body
assert "```\ngood()\n```" in body
assert "good()" 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, f = {"severity": "high", "path": "src/Foo.java", "line": 1,
"problem": "bad", "fix": "swap", "suggestion": "good();"} "problem": "bad", "fix": "swap", "suggestion": "good();"}
body = inline_comment_body(f) 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(): def test_inline_comment_body_no_suggestion():
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "f", "suggestion": ""} f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "f", "suggestion": ""}
body = inline_comment_body(f) body = inline_comment_body(f)
assert "```" not in body 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(): def test_summary_bullets_format():
fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f", "suggestion": ""}] fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f", "suggestion": ""}]
b = summary_bullets(fs) b = summary_bullets(fs)
assert "- **[HIGH]**" in b assert "🔴 [HIGH]" in b
assert "`a.py:7`" 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([]) == ""
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -376,7 +451,7 @@ def test_parse_review_output_summary_and_findings():
"]}", "]}",
"\n```", "\n```",
) )
summary, fs = parse_review_output("".join(txt)) summary, fs, *_ = parse_review_output("".join(txt))
assert "eval()" in summary assert "eval()" in summary
assert len(fs) == 1 assert len(fs) == 1
assert fs[0]["severity"] == "critical" assert fs[0]["severity"] == "critical"
@@ -386,16 +461,16 @@ def test_parse_review_output_summary_and_findings():
def test_parse_review_output_bare_findings_no_summary(): def test_parse_review_output_bare_findings_no_summary():
txt = '```json\n{"findings":[{"severity":"low","path":"a","line":1,"problem":"p"}]}\n```' 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 summary == ""
assert len(fs) == 1 assert len(fs) == 1
assert fs[0]["reference"] == "" # default assert fs[0]["reference"] == "" # default
def test_parse_review_output_empty_and_bogus(): def test_parse_review_output_empty_and_bogus():
assert parse_review_output("") == ("", []) assert parse_review_output("") == ("", [], [], [])
assert parse_review_output("no json here") == ("", []) assert parse_review_output("no json here") == ("", [], [], [])
assert parse_review_output('{"findings":[]}') == ("", []) assert parse_review_output('{"findings":[]}') == ("", [], [], [])
def test_parse_review_output_uses_last_json_block(): def test_parse_review_output_uses_last_json_block():
@@ -405,7 +480,7 @@ def test_parse_review_output_uses_last_json_block():
"more prose\n" "more prose\n"
"```json\n{\"summary\":\"real\",\"findings\":[{\"path\":\"y\",\"line\":2,\"severity\":\"high\"}]}\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 summary == "real"
assert len(fs) == 1 assert len(fs) == 1
assert fs[0]["path"] == "y" assert fs[0]["path"] == "y"
@@ -463,7 +538,7 @@ def test_parse_review_output_unfenced_at_tail():
'{"severity":"high","path":"VoidProtection.java","line":162,' '{"severity":"high","path":"VoidProtection.java","line":162,'
'"problem":"drop duplication race","fix":"use ItemMeta","suggestion":"","reference":""}]}' '"problem":"drop duplication race","fix":"use ItemMeta","suggestion":"","reference":""}]}'
) )
summary, fs = parse_review_output(txt) summary, fs, *_ = parse_review_output(txt)
assert "void protection" in summary.lower() assert "void protection" in summary.lower()
assert len(fs) == 1 assert len(fs) == 1
assert fs[0]["path"] == "VoidProtection.java" assert fs[0]["path"] == "VoidProtection.java"
@@ -474,7 +549,7 @@ def test_parse_review_output_bare_array_at_tail():
"All wrapped up.\n" "All wrapped up.\n"
'[{"severity":"low","path":"a","line":1,"problem":"p","fix":"","suggestion":"","reference":""}]' '[{"severity":"low","path":"a","line":1,"problem":"p","fix":"","suggestion":"","reference":""}]'
) )
summary, fs = parse_review_output(txt) summary, fs, *_ = parse_review_output(txt)
assert summary == "" assert summary == ""
assert len(fs) == 1 assert len(fs) == 1
@@ -522,13 +597,26 @@ def test_inline_comment_body_renders_reference():
f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "f", f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "f",
"suggestion": "", "reference": "https://cve.example/X"} "suggestion": "", "reference": "https://cve.example/X"}
body = inline_comment_body(f) 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(): def test_inline_comment_body_no_reference_no_ref_line():
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "", f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "",
"suggestion": "", "reference": ""} "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(): def test_summary_bullets_renders_reference():
@@ -585,13 +673,15 @@ def test_compute_attribution_noop_on_empty_or_zero_budget():
assert "_tok_attrib" not in fs[0] 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", f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap",
"suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29} "suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29}
body = inline_comment_body(f) body = inline_comment_body(f)
assert "🪙 ~180 tok" in body assert "🪙" not in body
assert "29%" in body assert "tok" not in body
assert "attributed output" in body assert "attributed" not in body
def test_inline_comment_body_no_attribution_no_coin_line(): def test_inline_comment_body_no_attribution_no_coin_line():
@@ -643,14 +733,14 @@ def test_format_usage_section_cost_nonzero():
assert "billed by provider" in sec 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`" usage_sec = "## 🔋 AI usage\n\n- model: `m`"
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890", body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890",
summary="This PR is risky.", usage_section=usage_sec) summary="This PR is risky.", usage_section=usage_sec)
# order: header < summary < usage < findings < marker assert body.index("risky.") < body.index("[high]")
assert body.index("risky.") < body.index("AI usage") assert body.index("[high]") < body.index("AI usage")
assert body.index("AI usage") < body.index("[high]") assert body.index("AI usage") < body.index("<!-- pragent:sha=")
assert body.index("[high]") < body.index("<!-- pragent:sha=")
assert "## 🔋 AI usage" in body assert "## 🔋 AI usage" in body
@@ -658,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") body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890")
assert "AI usage" not in body 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 # parse_diff_anchors — empty context lines
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------