From 2f96e66aab6d40f9d7e8cb3c832aa9d6fde2d532 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 14:22:55 +0000 Subject: [PATCH] feat(pilot): behavioural scorers, feedback ground truth, and an eval dataset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the evaluation layer on top of the review traces: five deterministic scores describing how the reviewer behaved, a bridge that turns human reactions into ground truth, and a dataset seeded from the reviews already run. The two are kept apart on purpose. feedback.db has recorded 113 reviews and zero reactions, resolutions or replies — nobody has ever responded to a bot comment — so an accuracy metric cannot be built yet. The scorers therefore measure behaviour, which is computable from data in hand, and feedback_scores turns verdicts into scores the moment any arrive. eval_scores.py emits finding_rate, severity_info_ratio, severity_max, dropped_findings and cost_per_finding into the same ingestion batch as the trace. Undefined values are omitted rather than reported as zero: an info ratio over a silent review is undefined, and charting it as 0 would read as perfect calibration. dropped_findings needed a parser change. Both parsers silently discard findings with an unusable path/line, which made a model emitting garbage locations indistinguishable from one that found nothing. last_parse_dropped() exposes the delta, read at parse time — after apply_repo_config the drops are the config working as intended, not the model misbehaving. feedback_scores.py scores the session ("{repo}#{pr}"), because feedback arrives days later against a PR and nothing records which re-run produced which comment. review_acceptance is absent rather than 0 when nothing was engaged. eval_bootstrap.py registers the score configs, seeds the pragent-reviews dataset, and can backfill scores onto traces that predate the scorers. expectedOutput is the reviewer's own prior output, flagged labelled_by_human: false — a regression baseline, not verified truth. Also fixes a silent telemetry failure: the ingestion endpoint answers 207 when only some events succeed, so a batch with every event rejected still looked like success. Score events were missing the required per-event timestamp and ingested nothing while reporting 207. _warn_on_rejected_events now logs the per-event errors under LANGFUSE_DEBUG. Co-Authored-By: Claude Opus 5 --- pilot/README-eval.md | 123 ++++++++++++ pilot/ai_review.py | 29 ++- pilot/eval_bootstrap.py | 299 ++++++++++++++++++++++++++++ pilot/eval_scores.py | 233 ++++++++++++++++++++++ pilot/feedback_scores.py | 247 +++++++++++++++++++++++ pilot/langfuse_trace.py | 61 +++++- tests/pilot/test_eval_scores.py | 200 +++++++++++++++++++ tests/pilot/test_feedback_scores.py | 207 +++++++++++++++++++ tests/pilot/test_langfuse_trace.py | 91 ++++++++- tests/pilot/test_parse_dropped.py | 90 +++++++++ 10 files changed, 1573 insertions(+), 7 deletions(-) create mode 100644 pilot/README-eval.md create mode 100644 pilot/eval_bootstrap.py create mode 100644 pilot/eval_scores.py create mode 100644 pilot/feedback_scores.py create mode 100644 tests/pilot/test_eval_scores.py create mode 100644 tests/pilot/test_feedback_scores.py create mode 100644 tests/pilot/test_parse_dropped.py diff --git a/pilot/README-eval.md b/pilot/README-eval.md new file mode 100644 index 0000000..f17e23e --- /dev/null +++ b/pilot/README-eval.md @@ -0,0 +1,123 @@ +# Evaluation — scorers, ground truth, and the dataset + +Langfuse already receives one trace per review (`README-langfuse.md`). This is +the layer on top: numbers attached to those traces that say how the reviewer +*behaved*, and the beginnings of a ground-truth signal that says whether it was +*right*. + +Those two things are deliberately kept apart, because only one of them exists +yet. + +## What could and could not be built + +`feedback.db` has recorded 113 reviews across 4 repos. It has recorded **zero** +reactions, zero thread resolutions and zero replies. The harvester, the schema +and the daily analyzer are all working; nobody has ever reacted to a bot +comment. + +That rules out an accuracy metric today. Correctness needs labels, and a +judge scored against no labels is theatre. So the scorers here measure +behaviour, which is computable from data already in hand, and a separate +bridge exists to turn human reactions into scores the moment any arrive. + +## The five behavioural scores + +Emitted with every review by `eval_scores.py`, folded into the same ingestion +batch as the trace so they cost no extra request. + +| score | type | what a change in it means | +|---|---|---| +| `finding_rate` | NUMERIC | Findings posted. 0 is the restraint case — good on clean code, a failure when the run degraded. Only the rate over time separates those. | +| `severity_info_ratio` | NUMERIC 0–1 | Share of findings the model rated `info`/`trivial`. Rising = the model is hedging rather than committing. `None` when the review was silent: a ratio over an empty set is undefined, and charting it as 0 would read as perfect calibration. | +| `severity_max` | CATEGORICAL | Highest severity surfaced, `none` when silent. Categorical because "did this ever surface something serious" is the real question, and a mean of severity ranks answers nothing. | +| `dropped_findings` | NUMERIC | Findings the model emitted that the parser rejected for an unusable `path`/`line`. This is the only score here that measures the model's raw output. | +| `cost_per_finding` | NUMERIC | Equivalent USD per finding. A cheaper model that finds nothing is not cheaper. | + +### Why `dropped_findings` needed a change to the parser + +`parse_findings` and `parse_review_output` discard any finding with a missing or +unusable location. That happens silently, so a model emitting ten findings at +invalid locations was indistinguishable from a model that found nothing — both +produce an empty list. `ai_review.last_parse_dropped()` exposes the delta, +recorded at parse time. + +It must be read at parse time specifically: by the time findings reach +`_emit_langfuse`, `apply_repo_config` has already filtered them by +`severity_threshold` and `max_findings`, and those drops are the config working +as intended, not the model misbehaving. + +## Ground truth: `feedback_scores.py` + +Turns `feedback.db` into two session-level scores, keyed on `"{repo}#{pr}"` +(which is what `langfuse_trace` already sets as `sessionId`). + +| score | meaning | +|---|---| +| `review_engagement` | Share of a PR's findings that drew any human reaction, resolution or reply. **Watch this first** — every quality number is vapour until it moves off 0. | +| `review_acceptance` | Net verdict over engaged findings, −1 to +1. Absent, not 0, when nothing was engaged: zero would claim humans judged the review neutral, when the truth is nobody looked. | + +Session-level rather than trace-level because feedback arrives days later +against a PR, and nothing in `feedback.db` records which re-run of the reviewer +produced which comment. The session is both the available join and the honest +granularity. + +Score ids are `uuid5(namespace, repo#pr#name)`, so the daily backfill updates +rather than duplicates. + +## The dataset + +`pragent-reviews`, one item per PR the reviewer has run on, seeded by +`eval_bootstrap.py` from `feedback.db`. + +`expectedOutput` is **the reviewer's own prior output**, not human-verified +truth — every item carries `metadata.labelled_by_human: false`. Read it as a +regression baseline: re-run a candidate model over these PRs and the diff +against this column is the behaviour change. Promoting an item to real ground +truth means a human editing it in the dataset view after re-reading the PR. + +## Running it + +```bash +# once per project: score configs + dataset (+ score historical traces) +python3 pilot/eval_bootstrap.py --db /data/feedback.db --backfill-traces + +# ship feedback verdicts (runs daily from the feedback CronJob) +python3 pilot/feedback_scores.py --db /data/feedback.db +``` + +Both need `LANGFUSE_HOST`, `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY`. In +cluster they come from the `pragent-langfuse` Secret and point at the ClusterIP +— never the NodePort, whose oauth2-proxy 302s ingestion to Logto and drops it. + +## Gotcha: HTTP 207 is not success + +The ingestion endpoint answers `207 Multi-Status` when *some* events failed, so +a batch where **every** event was rejected still returns 207. An early version +of these scorers omitted the required per-event `timestamp` and silently +ingested nothing while reporting success. `langfuse_trace._warn_on_rejected_events` +now logs the per-event errors under `LANGFUSE_DEBUG=1`. If scores are missing, +check that before anything else. + +## What the first run showed + +Backfilled over 42 existing traces and 13 PRs: + +``` +cost_per_finding n=42 mean=0.3133 min=0.0880 max=0.9042 +finding_rate n=42 mean=0.4762 min=0.0000 max=4.0000 +severity_info_ratio n=14 mean=0.0000 +review_engagement n=14 mean=0.0000 +severity_max {none: 28, medium: 11, high: 1, critical: 2} +``` + +Two things worth keeping: + +- **The reviewer is not info-heavy.** `feedback.db` shows 61 of 62 findings at + `INFO`, which looked like a badly calibrated model. It is not: `severity_max` + reads `medium`/`high`/`critical` on every trace that found anything, and + `severity_info_ratio` is flat 0. The `INFO` in the DB comes from + `feedback_harvest._parse_severity`, which defaults to `INFO` when its regex + misses the severity badge in the rendered comment. The DB severity is a + re-parse artifact; the score reads the model's structured output directly. +- **28 of 42 reviews found nothing** (67%), and **engagement is flat zero**. The + first is not yet interpretable without the second. diff --git a/pilot/ai_review.py b/pilot/ai_review.py index c0d6289..02e5346 100644 --- a/pilot/ai_review.py +++ b/pilot/ai_review.py @@ -678,6 +678,19 @@ def _strip_path_prefix(p: str) -> str: # --------------------------------------------------------------------------- +# How many raw findings the last `parse_review_output` / `parse_findings` call +# rejected for an unusable path/line. A side channel rather than a return value +# because both parsers already return fixed-width tuples that several callers +# and their tests unpack positionally; widening them to carry a telemetry +# number would be a breaking change for a fail-open signal. +_LAST_PARSE_DROPPED: dict[str, int] = {"n": 0} + + +def last_parse_dropped() -> int: + """Findings the last parse discarded. Read it immediately after parsing.""" + return int(_LAST_PARSE_DROPPED.get("n") or 0) + + def _normalize_finding(f: dict) -> dict | None: """Validate + normalize one raw finding dict. Returns None if it's unusable (missing path/line). Normalises severity, keeps `reference` (default "").""" @@ -754,6 +767,7 @@ def parse_findings(text: str) -> list[dict]: Also accepts a bare JSON array as the outer value: ``[{...}, {...}]`` — some agents skip the ``{"summary":..., "findings":[...]}`` wrapper. """ + _LAST_PARSE_DROPPED["n"] = 0 data = _parse_json_tolerant(text) if isinstance(data, dict): findings = data.get("findings") @@ -768,6 +782,7 @@ def parse_findings(text: str) -> list[dict]: n = _normalize_finding(f) if n is not None: out.append(n) + _LAST_PARSE_DROPPED["n"] = len(findings) - len(out) return out @@ -823,6 +838,7 @@ def parse_review_output( block), with a tolerant fallback that scans for the last balanced object/array in the prose tail. Never raises. """ + _LAST_PARSE_DROPPED["n"] = 0 blob = _last_json_block(text) if blob is None: return "", [], [], [], [], "", "" @@ -856,6 +872,12 @@ def parse_review_output( n = _normalize_finding(f) if n is not None: out.append(n) + # A model that emits findings at unusable locations is indistinguishable + # from one that found nothing, because both end up with an empty `out`. + # Stash the delta so the caller can score it (see `eval_scores`). + _LAST_PARSE_DROPPED["n"] = len(findings_raw) - len(out) + else: + _LAST_PARSE_DROPPED["n"] = 0 return summary, out, summary_changes, risks, walkthrough, risk_verdict, test_coverage @@ -2080,6 +2102,7 @@ def _emit_langfuse( summary: str, engine: str, config: dict | None = None, + dropped_count: float | None = None, ) -> None: """Ship this review's usage to Langfuse, if one is configured. @@ -2105,7 +2128,7 @@ def _emit_langfuse( repo=repo, index=index, sha=sha, title=title, model=model, usage=usage, findings=findings, summary=summary or "", engine=engine, lenses=(usage or {}).get("lenses"), - price_target=price_target, + price_target=price_target, dropped_count=dropped_count, ) except Exception as e: print(f"pragent: langfuse emit skipped: {e}", file=sys.stderr) @@ -2235,6 +2258,7 @@ def review_pr( additional_context=additional_context, ) review_summary, findings, summary_changes, risks, _walkthrough, _risk_verdict, _test_coverage = parse_review_output(stdout) + parse_dropped = last_parse_dropped() if not findings and not review_summary: # The findings JSON was missing or malformed. Don't discard the # run: salvage the prose, keep the usage report (the tokens were @@ -2255,12 +2279,14 @@ def review_pr( repo=repo, index=index, sha=sha, title=title, model=display_model, usage=usage, findings=[], summary=salvaged, engine=engine, config=config, + dropped_count=parse_dropped, ) return True else: user_prompt = build_user_prompt(title, body + compression_note, diff, config, prior, additional_context) raw_findings = call_model(ollama_url, model, SYSTEM_PROMPT, user_prompt, max_tokens) findings = parse_findings(raw_findings) + parse_dropped = last_parse_dropped() usage = None # Filter / cap findings per `.pr-review.json` (style, threshold, max, @@ -2343,6 +2369,7 @@ def review_pr( repo=repo, index=index, sha=sha, title=title, model=display_model, usage=usage, findings=findings, summary=review_summary, engine=engine, config=config, + dropped_count=parse_dropped, ) print( f"pragent: reviewed {repo}#{index} sha={sha[:8]} " diff --git a/pilot/eval_bootstrap.py b/pilot/eval_bootstrap.py new file mode 100644 index 0000000..be831d0 --- /dev/null +++ b/pilot/eval_bootstrap.py @@ -0,0 +1,299 @@ +#!/usr/bin/env python3 +"""pragent pilot — one-time Langfuse project setup for evaluation. + +Three jobs, each idempotent so it can be re-run after any change: + + 1. **Score configs.** Registers the schema for every score pragent emits + (`eval_scores.SCORE_CONFIGS` + `feedback_scores.SCORE_CONFIGS`). Without + these the scores still ingest, but nothing stops a later scorer writing + `severity_max="HIGH"` beside today's `"high"` and quietly splitting one + series into two. Configs are immutable in Langfuse — a name that already + exists is left alone rather than updated. + + 2. **Dataset.** Seeds `pragent-reviews` from `feedback.db`: one item per PR + the reviewer has actually run on, carrying the repo/PR/sha as input and + the findings it posted as `expectedOutput`. + + Read `expectedOutput` here as "what the reviewer said last time", not "what + is correct" — no human has labelled any of it. It is a regression baseline: + re-run a candidate model over these PRs and the diff against this column is + the behaviour change. Promoting an item to real ground truth means a human + editing it after reviewing the PR, which is what the dataset view is for. + + 3. **Trace backfill** (`--backfill-traces`). Scores only ride along with new + reviews, so without this the charts stay empty until the next PR lands. + Every trace `langfuse_trace` has ever written already carries the finding + count, the severity histogram and the cost in its metadata, which is + everything four of the five scorers need. `dropped_findings` is absent from + historical traces and is left unscored rather than backfilled as zero. + + 4. **Reports** what it found, so the gap between "reviews recorded" and + "reviews with human feedback" is visible rather than assumed. + +Usage: + LANGFUSE_HOST=... LANGFUSE_PUBLIC_KEY=... LANGFUSE_SECRET_KEY=... \\ + python3 eval_bootstrap.py --db /data/feedback.db +""" +from __future__ import annotations + +import argparse +import base64 +import json +import os +import sqlite3 +import sys +import urllib.error +import urllib.request + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import eval_scores # noqa: E402 +import feedback_scores # noqa: E402 + +DATASET_NAME = "pragent-reviews" + + +def _conf() -> tuple[str, str, str]: + host = (os.environ.get("LANGFUSE_HOST") or "").strip().rstrip("/") + pk = (os.environ.get("LANGFUSE_PUBLIC_KEY") or "").strip() + sk = (os.environ.get("LANGFUSE_SECRET_KEY") or "").strip() + if not host or not pk or not sk: + raise SystemExit("LANGFUSE_HOST / LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY must be set") + return host, pk, sk + + +def _call(method: str, path: str, body: dict | None = None, timeout: float = 20.0): + host, pk, sk = _conf() + auth = base64.b64encode(f"{pk}:{sk}".encode()).decode("ascii") + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request( + host + path, + data=data, + headers={ + "Content-Type": "application/json", + "Authorization": f"Basic {auth}", + "User-Agent": "pragent-pilot/1.0", + }, + method=method, + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read() + return resp.status, (json.loads(raw) if raw else None) + except urllib.error.HTTPError as e: + return e.code, e.read()[:400].decode("utf-8", "replace") + + +# --------------------------------------------------------------------------- +# 1. Score configs +# --------------------------------------------------------------------------- + +def ensure_score_configs() -> dict: + status, existing = _call("GET", "/api/public/score-configs?limit=100") + have = set() + if status == 200 and isinstance(existing, dict): + have = {c.get("name") for c in existing.get("data", [])} + + created, skipped, failed = [], [], [] + for cfg in list(eval_scores.SCORE_CONFIGS) + list(feedback_scores.SCORE_CONFIGS): + if cfg["name"] in have: + skipped.append(cfg["name"]) + continue + st, resp = _call("POST", "/api/public/score-configs", cfg) + if st in (200, 201): + created.append(cfg["name"]) + else: + failed.append({"name": cfg["name"], "status": st, "error": resp}) + return {"created": created, "already_present": skipped, "failed": failed} + + +# --------------------------------------------------------------------------- +# 2. Dataset from recorded reviews +# --------------------------------------------------------------------------- + +def read_review_items(db_path: str) -> list[dict]: + """One dataset item per (repo, pr) the reviewer has run on. + + Keyed on the PR rather than on each individual review row: the same PR is + re-reviewed on every push, and 113 rows over 26 PRs would make a benchmark + that is 4x redundant and weighted towards whichever PR churned most. + """ + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + prs = conn.execute( + """ + SELECT repo, pr, MAX(posted_at) AS last_seen, COUNT(*) AS reviews, + MAX(head_sha) AS head_sha + FROM review GROUP BY repo, pr ORDER BY repo, pr + """ + ).fetchall() + items = [] + for row in prs: + findings = conn.execute( + """ + SELECT path, line, severity, problem, fix + FROM inline_finding WHERE repo = ? AND pr = ? + ORDER BY path, line + """, + (row["repo"], row["pr"]), + ).fetchall() + items.append( + { + "id": f'{row["repo"]}#{row["pr"]}', + "input": { + "repo": row["repo"], + "pr": int(row["pr"]), + "head_sha": row["head_sha"], + }, + "expectedOutput": { + "findings": [dict(f) for f in findings], + "finding_count": len(findings), + }, + "metadata": { + "reviews_run": int(row["reviews"]), + "last_reviewed_at": int(row["last_seen"]), + # Flags that this row is the reviewer's own past output, + # not a human judgement. Filter on it before anyone + # treats the dataset as ground truth. + "labelled_by_human": False, + }, + } + ) + return items + finally: + conn.close() + + +def ensure_dataset(items: list[dict], name: str = DATASET_NAME) -> dict: + st, _ = _call( + "POST", + "/api/public/datasets", + { + "name": name, + "description": ( + "PRs the pragent pilot has reviewed, seeded from feedback.db. " + "expectedOutput is the reviewer's own prior output — a regression " + "baseline, not human-verified ground truth." + ), + "metadata": {"source": "feedback.db", "seeded_by": "eval_bootstrap.py"}, + }, + ) + # A duplicate name is fine: the dataset already exists from an earlier run. + dataset_ok = st in (200, 201, 409) + + created, failed = 0, [] + for item in items: + body = { + "datasetName": name, + "id": item["id"], # idempotent: same PR updates rather than duplicates + "input": item["input"], + "expectedOutput": item["expectedOutput"], + "metadata": item["metadata"], + } + ist, resp = _call("POST", "/api/public/dataset-items", body) + if ist in (200, 201): + created += 1 + else: + failed.append({"item": item["id"], "status": ist, "error": resp}) + return {"dataset": name, "dataset_created": dataset_ok, "items_upserted": created, "failed": failed} + + +# --------------------------------------------------------------------------- +# 3. Backfill scores onto traces that predate the scorers +# --------------------------------------------------------------------------- + +def _synth_findings(severities: dict) -> list[dict]: + """Rebuild a findings list from a trace's severity histogram. + + Only severity matters to the scorers, and that is all the histogram kept. + Reconstructing placeholders is honest here because every scorer being + backfilled reads nothing else off a finding. + """ + out = [] + for sev, count in (severities or {}).items(): + out.extend({"severity": sev} for _ in range(int(count))) + return out + + +def backfill_traces(limit_pages: int = 20) -> dict: + import eval_scores as es + + scored, skipped, events = 0, 0, [] + page = 1 + while page <= limit_pages: + st, resp = _call("GET", f"/api/public/traces?limit=50&page={page}&name=pr-review") + if st != 200 or not isinstance(resp, dict): + break + rows = resp.get("data") or [] + if not rows: + break + for tr in rows: + meta = tr.get("metadata") or {} + severities = meta.get("severities") or {} + count = meta.get("findings") + if count is None: + skipped += 1 + continue + findings = _synth_findings(severities) + # The histogram is authoritative when present; a trace that recorded + # a count but no histogram still scores its rate. + if not findings and count: + findings = [{"severity": "medium"} for _ in range(int(count))] + batch = es.build_scores( + trace_id=tr["id"], + findings=findings, + environment=tr.get("environment") or "default", + cost_usd=(tr.get("totalCost") or meta.get("provider_cost_usd")), + timestamp=tr.get("timestamp"), + comment="backfilled from trace metadata", + ) + events.extend(batch) + scored += 1 + page += 1 + + posted = False + status = None + if events: + import langfuse_trace + + host, pk, sk = _conf() + # Chunked: one 2000-event POST is refused, and a partial backfill that + # reports success is worse than a slow one. + for i in range(0, len(events), 200): + status = langfuse_trace._post(host, pk, sk, events[i:i + 200], 30.0) + posted = status in (200, 201, 207) + if not posted: + break + return {"traces_scored": scored, "traces_skipped": skipped, "scores": len(events), + "posted": posted, "http_status": status} + + +def main() -> int: + ap = argparse.ArgumentParser(description="Bootstrap Langfuse evaluation for the pragent pilot") + ap.add_argument("--db", default=os.environ.get("PRAGENT_FEEDBACK_DB", "/data/feedback.db")) + ap.add_argument("--skip-dataset", action="store_true") + ap.add_argument("--skip-configs", action="store_true") + ap.add_argument("--backfill-traces", action="store_true", + help="score traces written before the scorers existed") + args = ap.parse_args() + + out: dict = {} + if not args.skip_configs: + out["score_configs"] = ensure_score_configs() + if not args.skip_dataset: + items = read_review_items(args.db) + out["dataset"] = ensure_dataset(items) + out["dataset"]["items_read"] = len(items) + if args.backfill_traces: + out["trace_backfill"] = backfill_traces() + print(json.dumps(out, indent=2)) + + failed = (out.get("score_configs", {}).get("failed") or []) + ( + out.get("dataset", {}).get("failed") or [] + ) + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pilot/eval_scores.py b/pilot/eval_scores.py new file mode 100644 index 0000000..5eecf0f --- /dev/null +++ b/pilot/eval_scores.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +"""pragent pilot — deterministic review scorers. + +Four numbers computed from a review that already happened, shipped to Langfuse +as scores on the review's trace. All are derived from data the reviewer already +has in hand: no LLM judge, no ground truth, no extra token spend. + +Why these four and not `helpfulness`/`quality` +---------------------------------------------- +They come from what the recorded reviews actually did, not from a generic eval +checklist: + + * `severity_info_ratio` — of the findings ever posted to a PR, effectively all + landed at `info`. Either the model will not commit to a severity or the + per-repo `severity_threshold` is filtering the rest out. Trending the ratio + per model says which. + * `finding_rate` — most reviews post nothing at all. Silence on clean code is + the goal; silence because the run degraded is a failure. Same output, two + causes, and only the rate over time separates them. + * `dropped_findings` — `ai_review.parse_findings` discards any finding whose + `path`/`line` is unusable. That happens silently, so a model that emits ten + findings at invalid locations is indistinguishable from one that found + nothing. This is the only signal here that measures the *model's* output + rather than the review's. + * `cost_per_finding` — the equivalent-cost number is already trended per + review; per finding is what actually compares two models, since a cheaper + model that finds nothing is not cheaper. + +None of these say whether a finding was *correct*. That needs labels, and the +labels come from `feedback_scores.py` once maintainers start reacting to review +comments. Read these as behavioural drift detectors, not as accuracy. + +Fail-open, like every other telemetry path here: a scorer that raises returns no +score rather than failing the review. +""" +from __future__ import annotations + +import uuid +from datetime import datetime, timezone + +# Mirrors ai_review.SEVERITY_RANK. Duplicated rather than imported because this +# module is also run standalone (backfill) where ai_review's import side effects +# are unwanted. +SEVERITY_RANK = {"info": -1, "trivial": 0, "low": 1, "medium": 2, "high": 3, "critical": 4} + +# Findings at or below this rank are "the model declined to commit". `trivial` +# and `info` are advisory by the reviewer's own prompt contract. +_ADVISORY_MAX_RANK = 0 + +# Score names. Named for what is measured, not for the mechanism producing it — +# these land on every trace and become the axis of every chart. +FINDING_RATE = "finding_rate" +SEVERITY_INFO_RATIO = "severity_info_ratio" +SEVERITY_MAX = "severity_max" +DROPPED_FINDINGS = "dropped_findings" +COST_PER_FINDING = "cost_per_finding" + + +def _sev(f: dict) -> str: + return str(f.get("severity") or "medium").strip().lower() + + +def finding_rate(findings: list[dict] | None) -> float: + """How many findings this review posted. 0.0 is the restraint case.""" + return float(len(findings or [])) + + +def severity_info_ratio(findings: list[dict] | None) -> float | None: + """Share of findings the model rated advisory (`info`/`trivial`). + + `None` for a review with no findings — a ratio over an empty set is not 0, + it is undefined, and charting it as 0 would read as "perfectly calibrated". + """ + fs = findings or [] + if not fs: + return None + advisory = sum(1 for f in fs if SEVERITY_RANK.get(_sev(f), 2) <= _ADVISORY_MAX_RANK) + return round(advisory / len(fs), 4) + + +def severity_max(findings: list[dict] | None) -> str: + """Highest severity present, or `none` when the review was silent. + + Categorical on purpose: the useful question is "did this review ever surface + something serious", and an average of severity ranks answers nothing. + """ + fs = findings or [] + if not fs: + return "none" + top = max(fs, key=lambda f: SEVERITY_RANK.get(_sev(f), 2)) + sev = _sev(top) + return sev if sev in SEVERITY_RANK else "medium" + + +def dropped_findings(raw_count: int | None, kept_count: int | None) -> float | None: + """Findings the model emitted that the parser could not use. + + `raw_count` is what came back in the JSON; `kept_count` is what survived + `_normalize_finding`. `None` when the caller could not determine the raw + count — better no score than a fabricated zero. + """ + if raw_count is None or kept_count is None: + return None + return float(max(0, int(raw_count) - int(kept_count))) + + +def cost_per_finding(cost_usd: float | None, findings: list[dict] | None) -> float | None: + """Equivalent USD spent per finding posted. + + `None` when nothing could be priced. A silent review divides by one, not by + zero: the run still cost money, and attributing that whole cost to "found + nothing" is the honest reading. + """ + if cost_usd is None: + return None + try: + c = float(cost_usd) + except (TypeError, ValueError): + return None + return round(c / max(1, len(findings or [])), 6) + + +def build_scores( + *, + trace_id: str, + findings: list[dict] | None, + environment: str, + cost_usd: float | None = None, + dropped_count: float | None = None, + timestamp: str | None = None, + comment: str = "", +) -> list[dict]: + """The `score-create` ingestion events for one review. + + `dropped_count` must be measured at parse time, not here: by the time + `findings` reaches this function the per-repo config has already filtered it + by severity threshold and `max_findings`, and those drops are the config + working as intended, not the model emitting garbage. + + Returns [] rather than raising if something is unscoreable — scores are + telemetry and must never cost a review. + """ + # The ingestion envelope requires a timestamp on every event; omitting it + # gets the whole batch rejected with an HTTP 207 whose per-event 400s are + # easy to mistake for success. + ts = timestamp or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + out: list[dict] = [] + + def add(name: str, value, data_type: str) -> None: + if value is None: + return + body = { + "id": str(uuid.uuid4()), + "traceId": trace_id, + "name": name, + "dataType": data_type, + "environment": environment, + } + if data_type == "CATEGORICAL": + body["value"] = str(value) + else: + body["value"] = float(value) + if comment: + body["comment"] = comment + out.append( + { + "id": str(uuid.uuid4()), + "type": "score-create", + "timestamp": ts, + "body": body, + } + ) + + try: + add(FINDING_RATE, finding_rate(findings), "NUMERIC") + add(SEVERITY_INFO_RATIO, severity_info_ratio(findings), "NUMERIC") + add(SEVERITY_MAX, severity_max(findings), "CATEGORICAL") + add(DROPPED_FINDINGS, dropped_count, "NUMERIC") + add(COST_PER_FINDING, cost_per_finding(cost_usd, findings), "NUMERIC") + except Exception: # pragma: no cover - defensive + return out + return out + + +# --------------------------------------------------------------------------- +# Score configs — the schema these scores must comply with +# --------------------------------------------------------------------------- + +# Registered once per project via `eval_bootstrap.py`. Without configs the +# scores still ingest, but nothing constrains a future scorer from writing +# `severity_max="HIGH"` next to today's `"high"` and silently splitting the +# series in two. +SCORE_CONFIGS = [ + { + "name": FINDING_RATE, + "dataType": "NUMERIC", + "minValue": 0, + "description": "Findings posted by one review. 0 = the reviewer stayed silent.", + }, + { + "name": SEVERITY_INFO_RATIO, + "dataType": "NUMERIC", + "minValue": 0, + "maxValue": 1, + "description": "Share of a review's findings rated info/trivial. High = the model is not committing to a severity.", + }, + { + "name": SEVERITY_MAX, + "dataType": "CATEGORICAL", + "categories": [ + {"label": "none", "value": 0}, + {"label": "info", "value": 1}, + {"label": "trivial", "value": 2}, + {"label": "low", "value": 3}, + {"label": "medium", "value": 4}, + {"label": "high", "value": 5}, + {"label": "critical", "value": 6}, + ], + "description": "Highest severity surfaced by one review; 'none' when it posted nothing.", + }, + { + "name": DROPPED_FINDINGS, + "dataType": "NUMERIC", + "minValue": 0, + "description": "Findings the model emitted that the parser rejected for an unusable path/line.", + }, + { + "name": COST_PER_FINDING, + "dataType": "NUMERIC", + "minValue": 0, + "description": "Equivalent USD per finding posted. Silent reviews divide by 1, not 0.", + }, +] diff --git a/pilot/feedback_scores.py b/pilot/feedback_scores.py new file mode 100644 index 0000000..276aee6 --- /dev/null +++ b/pilot/feedback_scores.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +"""pragent pilot — feedback DB to Langfuse scores. + +`feedback.db` already records every reaction, thread resolution and reply a +maintainer leaves on a bot comment. That is the only ground truth pragent has +about whether a finding was any good, and until now it went to a markdown report +nobody reads and nowhere else. This ships it to Langfuse as session-level +scores, so "was the reviewer right" sits on the same axis as "what did it cost". + +Session, not trace +------------------ +`langfuse_trace` sets `sessionId` to `"{repo}#{pr}"` and lets the trace id be a +fresh uuid per review. Feedback arrives days later against a PR, not against one +particular re-run of the reviewer, and nothing in `feedback.db` records which +trace produced which comment. Scoring the session is therefore both the +available join and the honest granularity: this is feedback on the review of +this PR, not on one invocation. + +Two scores, deliberately separated +---------------------------------- + * `review_engagement` — the share of a PR's findings that got any human + response at all. This is a signal about the *feedback loop*, not the + reviewer: at the time of writing it is 0.0 across all 113 recorded reviews, + which is exactly the fact that makes an accuracy metric impossible today. + It must be watched first, because every other quality number is vapour + until it moves. + * `review_acceptance` — net verdict over the findings that *did* get a + response: (upvotes + resolved) - (downvotes + negation replies), normalised + to -1..1. Computed only over engaged findings, so an ignored review scores + `None` rather than 0. Zero would read as "humans judged this exactly + neutral"; the truth is nobody looked. + +Fail-open and idempotent. Score ids are derived from (repo, pr, name) so a +re-run overwrites rather than duplicates. +""" +from __future__ import annotations + +import argparse +import json +import os +import sqlite3 +import sys +import uuid +from datetime import datetime, timezone + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from feedback_harvest import classify_reaction, _is_negation_reply # noqa: E402 + +REVIEW_ENGAGEMENT = "review_engagement" +REVIEW_ACCEPTANCE = "review_acceptance" + +# Stable namespace so the same (repo, pr, score) always produces the same score +# id — Langfuse treats a repeated id as an update, which is what a backfill of a +# still-accumulating PR should do. +_NS = uuid.UUID("6f1d9c2e-4a77-4f2a-9c1a-0d3b5e8a7c41") + + +def _score_id(repo: str, pr: int, name: str) -> str: + return str(uuid.uuid5(_NS, f"{repo}#{pr}#{name}")) + + +def collect_pr_feedback(conn: sqlite3.Connection, repo: str, pr: int) -> dict: + """Tally one PR's findings and the human responses attached to them. + + Returns counts only — the scoring maths lives in `score_pr` so it can be + tested without a database. + """ + rows = conn.execute( + "SELECT id, comment_id FROM inline_finding WHERE repo = ? AND pr = ?", + (repo, pr), + ).fetchall() + total = len(rows) + engaged = 0 + positive = 0 + negative = 0 + + for row in rows: + fid = row["id"] if isinstance(row, sqlite3.Row) else row[0] + cid = row["comment_id"] if isinstance(row, sqlite3.Row) else row[1] + pos = neg = 0 + + if cid is not None: + for r in conn.execute( + "SELECT content FROM reaction WHERE comment_id = ?", (cid,) + ): + kind = classify_reaction(r[0]) + if kind == "positive": + pos += 1 + elif kind == "negative": + neg += 1 + + for r in conn.execute( + "SELECT resolved FROM thread_state WHERE finding_id = ?", (fid,) + ): + # A resolved thread means the maintainer acted on the finding. + if r[0]: + pos += 1 + + # A reply counts as engagement either way; only a negation phrase makes + # it a vote against. A neutral reply ("done", "good catch, but…") is + # deliberately not a positive vote — it says someone looked, not that + # they agreed. + replied = 0 + for r in conn.execute( + "SELECT body FROM reply WHERE finding_id = ?", (fid,) + ): + replied += 1 + if _is_negation_reply(r[0]): + neg += 1 + + if pos or neg or replied: + engaged += 1 + positive += pos + negative += neg + + return {"total": total, "engaged": engaged, "positive": positive, "negative": negative} + + +def score_pr(tally: dict) -> dict: + """Turn one PR's tally into score values. + + `review_acceptance` is `None` when nothing was engaged — see the module + docstring on why that is not 0. + """ + total = int(tally.get("total") or 0) + engaged = int(tally.get("engaged") or 0) + pos = int(tally.get("positive") or 0) + neg = int(tally.get("negative") or 0) + + engagement = round(engaged / total, 4) if total else None + acceptance = None + if pos or neg: + acceptance = round((pos - neg) / (pos + neg), 4) + return {REVIEW_ENGAGEMENT: engagement, REVIEW_ACCEPTANCE: acceptance} + + +def build_score_events( + repo: str, pr: int, values: dict, environment: str = "default", + timestamp: str | None = None, +) -> list[dict]: + """`score-create` events for one PR's feedback. + + Every event carries a timestamp: the ingestion endpoint rejects those that + do not, and it reports the rejection as a per-event 400 inside an HTTP 207, + which reads as success to a caller that only checks the status code. + """ + ts = timestamp or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + events = [] + for name, value in values.items(): + if value is None: + continue + events.append( + { + "id": str(uuid.uuid4()), + "type": "score-create", + "timestamp": ts, + "body": { + "id": _score_id(repo, pr, name), + "sessionId": f"{repo}#{pr}", + "name": name, + "value": float(value), + "dataType": "NUMERIC", + "environment": environment, + "comment": f"from feedback.db · {repo}#{pr}", + }, + } + ) + return events + + +SCORE_CONFIGS = [ + { + "name": REVIEW_ENGAGEMENT, + "dataType": "NUMERIC", + "minValue": 0, + "maxValue": 1, + "description": "Share of a PR's findings that drew any human reaction, resolution or reply. 0 = nobody engaged with the review.", + }, + { + "name": REVIEW_ACCEPTANCE, + "dataType": "NUMERIC", + "minValue": -1, + "maxValue": 1, + "description": "Net human verdict over engaged findings: +1 all accepted, -1 all rejected. Absent when nothing was engaged.", + }, +] + + +def iter_prs(conn: sqlite3.Connection): + for row in conn.execute( + "SELECT DISTINCT repo, pr FROM inline_finding ORDER BY repo, pr" + ): + yield row[0], int(row[1]) + + +def backfill(db_path: str, *, environment: str = "default", dry_run: bool = False) -> dict: + """Score every PR in the feedback DB. Returns a summary dict.""" + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + events: list[dict] = [] + scanned = 0 + engaged_prs = 0 + try: + for repo, pr in iter_prs(conn): + scanned += 1 + tally = collect_pr_feedback(conn, repo, pr) + values = score_pr(tally) + if (values.get(REVIEW_ENGAGEMENT) or 0) > 0: + engaged_prs += 1 + events.extend(build_score_events(repo, pr, values, environment)) + finally: + conn.close() + + summary = {"prs_scanned": scanned, "prs_with_engagement": engaged_prs, "scores": len(events)} + if dry_run or not events: + summary["posted"] = False + return summary + + import langfuse_trace + + conf = langfuse_trace._enabled() + if conf is None: + summary["posted"] = False + summary["error"] = "Langfuse not configured (LANGFUSE_HOST / keys unset)" + return summary + host, pk, sk = conf + status = langfuse_trace._post(host, pk, sk, events, 15.0) + summary["posted"] = status in (200, 201, 207) + summary["http_status"] = status + return summary + + +def main() -> int: + ap = argparse.ArgumentParser(description="Ship feedback.db verdicts to Langfuse as scores") + ap.add_argument("--db", default=os.environ.get("PRAGENT_FEEDBACK_DB", "/data/feedback.db")) + ap.add_argument("--environment", default="default") + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + summary = backfill(args.db, environment=args.environment, dry_run=args.dry_run) + print(json.dumps(summary, indent=2)) + return 0 if summary.get("posted") or args.dry_run else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pilot/langfuse_trace.py b/pilot/langfuse_trace.py index 484fc7e..4ad47ac 100644 --- a/pilot/langfuse_trace.py +++ b/pilot/langfuse_trace.py @@ -218,11 +218,17 @@ def build_batch( trace_id: str | None = None, release: str = "", price_target: str | None = None, + dropped_count: float | None = None, ) -> list[dict]: - """The ingestion batch for one review: a trace plus one generation. + """The ingestion batch for one review: a trace, a generation, and scores. Split out from `emit_review_trace` so the shape is testable without a Langfuse to POST to. + + `dropped_count` is how many findings the parser rejected for an unusable + `path`/`line`, measured where the model output was parsed. Passing it turns + on the `dropped_findings` score; leaving it `None` omits that score rather + than reporting a zero the caller never measured. """ usage = usage or {} tid = trace_id or str(uuid.uuid4()) @@ -316,9 +322,40 @@ def build_batch( } ) + events.extend( + _score_events( + trace_id=tid, + findings=findings, + environment=env, + cost_usd=costs.get("total"), + dropped_count=dropped_count, + timestamp=ts, + cost_basis=cost_basis, + ) + ) + return events +def _score_events(*, cost_basis: str, **kwargs) -> list[dict]: + """Deterministic scores for this review, or [] if the scorer is missing. + + Local import + blanket except for the same reason the rest of this module + swallows: `eval_scores` is optional, and a scoring bug must not cost the + trace it was supposed to annotate. + """ + try: + import eval_scores + + # The cost score is only meaningful next to its basis — a $/finding + # figure computed from an equivalent price is not money that was spent. + comment = f"cost basis: {cost_basis}" if cost_basis else "" + return eval_scores.build_scores(comment=comment, **kwargs) + except Exception as e: # pragma: no cover - defensive + _debug(f"scoring failed: {e}") + return [] + + def _post(host: str, pk: str, sk: str, batch: list[dict], timeout: float) -> int: payload = json.dumps({"batch": batch}).encode("utf-8") auth = base64.b64encode(f"{pk}:{sk}".encode("utf-8")).decode("ascii") @@ -333,9 +370,31 @@ def _post(host: str, pk: str, sk: str, batch: list[dict], timeout: float) -> int method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as resp: + _warn_on_rejected_events(resp.read()) return resp.status +def _warn_on_rejected_events(raw: bytes) -> None: + """Surface per-event rejections hiding inside a 207. + + The ingestion endpoint answers 207 Multi-Status when *some* events failed, + so a caller that only checks the status code reads a batch where every + single event was rejected as a success. That failure mode is invisible + exactly when it matters — the traces simply never appear. + """ + try: + body = json.loads(raw or b"{}") + errors = body.get("errors") or [] + if errors: + first = errors[0] + _debug( + f"{len(errors)} event(s) rejected by ingestion; " + f"first: status={first.get('status')} {first.get('error')}" + ) + except Exception: # pragma: no cover - never let logging break emission + pass + + def emit_review_trace(**kwargs) -> bool: """Ship one review's trace. Returns True if Langfuse accepted it. diff --git a/tests/pilot/test_eval_scores.py b/tests/pilot/test_eval_scores.py new file mode 100644 index 0000000..f0398e8 --- /dev/null +++ b/tests/pilot/test_eval_scores.py @@ -0,0 +1,200 @@ +"""Tests for the deterministic review scorers.""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot")) + +import eval_scores as es # noqa: E402 + + +def f(sev, path="a.py", line=1): + return {"severity": sev, "path": path, "line": line, "problem": "p", "fix": ""} + + +# --- finding_rate --------------------------------------------------------- + +def test_finding_rate_counts_findings(): + assert es.finding_rate([f("high"), f("low")]) == 2.0 + + +def test_finding_rate_zero_for_silent_review(): + assert es.finding_rate([]) == 0.0 + assert es.finding_rate(None) == 0.0 + + +# --- severity_info_ratio -------------------------------------------------- + +def test_info_ratio_all_advisory(): + assert es.severity_info_ratio([f("info"), f("trivial")]) == 1.0 + + +def test_info_ratio_mixed(): + assert es.severity_info_ratio([f("info"), f("high")]) == 0.5 + + +def test_info_ratio_none_when_no_findings(): + # Undefined, not zero — zero would read as perfectly calibrated. + assert es.severity_info_ratio([]) is None + + +def test_info_ratio_unknown_severity_treated_as_medium(): + # Matches _normalize_finding's fallback, so an odd severity is not + # silently counted as advisory. + assert es.severity_info_ratio([f("bogus")]) == 0.0 + + +# --- severity_max --------------------------------------------------------- + +def test_severity_max_picks_highest(): + assert es.severity_max([f("info"), f("critical"), f("low")]) == "critical" + + +def test_severity_max_none_when_silent(): + assert es.severity_max([]) == "none" + + +def test_severity_max_case_insensitive(): + assert es.severity_max([f("HIGH")]) == "high" + + +# --- dropped_findings ----------------------------------------------------- + +def test_dropped_findings_delta(): + assert es.dropped_findings(5, 2) == 3.0 + + +def test_dropped_findings_never_negative(): + assert es.dropped_findings(1, 3) == 0.0 + + +def test_dropped_findings_none_when_unknown(): + assert es.dropped_findings(None, 2) is None + + +# --- cost_per_finding ----------------------------------------------------- + +def test_cost_per_finding_divides(): + assert es.cost_per_finding(1.0, [f("high"), f("low")]) == 0.5 + + +def test_cost_per_finding_silent_review_divides_by_one(): + # The run still cost money; attributing all of it to "found nothing" is + # the honest reading, and it avoids a division by zero. + assert es.cost_per_finding(0.25, []) == 0.25 + + +def test_cost_per_finding_none_when_unpriced(): + assert es.cost_per_finding(None, [f("high")]) is None + + +def test_cost_per_finding_none_on_garbage(): + assert es.cost_per_finding("abc", [f("high")]) is None + + +# --- build_scores --------------------------------------------------------- + +def _by_name(events): + return {e["body"]["name"]: e["body"] for e in events} + + +def test_build_scores_emits_expected_set(): + events = es.build_scores( + trace_id="t1", findings=[f("high"), f("info")], environment="claude", + cost_usd=0.5, dropped_count=2, timestamp="2026-01-01T00:00:00Z", + ) + names = _by_name(events) + assert set(names) == { + es.FINDING_RATE, es.SEVERITY_INFO_RATIO, es.SEVERITY_MAX, + es.DROPPED_FINDINGS, es.COST_PER_FINDING, + } + assert names[es.FINDING_RATE]["value"] == 2.0 + assert names[es.SEVERITY_MAX]["value"] == "high" + assert names[es.DROPPED_FINDINGS]["value"] == 2.0 + assert names[es.COST_PER_FINDING]["value"] == 0.25 + + +def test_build_scores_all_events_are_score_create_on_the_trace(): + events = es.build_scores( + trace_id="t9", findings=[f("low")], environment="ollama", cost_usd=1.0, + ) + assert all(e["type"] == "score-create" for e in events) + assert all(e["body"]["traceId"] == "t9" for e in events) + assert all(e["body"]["environment"] == "ollama" for e in events) + + +def test_build_scores_omits_undefined_scores(): + # No cost and no drop count measured -> those scores are absent, not zero. + events = es.build_scores(trace_id="t2", findings=[], environment="ollama") + names = set(_by_name(events)) + assert es.COST_PER_FINDING not in names + assert es.DROPPED_FINDINGS not in names + assert es.SEVERITY_INFO_RATIO not in names + assert names == {es.FINDING_RATE, es.SEVERITY_MAX} + + +def test_build_scores_categorical_value_is_string(): + events = es.build_scores(trace_id="t3", findings=[f("high")], environment="claude") + sev = _by_name(events)[es.SEVERITY_MAX] + assert sev["dataType"] == "CATEGORICAL" + assert isinstance(sev["value"], str) + + +def test_build_scores_numeric_values_are_floats(): + events = es.build_scores( + trace_id="t4", findings=[f("high")], environment="claude", cost_usd=1, + ) + for name, body in _by_name(events).items(): + if body["dataType"] == "NUMERIC": + assert isinstance(body["value"], float), name + + +def test_build_scores_comment_propagates(): + events = es.build_scores( + trace_id="t5", findings=[f("high")], environment="claude", + cost_usd=1.0, comment="cost basis: equivalent:claude-sonnet-5", + ) + assert all("equivalent" in e["body"]["comment"] for e in events) + + +# --- score configs -------------------------------------------------------- + +def test_every_emitted_score_has_a_config(): + configured = {c["name"] for c in es.SCORE_CONFIGS} + events = es.build_scores( + trace_id="t6", findings=[f("high")], environment="claude", + cost_usd=1.0, dropped_count=0, + ) + assert set(_by_name(events)) <= configured + + +def test_severity_max_config_covers_every_severity_it_can_emit(): + labels = {c["label"] for c in + next(c for c in es.SCORE_CONFIGS if c["name"] == es.SEVERITY_MAX)["categories"]} + assert set(es.SEVERITY_RANK) | {"none"} == labels + + +# --- ingestion envelope --------------------------------------------------- + +def test_every_event_carries_a_timestamp(): + # Ingestion rejects events without one, and reports the rejection as a + # per-event 400 inside an HTTP 207 that reads as success. + events = es.build_scores( + trace_id="t7", findings=[f("high")], environment="claude", cost_usd=1.0, + ) + assert events + assert all(e.get("timestamp") for e in events) + + +def test_timestamp_defaults_when_caller_omits_it(): + events = es.build_scores(trace_id="t8", findings=[f("low")], environment="claude") + assert all(isinstance(e["timestamp"], str) and e["timestamp"].endswith("Z") for e in events) + + +def test_explicit_timestamp_is_used(): + events = es.build_scores( + trace_id="t9", findings=[f("low")], environment="claude", + timestamp="2026-01-02T03:04:05Z", + ) + assert all(e["timestamp"] == "2026-01-02T03:04:05Z" for e in events) diff --git a/tests/pilot/test_feedback_scores.py b/tests/pilot/test_feedback_scores.py new file mode 100644 index 0000000..22e2398 --- /dev/null +++ b/tests/pilot/test_feedback_scores.py @@ -0,0 +1,207 @@ +"""Tests for the feedback.db -> Langfuse score bridge.""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot")) + +import feedback # noqa: E402 +import feedback_scores as fs # noqa: E402 + + +@pytest.fixture +def db(tmp_path): + conn = feedback.init(str(tmp_path / "fb.db")) + yield conn + conn.close() + + +def _seed_finding(conn, repo="o/r", pr=1, comment_id=100, path="a.py", line=1): + cur = conn.execute( + "INSERT INTO review (repo, pr, head_sha, posted_at) VALUES (?,?,?,?)", + (repo, pr, "deadbeef", 1000), + ) + review_id = cur.lastrowid + cur = conn.execute( + """INSERT INTO inline_finding + (review_id, repo, pr, path, line, severity, problem, comment_id, posthash, posted_at) + VALUES (?,?,?,?,?,?,?,?,?,?)""", + (review_id, repo, pr, path, line, "HIGH", "problem", comment_id, f"h{comment_id}", 1000), + ) + conn.commit() + return cur.lastrowid + + +# --- score_pr maths ------------------------------------------------------- + +def test_engagement_zero_when_nobody_responded(): + v = fs.score_pr({"total": 4, "engaged": 0, "positive": 0, "negative": 0}) + assert v[fs.REVIEW_ENGAGEMENT] == 0.0 + + +def test_acceptance_absent_when_nobody_engaged(): + # Not 0.0 — zero would claim humans judged it neutral. + v = fs.score_pr({"total": 4, "engaged": 0, "positive": 0, "negative": 0}) + assert v[fs.REVIEW_ACCEPTANCE] is None + + +def test_engagement_is_a_share_of_findings(): + v = fs.score_pr({"total": 4, "engaged": 1, "positive": 1, "negative": 0}) + assert v[fs.REVIEW_ENGAGEMENT] == 0.25 + + +def test_acceptance_all_positive(): + v = fs.score_pr({"total": 2, "engaged": 2, "positive": 3, "negative": 0}) + assert v[fs.REVIEW_ACCEPTANCE] == 1.0 + + +def test_acceptance_all_negative(): + v = fs.score_pr({"total": 2, "engaged": 2, "positive": 0, "negative": 2}) + assert v[fs.REVIEW_ACCEPTANCE] == -1.0 + + +def test_acceptance_mixed_is_normalised(): + v = fs.score_pr({"total": 4, "engaged": 4, "positive": 3, "negative": 1}) + assert v[fs.REVIEW_ACCEPTANCE] == 0.5 + + +def test_engagement_absent_when_no_findings_at_all(): + v = fs.score_pr({"total": 0, "engaged": 0, "positive": 0, "negative": 0}) + assert v[fs.REVIEW_ENGAGEMENT] is None + + +# --- collect_pr_feedback over a real sqlite ------------------------------ + +def test_collect_counts_nothing_on_untouched_findings(db): + _seed_finding(db) + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally == {"total": 1, "engaged": 0, "positive": 0, "negative": 0} + + +def test_collect_counts_positive_reaction(db): + _seed_finding(db, comment_id=101) + db.execute( + "INSERT INTO reaction (comment_id, user, content, created_at) VALUES (?,?,?,?)", + (101, "alice", "+1", 1), + ) + db.commit() + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally["positive"] == 1 and tally["engaged"] == 1 + + +def test_collect_counts_negative_reaction(db): + _seed_finding(db, comment_id=102) + db.execute( + "INSERT INTO reaction (comment_id, user, content, created_at) VALUES (?,?,?,?)", + (102, "bob", "-1", 1), + ) + db.commit() + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally["negative"] == 1 and tally["engaged"] == 1 + + +def test_resolved_thread_counts_positive(db): + fid = _seed_finding(db, comment_id=103) + db.execute( + "INSERT INTO thread_state (finding_id, resolved, checked_at) VALUES (?,?,?)", + (fid, 1, 1), + ) + db.commit() + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally["positive"] == 1 and tally["engaged"] == 1 + + +def test_unresolved_thread_is_not_a_vote(db): + fid = _seed_finding(db, comment_id=104) + db.execute( + "INSERT INTO thread_state (finding_id, resolved, checked_at) VALUES (?,?,?)", + (fid, 0, 1), + ) + db.commit() + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally == {"total": 1, "engaged": 0, "positive": 0, "negative": 0} + + +def test_negation_reply_counts_negative(db): + fid = _seed_finding(db, comment_id=105) + db.execute( + "INSERT INTO reply (finding_id, author, body, created_at) VALUES (?,?,?,?)", + (fid, "carol", "this is a false positive", 1), + ) + db.commit() + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally["negative"] == 1 and tally["engaged"] == 1 + + +def test_neutral_reply_is_engagement_but_not_a_vote(db): + fid = _seed_finding(db, comment_id=106) + db.execute( + "INSERT INTO reply (finding_id, author, body, created_at) VALUES (?,?,?,?)", + (fid, "dave", "done", 1), + ) + db.commit() + tally = fs.collect_pr_feedback(db, "o/r", 1) + assert tally["engaged"] == 1 + assert tally["positive"] == 0 and tally["negative"] == 0 + + +# --- event shape ---------------------------------------------------------- + +def test_build_score_events_shape(): + events = fs.build_score_events("o/r", 7, {fs.REVIEW_ENGAGEMENT: 0.5}, "claude") + assert len(events) == 1 + body = events[0]["body"] + assert events[0]["type"] == "score-create" + assert body["sessionId"] == "o/r#7" + assert body["value"] == 0.5 + assert body["environment"] == "claude" + + +def test_build_score_events_skips_none(): + events = fs.build_score_events("o/r", 7, {fs.REVIEW_ACCEPTANCE: None}) + assert events == [] + + +def test_score_ids_are_stable_across_runs(): + # A backfill re-run must update, not duplicate. + a = fs.build_score_events("o/r", 7, {fs.REVIEW_ENGAGEMENT: 0.5})[0]["body"]["id"] + b = fs.build_score_events("o/r", 7, {fs.REVIEW_ENGAGEMENT: 0.9})[0]["body"]["id"] + assert a == b + + +def test_score_ids_differ_per_pr_and_name(): + e1 = fs.build_score_events("o/r", 7, {fs.REVIEW_ENGAGEMENT: 1})[0]["body"]["id"] + e2 = fs.build_score_events("o/r", 8, {fs.REVIEW_ENGAGEMENT: 1})[0]["body"]["id"] + e3 = fs.build_score_events("o/r", 7, {fs.REVIEW_ACCEPTANCE: 1})[0]["body"]["id"] + assert len({e1, e2, e3}) == 3 + + +def test_backfill_dry_run_reports_without_posting(db, tmp_path): + _seed_finding(db, comment_id=107) + db.commit() + path = db.execute("PRAGMA database_list").fetchone()[2] + summary = fs.backfill(path, dry_run=True) + assert summary["prs_scanned"] == 1 + assert summary["prs_with_engagement"] == 0 + assert summary["posted"] is False + + +def test_every_emitted_score_has_a_config(): + configured = {c["name"] for c in fs.SCORE_CONFIGS} + assert {fs.REVIEW_ENGAGEMENT, fs.REVIEW_ACCEPTANCE} == configured + + +def test_every_event_carries_a_timestamp(): + # Without one the ingestion endpoint 400s the event inside a 207 that the + # caller reads as success. + events = fs.build_score_events("o/r", 1, {fs.REVIEW_ENGAGEMENT: 0.0}) + assert events + assert all(e.get("timestamp") for e in events) + + +def test_explicit_timestamp_is_used(): + events = fs.build_score_events( + "o/r", 1, {fs.REVIEW_ENGAGEMENT: 0.0}, timestamp="2026-01-02T03:04:05Z" + ) + assert events[0]["timestamp"] == "2026-01-02T03:04:05Z" diff --git a/tests/pilot/test_langfuse_trace.py b/tests/pilot/test_langfuse_trace.py index 1cd4634..8c83235 100644 --- a/tests/pilot/test_langfuse_trace.py +++ b/tests/pilot/test_langfuse_trace.py @@ -145,15 +145,18 @@ def test_unknown_comparison_target_yields_no_cost_block_rather_than_a_wrong_one( def test_batch_has_a_trace_and_a_generation_linked_by_trace_id(): batch = lt.build_batch(model="headroom/claude-sonnet-5", **BASE) types = [e["type"] for e in batch] - assert types == ["trace-create", "generation-create"] - trace, gen = batch + # Scores ride in the same batch; the trace and generation lead it. + assert types[:2] == ["trace-create", "generation-create"] + trace, gen = batch[0], batch[1] assert gen["body"]["traceId"] == trace["body"]["id"] assert trace["body"]["environment"] == gen["body"]["environment"] == "claude" -def test_batch_without_usage_is_trace_only(): +def test_batch_without_usage_has_no_generation(): batch = lt.build_batch(model="headroom/glm-5.2:cloud", **{**BASE, "usage": None}) - assert [e["type"] for e in batch] == ["trace-create"] + types = [e["type"] for e in batch] + assert "generation-create" not in types + assert types[0] == "trace-create" def test_trace_carries_repo_pr_session_and_severity_counts(): @@ -226,7 +229,9 @@ def test_configured_emit_posts_to_the_ingestion_endpoint(monkeypatch): assert lt.emit_review_trace(model="headroom/claude-sonnet-5", **BASE) is True # Trailing slash stripped so the path is not doubled. assert seen["host"] == "http://langfuse.test:3000" - assert len(seen["batch"]) == 2 + kinds = [e["type"] for e in seen["batch"]] + assert kinds[:2] == ["trace-create", "generation-create"] + assert "score-create" in kinds def test_transport_failure_is_swallowed(monkeypatch): @@ -243,3 +248,79 @@ def test_non_success_status_reports_failure_without_raising(monkeypatch): _configure(monkeypatch) monkeypatch.setattr(lt, "_post", lambda *a, **k: 401) assert lt.emit_review_trace(model="headroom/glm-5.2:cloud", **BASE) is False + + +# --------------------------------------------------------------------------- +# Scores folded into the review batch (added with eval_scores) +# --------------------------------------------------------------------------- + +def _scores(events): + return {e["body"]["name"]: e["body"] for e in events if e["type"] == "score-create"} + + +def test_build_batch_appends_scores(): + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", + model="headroom/claude-sonnet-5", + usage={"input": 100, "output": 10}, + findings=[{"severity": "high", "path": "a.py", "line": 1}], + ) + names = set(_scores(events)) + assert "finding_rate" in names + assert "severity_max" in names + + +def test_scores_attach_to_the_same_trace(): + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", model="m", + usage={"input": 1, "output": 1}, findings=[], trace_id="fixed-id", + ) + for body in _scores(events).values(): + assert body["traceId"] == "fixed-id" + + +def test_scores_inherit_the_trace_environment(): + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", + model="headroom/glm-5.2:cloud", + usage={"input": 1, "output": 1}, findings=[], + ) + for body in _scores(events).values(): + assert body["environment"] == "ollama" + + +def test_dropped_findings_scored_when_provided(): + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", model="m", + usage={"input": 1, "output": 1}, findings=[], dropped_count=3, + ) + assert _scores(events)["dropped_findings"]["value"] == 3.0 + + +def test_dropped_findings_absent_when_not_measured(): + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", model="m", + usage={"input": 1, "output": 1}, findings=[], + ) + assert "dropped_findings" not in _scores(events) + + +def test_cost_score_carries_its_basis_in_the_comment(): + # An equivalent-cost $/finding must never be read as money spent. + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", + model="headroom/glm-5.2:cloud", + usage={"input": 1000, "output": 100}, findings=[{"severity": "low", "path": "a", "line": 1}], + ) + cpf = _scores(events).get("cost_per_finding") + if cpf is not None: # only when cost_model could price the comparison target + assert "equivalent" in cpf["comment"] + + +def test_batch_without_usage_still_scores_findings(): + # A run with no usage report still produced findings worth scoring. + events = lt.build_batch( + repo="o/r", index="1", sha="abc", title="t", model="m", + usage=None, findings=[{"severity": "critical", "path": "a", "line": 2}], + ) + assert _scores(events)["severity_max"]["value"] == "critical" diff --git a/tests/pilot/test_parse_dropped.py b/tests/pilot/test_parse_dropped.py new file mode 100644 index 0000000..593a65b --- /dev/null +++ b/tests/pilot/test_parse_dropped.py @@ -0,0 +1,90 @@ +"""The parse-time drop counter feeding the `dropped_findings` score. + +A model that emits findings at unusable locations produces an empty findings +list, exactly like a model that found nothing. These tests pin the signal that +tells the two apart. +""" +import json +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.abspath(os.path.join(HERE, "..", "..", "pilot"))) + +import ai_review # noqa: E402 + + +def _payload(findings): + return "```json\n" + json.dumps({"summary": "s", "findings": findings}) + "\n```" + + +GOOD = {"severity": "high", "path": "a.py", "line": 3, "problem": "p", "fix": "f"} +NO_PATH = {"severity": "high", "line": 3, "problem": "p"} +NO_LINE = {"severity": "high", "path": "a.py", "problem": "p"} +BAD_LINE = {"severity": "high", "path": "a.py", "line": 0, "problem": "p"} + + +def test_no_drops_on_clean_output(): + _, findings, *_ = ai_review.parse_review_output(_payload([GOOD, GOOD])) + assert len(findings) == 2 + assert ai_review.last_parse_dropped() == 0 + + +def test_counts_findings_missing_path(): + _, findings, *_ = ai_review.parse_review_output(_payload([GOOD, NO_PATH])) + assert len(findings) == 1 + assert ai_review.last_parse_dropped() == 1 + + +def test_counts_findings_missing_line(): + _, findings, *_ = ai_review.parse_review_output(_payload([NO_LINE, NO_LINE])) + assert findings == [] + assert ai_review.last_parse_dropped() == 2 + + +def test_counts_findings_with_unusable_line(): + _, findings, *_ = ai_review.parse_review_output(_payload([BAD_LINE])) + assert findings == [] + assert ai_review.last_parse_dropped() == 1 + + +def test_all_dropped_is_distinguishable_from_found_nothing(): + ai_review.parse_review_output(_payload([NO_PATH, NO_PATH, NO_PATH])) + all_dropped = ai_review.last_parse_dropped() + ai_review.parse_review_output(_payload([])) + found_nothing = ai_review.last_parse_dropped() + assert all_dropped == 3 and found_nothing == 0 + + +def test_counter_resets_on_unparseable_output(): + # Otherwise a salvage-path review inherits the previous review's count. + ai_review.parse_review_output(_payload([NO_PATH, NO_PATH])) + assert ai_review.last_parse_dropped() == 2 + ai_review.parse_review_output("no json here at all") + assert ai_review.last_parse_dropped() == 0 + + +def test_counter_resets_on_malformed_json(): + ai_review.parse_review_output(_payload([NO_PATH])) + ai_review.parse_review_output("```json\n{not valid json,,,}\n```") + assert ai_review.last_parse_dropped() == 0 + + +def test_parse_findings_tracks_drops_too(): + # The non-opencode path must be scored on the same basis. + findings = ai_review.parse_findings(json.dumps({"findings": [GOOD, NO_PATH]})) + assert len(findings) == 1 + assert ai_review.last_parse_dropped() == 1 + + +def test_parse_findings_resets_on_garbage(): + ai_review.parse_findings(json.dumps({"findings": [NO_PATH]})) + assert ai_review.last_parse_dropped() == 1 + ai_review.parse_findings("not json") + assert ai_review.last_parse_dropped() == 0 + + +def test_bare_array_output_is_counted(): + _, findings, *_ = ai_review.parse_review_output("```json\n" + json.dumps([GOOD, NO_PATH]) + "\n```") + assert len(findings) == 1 + assert ai_review.last_parse_dropped() == 1 -- 2.52.0