feat(guardian): add review quality metrics tracking

.github/workflows/guardian.yml

@@ -5,7 +5,7 @@ on:
     types: [created]
 
 permissions:
-  contents: read
+  contents: write
   pull-requests: write
   issues: read
 
@@ -60,13 +60,39 @@ jobs:
       - name: Build code graph
         run: uv run cgis ingest ./src --output graph.db
 
+      - name: Restore metrics from data branch
+        run: |
+          git fetch origin data/guardian-metrics 2>/dev/null && \
+            git show origin/data/guardian-metrics:guardian_metrics.jsonl \
+              > guardian_metrics.jsonl 2>/dev/null || true
+
       - name: Run Guardian Review
         env:
           GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
           MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
           GUARDIAN_PROVIDER: ${{ vars.GUARDIAN_PROVIDER }}
           GUARDIAN_MODEL: ${{ vars.GUARDIAN_MODEL }}
-        run: uv run python scripts/guardian_review.py --output guardian_report.md --db graph.db
+        run: |
+          uv run python scripts/guardian_review.py \
+            --output guardian_report.md \
+            --db graph.db \
+            --pr "$PR_NUMBER" \
+            --metrics guardian_metrics.jsonl
+
+      - name: Push metrics to data branch
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          if git ls-remote --exit-code origin data/guardian-metrics > /dev/null 2>&1; then
+            git worktree add /tmp/metrics origin/data/guardian-metrics
+          else
+            git worktree add --orphan -b data/guardian-metrics /tmp/metrics
+          fi
+          cp guardian_metrics.jsonl /tmp/metrics/
+          git -C /tmp/metrics add guardian_metrics.jsonl
+          git -C /tmp/metrics diff --cached --quiet || \
+            git -C /tmp/metrics commit -m "metrics: guardian review PR #${PR_NUMBER} [skip ci]"
+          git -C /tmp/metrics push origin HEAD:data/guardian-metrics
 
       - name: Post review as PR comment
         uses: peter-evans/create-or-update-comment@a111a2c3bacd7be7898ee22d0d71d9aec2bb972c  # v4

.gitignore

@@ -218,3 +218,4 @@ __marimo__/
 .streamlit/secrets.toml
 graph.json
 *.db
+guardian_metrics.jsonl

scripts/guardian_review.py

@@ -9,6 +9,7 @@
 
 from cgis.guardian.collector import ContextCollector
 from cgis.guardian.core import GuardianReviewer
+from cgis.guardian.metrics import record_review
 from cgis.guardian.providers.base import BaseProvider
 from cgis.guardian.providers.gemini import GeminiProvider
 from cgis.guardian.providers.mistral import MistralProvider
@@ -56,6 +57,18 @@ async def main() -> None:
         default=None,
         help="Path to graph.db for structural impact context (optional).",
     )
+    parser.add_argument(
+        "--pr",
+        type=int,
+        default=None,
+        help="GitHub PR number being reviewed (used for metrics tracking).",
+    )
+    parser.add_argument(
+        "--metrics",
+        type=Path,
+        default=Path("guardian_metrics.jsonl"),
+        help="Path to the append-only metrics log (default: guardian_metrics.jsonl).",
+    )
     args = parser.parse_args()
 
     provider, model = _build_provider()
@@ -97,6 +110,16 @@ async def main() -> None:
         footer_parts.append(f"graph {stats['with_graph']}/{stats['total']} files ({pct}%)")
     footer = "\n\n---\n> " + " · ".join(footer_parts)
 
+    metrics_path = record_review(
+        model=model,
+        pr=args.pr,
+        prompt_tokens=usage.prompt_tokens,
+        completion_tokens=usage.completion_tokens,
+        review_text=review_result,
+        metrics_path=args.metrics,
+    )
+    log.info("Metrics recorded.", path=str(metrics_path))
+
     if args.output:
         safe_root = Path.cwd().resolve()
         output_path = (safe_root / args.output).resolve()

src/cgis/cli.py

@@ -13,6 +13,7 @@
 from cgis.core.models import VIRTUAL_FILE_PATH, Edge, EdgeType, Node, NodeNamespace
 from cgis.extractors.python_extractor import PythonExtractor, file_path_to_module_fqn
 from cgis.extractors.typescript_extractor import TypeScriptExtractor
+from cgis.guardian.metrics import load_reviews, rate_review
 from cgis.pipeline import IngestionPipeline
 from cgis.query.analyzer import AnalyzerEngine
 from cgis.query.anomaly import AnomalyType, ArchitecturalAnomaly
