feat(examples): add review report builder and markdown renderer

Author: DNKYr

examples/skills_code_review_agent/review/report.py

@@ -0,0 +1,121 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Report assembly and rendering."""
+import json
+import os
+from datetime import datetime
+
+from .findings import severity_distribution
+from .redaction import redact_text
+
+_BLOCKING = {"critical", "high"}
+
+
+def _conclusion(reported, filter_events):
+    if any(f.severity in _BLOCKING for f in reported):
+        return "blocked"
+    intercepted = any(e.decision != "allow" for e in filter_events)
+    if reported or intercepted:
+        return "needs_attention"
+    return "pass"
+
+
+def build_report(task_id, input_ref, runtime, dry_run, diff_summary, reported,
+                 needs_review, deduped_count, filter_events, sandbox_outcomes,
+                 metrics, llm_summary, warnings) -> dict:
+    """Assemble the full report dict (issue acceptance criterion #8)."""
+    return {
+        "task_id": task_id,
+        "generated_at": datetime.now().isoformat(timespec="seconds"),
+        "input": {"ref": input_ref, "runtime": runtime, "dry_run": dry_run,
+                  "diff_summary": diff_summary},
+        "conclusion": _conclusion(reported, filter_events),
+        "summary": {
+            "total_findings": len(reported),
+            "by_severity": severity_distribution(reported),
+            "needs_human_review_count": len(needs_review),
+            "deduplicated": deduped_count,
+            "intercepts": sum(1 for e in filter_events if e.decision != "allow"),
+        },
+        "findings": [f.model_dump() for f in reported],
+        "needs_human_review": [f.model_dump() for f in needs_review],
+        "filter_intercepts": [
+            {"target": e.target, "decision": e.decision, "rule": e.rule, "reason": e.reason}
+            for e in filter_events if e.decision != "allow"],
+        "sandbox_runs": [
+            {"script": o.script, "status": o.status, "exit_code": o.exit_code,
+             "duration_ms": o.duration_ms, "timed_out": o.timed_out,
+             "error_type": o.error_type} for o in sandbox_outcomes],
+        "metrics": metrics,
+        "llm_summary": llm_summary,
+        "warnings": list(warnings),
+    }
+
+
+def _md_findings_table(findings):
+    if not findings:
+        return "_none_\n"
+    lines = ["| Severity | Category | File | Line | Title | Confidence | Source |",
+             "|---|---|---|---|---|---|---|"]
+    for f in findings:
+        lines.append(f"| {f['severity']} | {f['category']} | {f['file']} | {f['line']} "
+                     f"| {f['title']} | {f['confidence']} | {f['source']} |")
+    return "\n".join(lines) + "\n"
+
+
+def render_markdown(report: dict) -> str:
+    """Render the human-readable markdown report."""
+    s = report["summary"]
+    parts = [
+        "# Code Review Report",
+        f"- Task: `{report['task_id']}`",
+        f"- Conclusion: **{report['conclusion']}**",
+        f"- Input: {report['input']['ref']} (runtime={report['input']['runtime']}, "
+        f"dry_run={report['input']['dry_run']})",
+        f"- Findings: {s['total_findings']} (by severity: {s['by_severity']}), "
+        f"needs human review: {s['needs_human_review_count']}, "
+        f"deduplicated: {s['deduplicated']}, intercepts: {s['intercepts']}",
+        "",
+        "## Findings",
+        _md_findings_table(report["findings"]),
+        "### Recommendations",
+        "\n".join(f"- `{f['file']}:{f['line']}` {f['recommendation']}"
+                  for f in report["findings"]) or "_none_",
+        "",
+        "## Needs Human Review",
+        _md_findings_table(report["needs_human_review"]),
+        "## Filter Intercepts",
+        "\n".join(f"- [{e['decision']}] `{e['target']}` ({e['rule']}): {e['reason']}"
+                  for e in report["filter_intercepts"]) or "_none_",
+        "",
+        "## Sandbox Runs",
+        "\n".join(f"- `{o['script']}`: {o['status']} (exit={o['exit_code']}, "
+                  f"{o['duration_ms']}ms, timed_out={o['timed_out']})"
+                  for o in report["sandbox_runs"]) or "_none_",
+        "",
+        "## Metrics",
+        "```json",
+        json.dumps(report["metrics"], indent=2),
+        "```",
+        "",
+        f"## LLM Summary\n{report['llm_summary'] or '_none_'}",
+        "",
+        "## Warnings",
+        "\n".join(f"- {w}" for w in report["warnings"]) or "_none_",
+    ]
+    return "\n".join(parts) + "\n"
+
+
+def write_reports(report: dict, output_dir: str):
+    """Write review_report.json / review_report.md (final redaction pass)."""
+    os.makedirs(output_dir, exist_ok=True)
+    json_path = os.path.join(output_dir, "review_report.json")
+    md_path = os.path.join(output_dir, "review_report.md")
+    with open(json_path, "w", encoding="utf-8") as fh:
+        fh.write(redact_text(json.dumps(report, indent=2, ensure_ascii=False, default=str)))
+    with open(md_path, "w", encoding="utf-8") as fh:
+        fh.write(redact_text(render_markdown(report)))
+    return json_path, md_path

examples/skills_code_review_agent/tests/test_report.py

@@ -0,0 +1,74 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tests for report building and rendering."""
+import json
+from pathlib import Path
+
+from review.findings import Finding
+from review.governance import GovernanceDecision
+from review.report import build_report, render_markdown, write_reports
+from review.sandbox import SandboxRunOutcome
+
+
+def _report(reported=None, events=None):
+    return build_report(
+        task_id="t1", input_ref="x.diff", runtime="local", dry_run=True,
+        diff_summary={"files_changed": 1},
+        reported=reported or [],
+        needs_review=[Finding(severity="low", category="db_lifecycle", file="b.py",
+                              line=2, title="cursor", confidence=0.5)],
+        deduped_count=1,
+        filter_events=events or [GovernanceDecision("curl x", "deny", "network_policy", "no net")],
+        sandbox_outcomes=[SandboxRunOutcome(script="check_security.py", status="ok",
+                                            exit_code=0, duration_ms=10, timed_out=False,
+                                            stdout="{}", stderr="")],
+        metrics={"total_duration_ms": 100, "sandbox_duration_ms": 50, "tool_calls": 6,
+                 "intercepts": 1, "findings_total": 1,
+                 "severity_distribution": {"high": 1}, "error_distribution": {}},
+        llm_summary="fine", warnings=["w1"])
+
+
+def test_conclusion_blocked_on_high():
+    rep = _report(reported=[Finding(severity="high", category="security", file="a.py",
+                                    line=1, title="eval", confidence=0.9)])
+    assert rep["conclusion"] == "blocked"
+
+
+def test_conclusion_needs_attention_on_intercepts_only():
+    rep = _report()
+    assert rep["conclusion"] == "needs_attention"
+
+
+def test_conclusion_pass_when_clean():
+    rep = _report(events=[GovernanceDecision("ok", "allow")])
+    assert rep["conclusion"] == "pass"
+
+
+def test_report_sections_present():
+    rep = _report()
+    for key in ("summary", "findings", "needs_human_review", "filter_intercepts",
+                "sandbox_runs", "metrics", "llm_summary", "warnings"):
+        assert key in rep
+    assert rep["summary"]["needs_human_review_count"] == 1
+    assert rep["summary"]["deduplicated"] == 1
+
+
+def test_markdown_contains_sections():
+    md = render_markdown(_report())
+    for heading in ("# Code Review Report", "## Findings", "## Needs Human Review",
+                    "## Filter Intercepts", "## Sandbox Runs", "## Metrics"):
+        assert heading in md
+
+
+def test_write_reports_redacts(tmp_path):
+    rep = _report(reported=[Finding(
+        severity="critical", category="secret_leak", file="s.py", line=1,
+        title="secret", evidence='k = "sk-abcdefghijklmnopqrstuvwxyz123456"',
+        confidence=0.95)])
+    json_path, md_path = write_reports(rep, str(tmp_path))
+    text = Path(json_path).read_text() + Path(md_path).read_text()
+    assert "sk-abcdefghijklmnopqrstuvwxyz123456" not in text
+    assert json.loads(Path(json_path).read_text())["task_id"] == "t1"