"""MCP tool wrappers for the unified-review scoring workflow.
Wraps :mod:`code_review_graph.scoring` (score_review / dedupe_findings /
build_report_data) into the three MCP tools consumed by the
``unified-review`` skill:
* ``score_review_tool`` - objective Layer-2 metrics for changed files
* ``dedupe_findings_tool`` - fingerprint dedup + confidence merge
* ``generate_report_tool`` - render ``code-review-report.html``
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from ..incremental import get_changed_files, get_staged_and_unstaged
from ..scoring import build_report_data, dedupe_findings, score_review
from ._common import _get_store, _error_response
try:
from importlib.resources import files as _pkg_files # Python 3.9+
_HAS_IMPORTLIB_RESOURCES = True
except ImportError: # pragma: no cover
_HAS_IMPORTLIB_RESOURCES = False
# ---------------------------------------------------------------------------
# Tool: score_review
# ---------------------------------------------------------------------------
def score_review_func(
changed_files: list[str] | None = None,
base: str = "HEAD~1",
include_churn: bool = True,
repo_root: str | None = None,
detail_level: str = "standard",
) -> dict[str, Any]:
"""Compute objective Layer-2 review metrics for changed files.
Runs the git-history / graph risk factors plus the heuristic metrics
(SQL risk, exception coverage, redundancy, high-risk density,
vulnerability). LLM-judged metrics are listed in ``llm_judged`` so the
calling agent knows what still needs judgement.
Args:
changed_files: Files to score (auto-detected from git diff if
omitted).
base: Git ref to diff against (default: HEAD~1).
include_churn: Include git-churn risk factors (default: True).
repo_root: Repository root. Auto-detected if omitted.
detail_level: Output detail level. ``minimal`` returns only grades
and values; ``standard`` includes evidence.
"""
store, root = _get_store(repo_root)
try:
if changed_files is None:
changed_files = get_changed_files(root, base)
if not changed_files:
changed_files = get_staged_and_unstaged(root)
if not changed_files:
return {
"status": "ok",
"summary": "No changed files detected. Nothing to score.",
"metrics": {},
"objective_grade": "good",
}
result = score_review(
store,
root,
changed_files,
include_churn=include_churn,
)
if detail_level == "minimal":
return {
"status": "ok",
"summary": result["summary"],
"objective_grade": result["objective_grade"],
"metrics": {
name: {"value": m["value"], "grade": m["grade"]}
for name, m in result["metrics"].items()
},
"llm_judged": result["llm_judged"],
}
result["changed_files"] = changed_files
result["next_tool_suggestions"] = [
"dedupe_findings -- merge specialist findings",
"detect_changes -- risk-scored impact analysis",
"generate_report -- export HTML report",
]
return result
except Exception as exc:
return _error_response(str(exc))
finally:
store.close()
# ---------------------------------------------------------------------------
# Tool: dedupe_findings
# ---------------------------------------------------------------------------
def dedupe_findings_func(
findings: list[dict[str, Any]],
suppress_prior: list[dict[str, Any]] | None = None,
repo_root: str | None = None,
) -> dict[str, Any]:
"""Merge review findings by fingerprint and boost multi-source confidence.
Mirrors the gstack-review "collect and merge" step: findings with the
same ``path:line:category`` fingerprint are merged (highest confidence
wins); findings confirmed by more than one source get confidence +1
(cap 10). Confidence gates route low-confidence findings to the
appendix or suppress them, and a PR quality score is computed as
``max(0, 10 - (critical*2 + informational*0.5))``.
Args:
findings: Raw finding dicts with ``path``, ``category``,
``severity``, ``confidence`` and optional ``source``/``line``.
suppress_prior: Previously user-skipped findings (from a prior
review-log) to suppress when their file has not changed.
repo_root: Repository root (used to resolve changed files when
suppressing prior findings).
"""
try:
suppressed_prior_list: list[dict[str, Any]] = []
if suppress_prior:
for f in suppress_prior:
suppressed_prior_list.append(f)
result = dedupe_findings(findings, suppressed_prior_list)
result["next_tool_suggestions"] = [
"generate_report -- export HTML report",
"detect_changes -- risk-scored impact analysis",
]
return result
except Exception as exc:
return _error_response(str(exc))
# ---------------------------------------------------------------------------
# Tool: generate_report
# ---------------------------------------------------------------------------
def _load_report_template() -> str:
"""Load ``report-template.html`` from the installed package assets.
Falls back to the bundled template string when the asset cannot be
loaded, so the tool never fails solely because the package data is
missing.
"""
if _HAS_IMPORTLIB_RESOURCES:
try:
data = _pkg_files("code_review_graph").joinpath(
"assets/report-template.html"
).read_text(encoding="utf-8")
if data:
return data
except (FileNotFoundError, OSError):
pass
return _FALLBACK_TEMPLATE
# Minimal self-contained template (used if the asset file is unavailable).
_FALLBACK_TEMPLATE = """
Code Review Report
"""
def generate_report_func(
review_data: dict[str, Any],
output_path: str | None = None,
repo_root: str | None = None,
) -> dict[str, Any]:
"""Generate a self-contained HTML code review report.
Injects ``review_data`` (the output of ``score_review_tool`` plus
``dedupe_findings_tool`` results and free-form verdict/tier/scope) into
the bundled ``report-template.html`` as ``{{REPORT_DATA}}`` and writes
the standalone file (default ``repo_root/code-review-report.html``).
Args:
review_data: Review data dict (metrics, findings, verdict, tier,
scope, files, baseline, timestamp, manual_review).
output_path: Output file path. Defaults to
``/code-review-report.html``.
repo_root: Repository root. Auto-detected if omitted.
"""
store, root = _get_store(repo_root)
try:
data = build_report_data(review_data)
data["summary"] = review_data.get("summary", "")
template = _load_report_template()
rendered = template.replace(
"{{REPORT_DATA}}", json.dumps(data, ensure_ascii=False)
)
if output_path:
out = Path(output_path)
if not out.is_absolute():
out = root / out
else:
out = root / "code-review-report.html"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(rendered, encoding="utf-8")
return {
"status": "ok",
"summary": f"Report written to {out}",
"output_path": str(out),
"report_size_bytes": len(rendered),
}
except Exception as exc:
return _error_response(str(exc))
finally:
store.close()