#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from typing import Any _SEVERITY_ORDER = {"P0": 0, "P1": 1, "P2": 2, "P3": 3} _STATUS_ICON = {"PASS": "\u2705", "FAIL": "\u274c", "SKIP": "\u23ed\ufe0f", "BLOCKED": "\u26d4"} def _verdict(payload: dict[str, Any]) -> str: """Determine review verdict from invariant results and verification runs. FAIL or BLOCKED in either → request_changes. Otherwise → approve. """ for result in payload.get("invariant_results", []): if str(result.get("status", "")).upper() == "FAIL": return "request_changes" for run in payload.get("verification_runs", []): if str(run.get("status", "")).upper() in {"FAIL", "BLOCKED"}: return "request_changes" return "approve" def build_markdown(payload: dict[str, Any]) -> str: """Render a full PR review comment from the orchestrator payload.""" base = payload.get("base", "main") head = payload.get("head", "HEAD") changed_files: list[str] = payload.get("changed_files", []) diff_stat: str = payload.get("diff_stat", "") risk_map: dict[str, Any] = payload.get("risk_map", {}) shards: dict[str, Any] = payload.get("shards", {}) invariant_results: list[dict[str, Any]] = payload.get("invariant_results", []) verification_plan: list[dict[str, Any]] = payload.get("verification_plan", []) verification_runs: list[dict[str, Any]] = payload.get("verification_runs", []) overall_risk = risk_map.get("overall_risk", "P3") docs_only = risk_map.get("docs_only", False) verdict = _verdict(payload) lines: list[str] = [] # -- Summary -- lines.append("## Summary") lines.append("") lines.append(f"Automated review for `{base}...{head}`") lines.append("") lines.append(f"- **Verdict:** {verdict}") lines.append(f"- **Changed files:** {len(changed_files)}") lines.append(f"- **Overall risk:** {overall_risk}") if diff_stat: lines.append(f"- **Diff stat:** `{diff_stat}`") if docs_only: lines.append("- **Docs-only PR** — no verification needed.") lines.append("") # -- Risk Map -- buckets: list[dict[str, Any]] = risk_map.get("buckets", []) if buckets: sorted_buckets = sorted( buckets, key=lambda b: (_SEVERITY_ORDER.get(b.get("risk", "P3"), 9), b.get("name", "")), ) lines.append("## Risk Map") lines.append("") lines.append("| Bucket | Risk | Files |") lines.append("|--------|------|------:|") for b in sorted_buckets: lines.append(f"| {b.get('name', '?')} | {b.get('risk', '?')} | {b.get('file_count', 0)} |") lines.append("") # -- Invariant Checks -- if invariant_results: lines.append("## Invariant Checks") lines.append("") for res in invariant_results: status = str(res.get("status", "SKIP")).upper() icon = _STATUS_ICON.get(status, "\u2753") check_id = res.get("id", "unknown") lines.append(f"- {icon} **{check_id}**: {status}") for detail in res.get("details", []): lines.append(f" - {detail}") for f in res.get("files", []): lines.append(f" - `{f}`") lines.append("") # -- Verification Plan -- if verification_plan: lines.append("## Verification Plan") lines.append("") for check in verification_plan: check_id = check.get("id", "?") desc = check.get("description", "") cmd = check.get("command", "") bucket_list = ", ".join(check.get("buckets", [])) label = f"{check_id}: {desc}" if desc else check_id line = f"- **{label}** — `{cmd}`" if bucket_list: line += f" (buckets: {bucket_list})" lines.append(line) lines.append("") # -- Verification Execution -- if verification_runs: lines.append("## Verification Execution") lines.append("") for run in verification_runs: status = str(run.get("status", "unknown")).upper() icon = _STATUS_ICON.get(status, "\u2753") cmd = run.get("command", "") run_id = run.get("id", "?") exit_code = run.get("exit_code") line = f"- {icon} **{run_id}**: {status}" if exit_code is not None: line += f" (exit {exit_code})" line += f" — `{cmd}`" lines.append(line) lines.append("") # -- Shards -- shard_list: list[dict[str, Any]] = shards.get("shards", []) if shard_list: lines.append("## Shards") lines.append("") lines.append(f"Total shards: {shards.get('total_shards', len(shard_list))}") lines.append("") for s in shard_list: name = s.get("name", "?") risk = s.get("risk", "?") file_count = len(s.get("files", [])) lines.append(f"- **{name}** ({risk}, {file_count} files): {s.get('description', '')}") lines.append("") # -- Artifacts -- lines.append("## Artifacts") lines.append("") lines.append("- `artifacts/review/risk_map.json`") lines.append("- `artifacts/review/shards.json`") lines.append("- `artifacts/review/verification_plan.json`") lines.append("- `artifacts/review/invariants.json`") lines.append("") return "\n".join(lines).rstrip() + "\n" def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Render markdown review comment from review payload JSON", ) parser.add_argument("--input", type=Path, required=True, help="Input review payload JSON") parser.add_argument("--out", type=Path, default=None, help="Optional output markdown path") return parser.parse_args() def main() -> None: args = _parse_args() payload = json.loads(args.input.read_text(encoding="utf-8")) markdown = build_markdown(payload) if args.out is not None: out_path = args.out.resolve() out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(markdown, encoding="utf-8") print(f"Wrote formatted comment to {out_path}") else: print(markdown) if __name__ == "__main__": main()