Compare commits
2 Commits
ec26ec000a
...
f6be2b3c61
| Author | SHA1 | Date | |
|---|---|---|---|
 Marcos
|
f6be2b3c61 | ||
|
Marcos
|
5302e8dcd7 |
@@ -142,12 +142,18 @@ containing STRICT JSON, nothing else after it:
|
||||
```json
|
||||
{
|
||||
"summary": "One-paragraph overview of the change and its risk.",
|
||||
"summary_changes": [
|
||||
"2–4 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": "1–2 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` (2–4 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 1–2 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).
|
||||
+460
-92
@@ -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** — 2–4 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):
|
||||
return []
|
||||
if isinstance(data, dict):
|
||||
findings = data.get("findings")
|
||||
elif isinstance(data, list):
|
||||
findings = data
|
||||
else:
|
||||
return []
|
||||
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 "", []
|
||||
return "", [], [], []
|
||||
summary = ""
|
||||
summary_changes: list[str] = []
|
||||
risks: list[str] = []
|
||||
findings_raw = None
|
||||
if isinstance(data, dict):
|
||||
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):
|
||||
# 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).
|
||||
* 1–2 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)
|
||||
|
||||
@@ -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
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user