@@ -665,5 +666,82 @@ def analyze(
             console.print(f"  [italic]💡 {a.refactoring_hint}[/italic]\n")
 
 
+_DEFAULT_METRICS = "guardian_metrics.jsonl"
+
+
+@app.command()
+def guardian_rate(
+    pr: int = typer.Argument(..., help="GitHub PR number to rate."),
+    applied: int = typer.Argument(..., help="Number of findings actually applied."),
+    metrics: str = typer.Option(_DEFAULT_METRICS, "--metrics", "-m", help="Path to metrics file."),
+) -> None:
+    """Record how many Guardian findings were applied for a given PR."""
+    updated = rate_review(pr=pr, applied=applied, metrics_path=Path(metrics))
+    if updated:
+        console.print(f"[green]✅ PR #{pr}: recorded {applied} applied findings.[/green]")
+    else:
+        console.print(f"[red]❌ No unrated entry found for PR #{pr} in {metrics}.[/red]")
+        raise typer.Exit(code=1)
+
+
+@app.command()
+def guardian_stats(
+    metrics: str = typer.Option(_DEFAULT_METRICS, "--metrics", "-m", help="Path to metrics file."),
+    last: int = typer.Option(20, "--last", "-n", help="Show only the last N reviews."),
+) -> None:
+    """Show Guardian review quality metrics trend."""
+    reviews = load_reviews(Path(metrics))
+    if not reviews:
+        console.print(f"[yellow]No metrics found in {metrics}.[/yellow]")
+        raise typer.Exit
+
+    reviews = reviews[-last:]
+
+    table = Table(title=f"Guardian Review Metrics (last {len(reviews)})")
+    table.add_column("PR", style="cyan", justify="right")
+    table.add_column("Model", style="dim")
+    table.add_column("Tokens", justify="right")
+    table.add_column("Findings", justify="right")
+    table.add_column("Applied", justify="right")
+    table.add_column("Precision", justify="right")
+    table.add_column("LGTM")
+
+    total_tokens = 0
+    rated = [r for r in reviews if r.get("findings_applied") is not None]
+
+    for r in reviews:
+        pr_str = f"#{r['pr']}" if r.get("pr") else "—"
+        tokens = int(r.get("total_tokens", 0))
+        total_tokens += tokens
+        findings = int(r.get("findings_total", 0))
+        applied = r.get("findings_applied")
+        if applied is not None:
+            applied_str = str(applied)
+            precision = f"{int(applied) / findings * 100:.0f}%" if findings else "—"
+        else:
+            applied_str = "[dim]?[/dim]"
+            precision = "[dim]?[/dim]"
+        lgtm = "✅" if r.get("lgtm") else ""
+        table.add_row(
+            pr_str,
+            str(r.get("model", "")),
+            f"{tokens:,}",
+            str(findings),
+            applied_str,
+            precision,
+            lgtm,
+        )
+
+    console.print(table)
+
+    if rated:
+        total_findings = sum(int(r["findings_total"]) for r in rated)
+        total_applied = sum(int(r["findings_applied"]) for r in rated)
+        avg_precision = f"{total_applied / total_findings * 100:.0f}%" if total_findings else "—"
+        console.print(f"\n  Avg tokens/review : [cyan]{total_tokens // len(reviews):,}[/cyan]")
+        rated_label = f"rated {len(rated)}/{len(reviews)} reviews"
+        console.print(f"  Overall precision : [cyan]{avg_precision}[/cyan]  ({rated_label})")
+
+
 if __name__ == "__main__":  # pragma: no cover
     app()

src/cgis/guardian/metrics.py

@@ -0,0 +1,91 @@
+"""Guardian review quality metrics: append-only JSONL log with precision tracking."""
+
+import json
+import re
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+_DEFAULT_METRICS_FILE = Path("guardian_metrics.jsonl")
+
+
+def _count_findings(review_text: str) -> tuple[int, bool]:
+    """Parse finding count and LGTM flag from review text.
+
+    Returns (findings_total, lgtm).
+    Counts headings that match the output format: lines starting with '**['.
+    """
+    findings = len(
+        re.findall(
+            r"^\*\*\[(?:Logic Bug|Test Coverage|Type Safety|Ontology)", review_text, re.MULTILINE
+        )
+    )
+    lgtm = findings == 0 and "lgtm" in review_text.lower()
+    return findings, lgtm
+
+
+def record_review(
+    *,
+    model: str,
+    pr: int | None,
+    prompt_tokens: int,
+    completion_tokens: int,
+    review_text: str,
+    metrics_path: Path = _DEFAULT_METRICS_FILE,
+) -> Path:
+    """Append one review entry to the metrics JSONL file and return the path.
+
+    The file is created if it does not exist. Each line is a self-contained
+    JSON object so the file can be streamed line-by-line without loading it all.
+    """
+    findings_total, lgtm = _count_findings(review_text)
+    entry = {
+        "timestamp": datetime.now(UTC).isoformat(),
+        "pr": pr,
+        "model": model,
+        "prompt_tokens": prompt_tokens,
+        "completion_tokens": completion_tokens,
+        "total_tokens": prompt_tokens + completion_tokens,
+        "findings_total": findings_total,
+        "findings_applied": None,
+        "lgtm": lgtm,
+    }
+    with metrics_path.open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(entry) + "\n")
+    return metrics_path
+
+
+def rate_review(pr: int, applied: int, metrics_path: Path = _DEFAULT_METRICS_FILE) -> bool:
+    """Set findings_applied for the most recent entry matching the given PR.
+
+    Returns True if an entry was updated, False if none found.
+    """
+    if not metrics_path.exists():
+        return False
+
+    lines = metrics_path.read_text(encoding="utf-8").splitlines()
+    updated = False
+    result: list[str] = []
+    for line in reversed(lines):
+        if not updated:
+            entry = json.loads(line)
+            if entry.get("pr") == pr and entry.get("findings_applied") is None:
+                entry["findings_applied"] = applied
+                line = json.dumps(entry)  # noqa: PLW2901
+                updated = True
+        result.append(line)
+
+    if updated:
+        metrics_path.write_text("\n".join(reversed(result)) + "\n", encoding="utf-8")
+    return updated
+
+
+def load_reviews(metrics_path: Path = _DEFAULT_METRICS_FILE) -> list[dict[str, Any]]:
+    """Return all review records from the metrics file, oldest first."""
+    if not metrics_path.exists():
+        return []
+    return [
+        json.loads(line)
+        for line in metrics_path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]