chore(llma): Add PostHog LLM Analytics instrumentation to PR approval agent

frontend/src/scenes/web-analytics/LiveMetricsDashboard/LiveAnimatedTable.tsx

@@ -110,7 +110,7 @@ export function LiveAnimatedTable<T>({
 
     useLayoutEffect(() => {
         prevPositionsRef.current = new Map(items.map((item, index) => [keyExtractor(item), index]))
-    }, [items])
+    }, [items, keyExtractor])
 
     return (
         <div className={clsx('bg-bg-light rounded-lg border p-4', className)}>

tools/pr-approval-agent/README.md

@@ -119,12 +119,32 @@ Every run produces a JSON evidence bundle (`--output-json` locally, uploaded as
 
 The GitHub Action uploads this as a build artifact with 30-day retention.
 
+## LLM Analytics
+
+Every run automatically sends telemetry to PostHog's internal LLM Analytics
+dashboard using the standard internal project API key (`sTMFPsFhdP1Ssg`).
+No extra secrets are needed.
+
+To disable, set `OPT_OUT_CAPTURE=1`.
+To override the destination, set `STAMPHOG_POSTHOG_API_KEY` and/or `STAMPHOG_POSTHOG_HOST`.
+
+Events captured:
+
+| Event            | When                   | Key properties                              |
+| ---------------- | ---------------------- | ------------------------------------------- |
+| `$ai_generation` | Each LLM reviewer call | model, tokens, cost, latency, cache metrics |
+| `$ai_trace`      | Pipeline completion    | total cost, verdict, tier, gate results     |
+
+All events include `stamphog_*` custom properties (PR number, author, tier, verdict)
+for filtering in the dashboard.
+
 ## Architecture
 
 - `review_pr.py` — pipeline orchestrator (fetch → classify → gates → LLM)
 - `gates.py` — deterministic classification and deny-list logic
 - `github.py` — GitHub data fetching via `gh` CLI
 - `reviewer.py` — Claude Agent SDK reviewer (showstoppers prompt)
+- `analytics.py` — PostHog LLM Analytics instrumentation
 - `.github/workflows/pr-approval-agent.yml` — GitHub Action (label trigger)
 
 ## Empirical basis

tools/pr-approval-agent/analytics.py

@@ -0,0 +1,202 @@
+"""PostHog LLM Analytics instrumentation for the PR approval agent.
+
+Captures $ai_generation and $ai_trace events so stamphog runs
+are visible in the LLM Analytics dashboard.
+"""
+
+import os
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+
+from posthoganalytics import Posthog
+
+_INTERNAL_PROJECT_API_KEY = "sTMFPsFhdP1Ssg"
+_INTERNAL_HOST = "https://us.i.posthog.com"
+DISTINCT_ID = "stamphog-ci-bot"
+
+
+def create_client() -> Posthog | None:
+    """Create a PostHog client for LLM analytics.
+
+    Uses the internal PostHog project key by default so no extra secrets
+    are needed.  Set OPT_OUT_CAPTURE=1 to disable.
+    """
+    if os.environ.get("OPT_OUT_CAPTURE", "").lower() in ("true", "yes", "1"):
+        return None
+    api_key = os.environ.get("STAMPHOG_POSTHOG_API_KEY", _INTERNAL_PROJECT_API_KEY)
+    host = os.environ.get("STAMPHOG_POSTHOG_HOST", _INTERNAL_HOST)
+    return Posthog(api_key, host=host)
+
+
+@dataclass
+class TraceRecorder:
+    """Records LLM analytics events for a single pipeline run.
+
+    Collects $ai_generation events from reviewer calls and emits
+    a $ai_trace event when the pipeline completes.
+    """
+
+    client: Posthog | None
+    trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    _start_time: float = field(default_factory=time.monotonic)
+    _generations: list[dict[str, Any]] = field(default_factory=list)
+    _pr_metadata: dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def enabled(self) -> bool:
+        return self.client is not None
+
+    def set_pr_metadata(
+        self,
+        pr_number: int,
+        repo: str,
+        author: str,
+        title: str,
+        tier: str,
+        t1_subclass: str,
+        lines_total: int,
+        files_changed: int,
+    ) -> None:
+        self._pr_metadata = {
+            "stamphog_pr_number": pr_number,
+            "stamphog_repo": repo,
+            "stamphog_author": author,
+            "stamphog_title": title[:200],
+            "stamphog_tier": tier,
+            "stamphog_t1_subclass": t1_subclass,
+            "stamphog_lines_total": lines_total,
+            "stamphog_files_changed": files_changed,
+        }
+
+    def record_generation(
+        self,
+        *,
+        model: str,
+        input_messages: list[dict[str, str]],
+        output_text: str,
+        usage: dict[str, Any] | None = None,
+        model_usage: dict[str, Any] | None = None,
+        duration_ms: int = 0,
+        total_cost_usd: float | None = None,
+        num_turns: int = 0,
+        stop_reason: str | None = None,
+    ) -> None:
+        """Record a single LLM generation (reviewer call)."""
+        if not self.enabled:
+            return
+
+        input_tokens = 0
+        output_tokens = 0
+        cache_read_tokens = 0
+        cache_creation_tokens = 0
+
+        if usage:
+            input_tokens = usage.get("input_tokens", 0)
+            output_tokens = usage.get("output_tokens", 0)
+            cache_read_tokens = usage.get("cache_read_input_tokens", 0)
+            cache_creation_tokens = usage.get("cache_creation_input_tokens", 0)
+
+        generation = {
+            "input_tokens": input_tokens,
+            "output_tokens": output_tokens,
+            "duration_ms": duration_ms,
+            "total_cost_usd": total_cost_usd,
+        }
+        self._generations.append(generation)
+
+        properties: dict[str, Any] = {
+            "$ai_trace_id": self.trace_id,
+            "$ai_model": model,
+            "$ai_provider": "anthropic",
+            "$ai_input": input_messages,
+            "$ai_input_tokens": input_tokens,
+            "$ai_output_choices": [{"role": "assistant", "content": output_text[:5000]}],
+            "$ai_output_tokens": output_tokens,
+            "$ai_latency": duration_ms / 1000.0,
+            "$ai_is_error": False,
+            "$ai_stream": True,
+            **self._pr_metadata,
+            "stamphog_num_turns": num_turns,
+            "stamphog_stop_reason": stop_reason or "",
+        }
+
+        if total_cost_usd is not None:
+            properties["$ai_total_cost_usd"] = total_cost_usd
+
+        if cache_read_tokens:
+            properties["$ai_cache_read_input_tokens"] = cache_read_tokens
+        if cache_creation_tokens:
+            properties["$ai_cache_creation_input_tokens"] = cache_creation_tokens
+
+        if model_usage:
+            for model_name, model_stats in model_usage.items():
+                properties[f"stamphog_model_{model_name}_input_tokens"] = model_stats.get("input_tokens", 0)
+                properties[f"stamphog_model_{model_name}_output_tokens"] = model_stats.get("output_tokens", 0)
+
+        try:
+            self.client.capture(
+                event="$ai_generation",
+                distinct_id=DISTINCT_ID,
+                properties=properties,
+            )
+        except Exception:
+            pass
+
+    def record_trace(
+        self,
+        *,
+        verdict: str,
+        gate_verdict: str,
+        gate_results: list[dict[str, Any]],
+        reviewer_output: dict[str, Any] | None = None,
+    ) -> None:
+        """Record the overall pipeline trace."""
+        if not self.enabled:
+            return
+
+        total_latency = time.monotonic() - self._start_time
+        total_input_tokens = sum(g["input_tokens"] for g in self._generations)
+        total_output_tokens = sum(g["output_tokens"] for g in self._generations)
+        total_cost = sum(g["total_cost_usd"] for g in self._generations if g["total_cost_usd"] is not None)
+
+        properties: dict[str, Any] = {
+            "$ai_trace_id": self.trace_id,
+            "$ai_latency": total_latency,
+            "$ai_input_tokens": total_input_tokens,
+            "$ai_output_tokens": total_output_tokens,
+            "$ai_input_state": {
+                "gate_verdict": gate_verdict,
+                "gates": gate_results,
+                **self._pr_metadata,
+            },
+            "$ai_output_state": {
+                "final_verdict": verdict,
+                "reviewer": reviewer_output,
+            },
+            **self._pr_metadata,
+            "stamphog_final_verdict": verdict,
+            "stamphog_gate_verdict": gate_verdict,
+            "stamphog_generation_count": len(self._generations),
+        }
+
+        if total_cost > 0:
+            properties["$ai_total_cost_usd"] = total_cost
+
+        try:
+            self.client.capture(
+                event="$ai_trace",
+                distinct_id=DISTINCT_ID,
+                properties=properties,
+            )
+        except Exception:
+            pass
+
+    def flush(self) -> None:
+        """Ensure all events are sent before the process exits."""
+        if self.enabled:
+            try:
+                self.client.flush()
+            except Exception:
+                pass

tools/pr-approval-agent/review_pr.py

@@ -4,6 +4,7 @@
 # dependencies = [
 #     "claude-agent-sdk",
 #     "anthropic",
+#     "posthoganalytics",
 # ]
 # ///
 # ruff: noqa: T201
@@ -25,6 +26,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 
+from analytics import TraceRecorder, create_client
 from gates import (
     MAX_FILES,
     MAX_LINES,
@@ -109,6 +111,7 @@ def __init__(self, pr_number: int, repo: str, *, dry_run: bool = False, verbose:
         self.gate_results: list[GateResult] = []
         self.reviewer_output: dict | None = None
         self.final_verdict: str = ""
+        self.trace = TraceRecorder(client=create_client())
 
     def run(self) -> str:
         """Run the full pipeline, return final verdict string."""
@@ -118,13 +121,37 @@ def run(self) -> str:
 
         gate_verdict = self._gate_verdict()
 
+        # Populate trace metadata after classification
+        self.trace.set_pr_metadata(
+            pr_number=self.pr_number,
+            repo=self.repo,
+            author=self.pr.author,
+            title=self.pr.title,
+            tier=self.classification["tier"],
+            t1_subclass=self.classification.get("t1_subclass", ""),
+            lines_total=self.pr.lines_total,
+            files_changed=len(self.pr.files),
+        )
+
         if self.dry_run:
             self.final_verdict = "DRY-RUN"
+            self._emit_trace(gate_verdict)
             return self.final_verdict
 
         self._llm_review(gate_verdict)
+        self._emit_trace(gate_verdict)
         return self.final_verdict
 
+    def _emit_trace(self, gate_verdict: str) -> None:
+        """Record the trace-level event and flush analytics."""
+        self.trace.record_trace(
+            verdict=self.final_verdict,
+            gate_verdict=gate_verdict,
+            gate_results=[{"gate": g.gate, "passed": g.passed, "message": g.message} for g in self.gate_results],
+            reviewer_output=self.reviewer_output,
+        )
+        self.trace.flush()
+
     def _gate_verdict(self) -> str:
         """Determine what gates say — this is authoritative."""
         if self._any_gate_denied():
@@ -295,7 +322,7 @@ def _check_tier(self) -> tuple[bool, str]:
 
     def _llm_review(self, gate_verdict: str) -> None:
         print(f"\n{_bold('LLM Review')}")
-        reviewer = Reviewer(REPO_ROOT, verbose=self.verbose)
+        reviewer = Reviewer(REPO_ROOT, verbose=self.verbose, trace=self.trace)
 
         gate_context = {
             "gate_verdict": gate_verdict,

tools/pr-approval-agent/reviewer.py

@@ -12,6 +12,7 @@
 import subprocess
 from pathlib import Path
 
+from analytics import TraceRecorder
 from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
 from claude_agent_sdk.types import AssistantMessage, ToolUseBlock
 from github import PRData
@@ -183,9 +184,10 @@ def _validate_verdict(result: dict) -> dict:
 class Reviewer:
     """LLM reviewer using Agent SDK."""
 
-    def __init__(self, repo_root: Path, *, verbose: bool = False):
+    def __init__(self, repo_root: Path, *, verbose: bool = False, trace: TraceRecorder | None = None):
         self.repo_root = repo_root
         self.verbose = verbose
+        self.trace = trace
 
     def review(self, pr: PRData, classification: dict, gate_context: dict) -> dict:
         """Claude explores the repo and produces a verdict."""
@@ -213,10 +215,12 @@ async def _review(self, pr: PRData, classification: dict, gate_context: dict) ->
         )
 
         structured_output = None
+        result_message: ResultMessage | None = None
         async for message in query(prompt=prompt, options=options):
             if self.verbose:
                 print(f"\033[2m    [{type(message).__name__}]\033[0m", flush=True)
             if isinstance(message, ResultMessage):
+                result_message = message
                 if message.subtype == "error_max_structured_output_retries":
                     raise RuntimeError("Agent could not produce valid structured output after retries")
                 if message.structured_output:
@@ -230,7 +234,23 @@ async def _review(self, pr: PRData, classification: dict, gate_context: dict) ->
 
         if structured_output is None:
             raise RuntimeError("Reviewer agent returned no structured output")
-        return _validate_verdict(structured_output)
+
+        verdict = _validate_verdict(structured_output)
+
+        if self.trace and result_message:
+            self.trace.record_generation(
+                model=MODEL,
+                input_messages=[{"role": "user", "content": prompt[:2000]}],
+                output_text=verdict.get("reasoning", ""),
+                usage=result_message.usage,
+                model_usage=result_message.model_usage,
+                duration_ms=result_message.duration_ms,
+                total_cost_usd=result_message.total_cost_usd,
+                num_turns=result_message.num_turns,
+                stop_reason=result_message.stop_reason,
+            )
+
+        return verdict
 
     def _log_tool_call(self, block: ToolUseBlock) -> None:
         name = block.name