Claude
470 lines
18 KiB
Python
470 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
from . import pipeline
|
||
from .pipeline import *
|
||
from .pipeline import _CONFIDENCE_BADGE
|
||
from .configuration import effective_config
|
||
from .output import _string_list, findings_table
|
||
|
||
# Pure helpers (unit-tested, no network)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def truncate_diff(text: str, max_chars: int) -> tuple[str, bool, int]:
|
||
"""Return (text, was_truncated, original_len). Never raises on bad input."""
|
||
if text is None:
|
||
return "", False, 0
|
||
orig_len = len(text)
|
||
if orig_len <= max_chars:
|
||
return text, False, orig_len
|
||
return text[:max_chars] + f"\n\n[diff truncated at {max_chars} characters]\n", True, orig_len
|
||
|
||
|
||
def fmt_tokens(n) -> str:
|
||
"""1234567 -> '1,234,567 (1.2M)'; 0 -> '0'; <1000 -> comma-only; None/negative -> '?'.
|
||
|
||
Always returns the full comma-separated number; the short suffix is a
|
||
parenthetical for fast scanning. Caps at B; the cost model never exceeds M.
|
||
"""
|
||
if n is None:
|
||
return "?"
|
||
if not isinstance(n, (int, float)) or n < 0:
|
||
return "?"
|
||
n = int(n)
|
||
if n < 1000:
|
||
return f"{n:,}"
|
||
if n < 1_000_000:
|
||
return f"{n:,} ({n / 1000:.1f}K)"
|
||
if n < 1_000_000_000:
|
||
return f"{n:,} ({n / 1_000_000:.1f}M)"
|
||
return f"{n:,} ({n / 1_000_000_000:.1f}B)"
|
||
|
||
|
||
def parse_text_blocks(content: list) -> str:
|
||
"""Join `type:"text"` blocks from an Anthropic /v1/messages response.
|
||
|
||
Drops `thinking` blocks (glm-5.2:cloud is a reasoning model and emits them).
|
||
Tolerates missing/malformed blocks by skipping them.
|
||
"""
|
||
from model_client import parse_text_blocks as _parse_text_blocks
|
||
return _parse_text_blocks(content)
|
||
|
||
|
||
def _int_env(name: str, default: int) -> int:
|
||
"""Read an int from the environment, falling back on anything unparseable.
|
||
|
||
A typo in a tuning knob must not take down a review that is already
|
||
mid-flight — the operator gets a stderr line and the default instead.
|
||
"""
|
||
raw = os.environ.get(name, "")
|
||
if not str(raw).strip():
|
||
return default
|
||
try:
|
||
return int(str(raw).strip())
|
||
except (TypeError, ValueError):
|
||
print(
|
||
f"pragent: ignoring {name}={raw!r} (not an integer); using {default}",
|
||
file=sys.stderr, flush=True,
|
||
)
|
||
return default
|
||
|
||
|
||
# 1-5 merge-verdict score (higher = safer). Buckets:
|
||
# 5 = clean (or low/info/trivial only — nothing worth blocking on)
|
||
# 4 = medium present
|
||
# 3 = high present (operator should at least look)
|
||
# 1 = critical present (block the merge by default)
|
||
# Cross-lens agreement on any finding takes one more off, floored at 1.
|
||
_CONFIDENCE_BADGE = {5: "🟢", 4: "🟢", 3: "🟡", 2: "🟠", 1: "🔴"}
|
||
|
||
|
||
def merge_confidence(findings: list[dict], *, multi_lens_observed: bool = False) -> int:
|
||
"""1-5 merge verdict: higher = safer.
|
||
|
||
Tier drops driven by the most severe finding present:
|
||
- critical → 1
|
||
- high → 3
|
||
- medium → 4
|
||
- else → 5 (low / trivial / info / unknown → no drop)
|
||
|
||
An extra -1 when cross-lens agreement was observed on any finding
|
||
(``multi_lens_observed``). The flag is passed in explicitly because the
|
||
raw ``_multi_lens`` marker is stripped from findings by the time they
|
||
reach this function — first by ``opencode_review.run_lenses_review``
|
||
(the ``_``-prefix scrub) and again by ``_normalize_finding`` (the
|
||
7-key schema rebuild). The caller (``review_pr``) must capture the
|
||
signal before those strips fire. Final score is clamped to [1, 5] so
|
||
a critical + multi_lens combo doesn't go negative.
|
||
"""
|
||
if not findings:
|
||
return 5
|
||
max_rank = max(SEVERITY_RANK.get(f.get("severity", "low"), 0) for f in findings)
|
||
if max_rank >= SEVERITY_RANK["critical"]:
|
||
score = 1
|
||
elif max_rank >= SEVERITY_RANK["high"]:
|
||
score = 3
|
||
elif max_rank >= SEVERITY_RANK["medium"]:
|
||
score = 4
|
||
else:
|
||
score = 5
|
||
if multi_lens_observed:
|
||
score -= 1
|
||
return max(1, min(5, score))
|
||
|
||
|
||
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,
|
||
confidence: int = 5,
|
||
walkthrough: list[str] | None = None,
|
||
risk_verdict: str = "",
|
||
test_coverage: str = "",
|
||
static_message: str = "",
|
||
|
||
) -> str:
|
||
"""Format the posted review summary body.
|
||
|
||
Layout (per the operator's format guide):
|
||
|
||
* Header line (``🤖 AI Review …``) including the merge-confidence badge.
|
||
* Optional static banner (``> {static_message}``) — repo-wide call-out
|
||
from `.pr-review.json:static_message`, placed under the header so
|
||
every reviewer sees it on every review without scrolling.
|
||
|
||
* **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.
|
||
* **Risk Verdict** — one-line "<level> risk: <reason>" verdict
|
||
(`risk_verdict`); omitted when empty.
|
||
* **Walkthrough** — up to 6 file- or change-grouped bullets
|
||
(`walkthrough`); the file part is wrapped in backticks so paths
|
||
render as code in Gitea. Omitted when empty.
|
||
* **Test Coverage** — short `test_coverage` string ("Tests added" /
|
||
etc.); omitted when empty.
|
||
* **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.
|
||
|
||
`confidence` is a 1-5 merge verdict rendered as `<N>/5 <badge>` in the
|
||
header. Clamped to [1, 5] so a stray value (e.g. 0 from a missing
|
||
finding list) doesn't print a broken badge.
|
||
|
||
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.
|
||
"""
|
||
score = max(1, min(5, confidence))
|
||
badge = _CONFIDENCE_BADGE.get(score, "🟢")
|
||
confidence_str = f"{score}/5 {badge}"
|
||
header = REVIEW_HEADER.format(
|
||
model=model,
|
||
sha=sha[:8] if sha else "unknown",
|
||
confidence=confidence_str,
|
||
)
|
||
parts: list[str] = [header]
|
||
|
||
# Optional free-text banner. Rendered as a Markdown blockquote immediately
|
||
# after the header — front-of-mind for any maintainer scanning the review.
|
||
if static_message and static_message.strip():
|
||
parts.append(f"> {static_message.strip()}")
|
||
|
||
# --- 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._")
|
||
|
||
# --- Risk Verdict ---
|
||
if risk_verdict:
|
||
parts.append(f"### Risk Verdict\n\n{risk_verdict}")
|
||
|
||
# --- Walkthrough ---
|
||
wt = list(walkthrough or [])
|
||
if wt:
|
||
wt = wt[:6]
|
||
rendered = []
|
||
for item in wt:
|
||
# Items typically look like "a.py — adds X" (em-dash separator).
|
||
# Wrap the file path in backticks so it renders as code in the
|
||
# Gitea markdown body; leave the description as plain prose. When
|
||
# no separator is present, render the whole line as plain prose
|
||
# (the agent's "plain prose" fallback for change-grouped bullets).
|
||
if " — " in item:
|
||
path, _, rest = item.partition(" — ")
|
||
rendered.append(f"- `{path}` — {rest}")
|
||
else:
|
||
rendered.append(f"- {item}")
|
||
parts.append(f"### Walkthrough\n\n" + "\n".join(rendered))
|
||
|
||
# --- Test Coverage ---
|
||
if test_coverage:
|
||
parts.append(f"### Test Coverage\n\n{test_coverage}")
|
||
|
||
# --- 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())
|
||
|
||
# --- Hidden marker ---
|
||
marker = SHA_MARKER.format(sha=sha) if sha else ""
|
||
|
||
body = "\n\n".join(parts)
|
||
if marker:
|
||
body += f"\n{marker}"
|
||
return body
|
||
|
||
|
||
def _finding_weight(f: dict) -> int:
|
||
"""Body-weight used to attribute output tokens to a finding (char length of
|
||
its rendered problem + fix + suggestion). One model pass produces all
|
||
findings, so per-finding tokens can't be measured directly — we split the
|
||
measured output total by this weight as an honest attribution."""
|
||
return (
|
||
len(f.get("problem") or "")
|
||
+ len(f.get("fix") or "")
|
||
+ len(f.get("suggestion") or "")
|
||
)
|
||
|
||
|
||
def compute_attribution(findings: list[dict], output_tokens: int) -> None:
|
||
"""Stash `_tok_attrib` (attributed output tokens) and `_tok_pct` (0..1) on
|
||
each finding, splitting `output_tokens` by each finding's body weight.
|
||
Mutates in place. No-op when there are no findings or no output budget."""
|
||
if not findings or not output_tokens:
|
||
return
|
||
weights = [_finding_weight(f) for f in findings]
|
||
total_w = sum(weights)
|
||
if total_w <= 0:
|
||
# All-zero weights (no prose): split evenly.
|
||
share = output_tokens / len(findings)
|
||
for f in findings:
|
||
f["_tok_attrib"] = int(round(share))
|
||
f["_tok_pct"] = 1.0 / len(findings)
|
||
return
|
||
for f, w in zip(findings, weights):
|
||
f["_tok_attrib"] = int(round(output_tokens * w / total_w))
|
||
f["_tok_pct"] = w / total_w
|
||
|
||
|
||
def _resolve_price_target(config: dict | None) -> tuple[str, str | None]:
|
||
"""Pick which provider to compute the equivalent cost against.
|
||
|
||
Order: `.pr-review.json:cost_target` > `PRAGENT_PRICE_TARGET` env >
|
||
`DEFAULT_PRICE_TARGET` (claude-sonnet-5). Returns `(price_key, error)`.
|
||
|
||
If any of the user-set keys is unknown, falls back to the default AND
|
||
reports the error so the operator sees their typo (a config-level typo
|
||
silently picking the default would defeat the purpose of letting repos
|
||
opt into a different comparison model).
|
||
"""
|
||
from cost_model import PRICES # local import keeps ollama path dep-free
|
||
candidates: list[tuple[str, str]] = []
|
||
if isinstance(config, dict) and config.get("cost_target"):
|
||
candidates.append(("repo config", str(config["cost_target"]).strip()))
|
||
env = os.environ.get("PRAGENT_PRICE_TARGET", "").strip()
|
||
if env:
|
||
candidates.append(("PRAGENT_PRICE_TARGET env", env))
|
||
candidates.append(("default", DEFAULT_PRICE_TARGET))
|
||
|
||
chosen = DEFAULT_PRICE_TARGET
|
||
for source, key in candidates:
|
||
if key in PRICES:
|
||
chosen = key
|
||
break
|
||
else:
|
||
# No candidate was valid. Use default + report.
|
||
return chosen, (
|
||
f"unknown price target (checked {', '.join(f'{s}={k!r}' for s, k in candidates)}); "
|
||
f"valid: {', '.join(sorted(PRICES))}"
|
||
)
|
||
|
||
# Even when we picked a valid key, if the *user* set one and it was
|
||
# unknown, surface that. (We only get here if a later candidate resolved,
|
||
# so the invalid one was upstream.)
|
||
invalid = [(s, k) for s, k in candidates if k not in PRICES and s != "default"]
|
||
if invalid:
|
||
return chosen, (
|
||
f"unknown price target (set {', '.join(f'{s}={k!r}' for s, k in invalid)}); "
|
||
f"valid: {', '.join(sorted(PRICES))}; falling back to `{chosen}`"
|
||
)
|
||
return chosen, None
|
||
|
||
|
||
def _resolve_display_model(base_model: str, config: dict | None) -> str:
|
||
"""Resolve the *display* model for one review.
|
||
|
||
Precedence (highest first):
|
||
1. `OPENCODE_MODEL` env var — operator override, used as-is (already a
|
||
provider-prefixed opencode ref like `headroom/MiniMax-M2.7`).
|
||
2. `.pr-review.json:model` — per-repo override. Already validated
|
||
against `cost_model.PRICES` by `parse_repo_config`, so a bare key
|
||
like `claude-sonnet-5` or `qwen3.8-27b` is safe. Re-prefixed with
|
||
the model's `provider` field from `cost_model.Price` (default
|
||
`headroom`) so the opencode subprocess routes correctly — e.g.
|
||
`qwen3.8-27b` → `vllm-qwen38/qwen3.8-27b` (vLLM on RTX 3090 at
|
||
192.168.1.79:18020), `claude-sonnet-5` → `headroom/claude-sonnet-5`
|
||
(Anthropic pricing proxy).
|
||
3. Default — `f"headroom/{base_model}"` where `base_model` is the bare
|
||
`OLLAMA_MODEL` (e.g. `"MiniMax-M2.7" → "headroom/MiniMax-M2.7"`).
|
||
|
||
The same value flows to every consumer (opencode subprocess, REVIEW_HEADER,
|
||
cost-line parenthetical) so reviewers never see a mix of `glm-5.2:cloud`
|
||
and the routed model in one body.
|
||
"""
|
||
env = os.environ.get("OPENCODE_MODEL")
|
||
if env:
|
||
return env
|
||
cfg_model = (config or {}).get("model")
|
||
if isinstance(cfg_model, str) and cfg_model.strip():
|
||
# Look up the provider from PRICES so the opencode subprocess routes
|
||
# through the right provider block (vllm-qwen38 vs headroom). Lazy
|
||
# import — the ollama path doesn't touch cost_model.
|
||
from cost_model import PRICES
|
||
provider = PRICES.get(cfg_model.strip())
|
||
if provider is not None:
|
||
return f"{provider.provider}/{cfg_model.strip()}"
|
||
# parse_repo_config already drops unknowns, but stay defensive: fall
|
||
# back to headroom so the review still runs rather than crash.
|
||
return f"headroom/{cfg_model.strip()}"
|
||
return f"headroom/{base_model}"
|
||
|
||
|
||
def equivalent_cost(usage: dict, price_key: str) -> float:
|
||
"""USD the measured usage would have billed on `price_key`'s provider.
|
||
|
||
`usage` is the dict from `parse_opencode_events` (input/output/reasoning/
|
||
cache_read/cache_write). Builds a `cost_model.Usage` and runs `cost()`. The
|
||
pilot's actual provider (headroom/glm-5.2:cloud) reports $0 — this is what
|
||
the same tokens would cost on a paid model, so maintainers can budget.
|
||
"""
|
||
from cost_model import Usage, cost, PRICES # local import: ollama path dep-free
|
||
if price_key not in PRICES:
|
||
return 0.0
|
||
u = Usage(
|
||
uncached_input=(usage.get("input", 0) - usage.get("cache_read", 0)),
|
||
cached_input=usage.get("cache_read", 0),
|
||
cache_writes=usage.get("cache_write", 0),
|
||
output=usage.get("output", 0),
|
||
)
|
||
return cost(u, PRICES[price_key])
|
||
|
||
|
||
def build_user_prompt(
|
||
title: str,
|
||
body: str,
|
||
diff: str,
|
||
config: dict | None = None,
|
||
prior_reviews: list[str] | None = None,
|
||
additional_context: str = "",
|
||
) -> str:
|
||
"""Assemble the user prompt: repo config + additional context + prior reviews + PR meta + diff."""
|
||
parts: list[str] = []
|
||
|
||
eff = effective_config(config) if config else {}
|
||
if eff:
|
||
cfg_lines = []
|
||
if eff.get("focus"):
|
||
cfg_lines.append("Focus areas: " + ", ".join(eff["focus"]))
|
||
if eff.get("exclude_paths"):
|
||
cfg_lines.append("Ignore paths: " + ", ".join(eff["exclude_paths"]))
|
||
if eff.get("languages"):
|
||
cfg_lines.append("Languages: " + ", ".join(eff["languages"]))
|
||
if eff.get("style"):
|
||
cfg_lines.append(f"Review style: {eff['style']} "
|
||
f"(max {eff['max_findings']} findings, threshold "
|
||
f"{eff['severity_threshold']}+)")
|
||
if eff.get("patterns", {}).get("allow"):
|
||
cfg_lines.append("Allow paths (only these are reviewed): "
|
||
+ ", ".join(eff["patterns"]["allow"]))
|
||
if eff.get("patterns", {}).get("deny"):
|
||
cfg_lines.append("Deny paths: " + ", ".join(eff["patterns"]["deny"]))
|
||
if eff.get("exclude_tests"):
|
||
cfg_lines.append("Skip test files entirely.")
|
||
if eff.get("require_tests"):
|
||
cfg_lines.append("Flag behavioral changes that don't add a test "
|
||
"alongside (added as a `low` finding).")
|
||
if eff.get("instructions"):
|
||
cfg_lines.append("Instructions:\n" + str(eff["instructions"]).strip())
|
||
if cfg_lines:
|
||
parts.append("## Repo review config (.pr-review.json)\n" + "\n".join(cfg_lines))
|
||
|
||
if additional_context:
|
||
# Repo-provided static background (architecture summary, module map,
|
||
# conventions, glossary, …). Cached for the review; the agent reads
|
||
# this ONCE per review and the prompt-cached prefix absorbs it on
|
||
# later steps — much cheaper than re-discovering the same facts from
|
||
# the source tree on every PR.
|
||
parts.append(
|
||
"## Repo-provided context (.pr-review.json:additional_context_urls "
|
||
"+ PRAGENT_ADDITIONAL_CONTEXT_URL — cached per review)\n" + additional_context
|
||
)
|
||
|
||
if prior_reviews:
|
||
joined = "\n\n---\n\n".join(prior_reviews)
|
||
if len(joined) > 4000:
|
||
joined = joined[:4000] + "\n…[prior reviews truncated]"
|
||
parts.append("## PREVIOUS REVIEWS (already posted — do NOT repeat these points)\n" + joined)
|
||
|
||
parts.append(f"## PR\nTitle: {title or '(none)'}")
|
||
if body and body.strip():
|
||
b = body.strip()
|
||
if len(b) > 4000:
|
||
b = b[:4000] + "\n…[PR body truncated]"
|
||
parts.append(f"Description:\n{b}")
|
||
parts.append(f"## Diff\n```diff\n{diff}\n```")
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|