diff --git a/frontend/src/scenes/web-analytics/LiveMetricsDashboard/LiveAnimatedTable.tsx b/frontend/src/scenes/web-analytics/LiveMetricsDashboard/LiveAnimatedTable.tsx index 4eb006616ca1..9c192dcac5fa 100644 --- a/frontend/src/scenes/web-analytics/LiveMetricsDashboard/LiveAnimatedTable.tsx +++ b/frontend/src/scenes/web-analytics/LiveMetricsDashboard/LiveAnimatedTable.tsx @@ -110,7 +110,7 @@ export function LiveAnimatedTable({ useLayoutEffect(() => { prevPositionsRef.current = new Map(items.map((item, index) => [keyExtractor(item), index])) - }, [items]) + }, [items, keyExtractor]) return (
diff --git a/tools/pr-approval-agent/README.md b/tools/pr-approval-agent/README.md index 658439f4d0ea..29105a7abf77 100644 --- a/tools/pr-approval-agent/README.md +++ b/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 diff --git a/tools/pr-approval-agent/analytics.py b/tools/pr-approval-agent/analytics.py new file mode 100644 index 000000000000..ee6dc930e9a3 --- /dev/null +++ b/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 diff --git a/tools/pr-approval-agent/review_pr.py b/tools/pr-approval-agent/review_pr.py index d0ff6a28f43b..0d1182a2b0cf 100644 --- a/tools/pr-approval-agent/review_pr.py +++ b/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, diff --git a/tools/pr-approval-agent/reviewer.py b/tools/pr-approval-agent/reviewer.py index 865d0b715300..0d8efc3908c2 100644 --- a/tools/pr-approval-agent/reviewer.py +++ b/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