gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: split review pipeline responsibilities

Browse Source
This commit is contained in:
Claude
2026-09-01 01:14:49 +00:00
parent c948f2818b
commit 87dd695d97
9 changed files with 2459 additions and 2380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/review/analysis.py
+469
View File
@@ -0,0 +1,469 @@
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)
# ---------------------------------------------------------------------------