test: trigger after reopen

Two llm_as_judge evaluators score the review generation directly: a
NUMERIC 0-1 on finding actionability, a BOOLEAN on whether the summary
agrees with the findings. Both run on every observation whose trace
name is pr-review or opencode-review.

The judge is kimi-k2.7-code through the headroom hub. Local Ollama
returns Anthropic-format responses but the thinking blocks lack the
signature field Langfuse Zod schema requires; the evaluator preflight
fails as Invalid JSON response. A small judge-proxy pod on 8802
forwards to the hub and patches every thinking block with a synthetic
signature before returning.

Trace + generation output now includes the findings themselves
(capped at 25) rather than just the count, so a judge has something
to grade. generation input/output mirrors the trace so an
observation-level evaluator can read them.

Idempotent: existing evaluators and rules are skipped on re-run,
not duplicated. The connection is upserted on provider.

.judge-trigger.txt

@@ -1 +0,0 @@
-# judge trigger 1788203999

.pr-review.json

@@ -0,0 +1,5 @@
+{
+  "enabled": true,
+  "model": "headroom/MiniMax-M2.7",
+  "static_message": "PR-Agent pilot on this repo. Comments are LLM-generated; treat as suggestions, not mandates."
+}

README.md

@@ -185,5 +185,3 @@ review time.
 ## License
 
 Not yet chosen. Until one is added, no reuse rights are granted.
-
-_pilot eval judges test 1788201461_

k8s/judge-proxy.yaml

@@ -0,0 +1,82 @@
+# Judge-side think-block patcher. Stands between Langfuse evaluators and the
+# headroom-ollama hub (port 8790). Local Ollama does not emit the `signature`
+# field that Langfuse's Anthropic adapter's Zod schema requires on every
+# `thinking` content block — without it, the evaluator preflight fails as
+# "Invalid JSON response". The proxy forwards /v1/* verbatim and adds a dummy
+# signature to each thinking block before returning.
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: judge-proxy
+  namespace: pragent
+data:
+  proxy.py: |
+    #!/usr/bin/env python3
+    """Judge proxy: forward to headroom-ollama, fix thinking blocks."""
+    import json, sys, urllib.request, urllib.error
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+    from socketserver import ThreadingMixIn
+    UPSTREAM = "http://100.74.17.70:8790"
+    DUMMY_SIG = "kimi-local-judge-no-signature"
+    class H(BaseHTTPRequestHandler):
+        def _proxy(self):
+            n = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(n) if n else b""
+            h = {k: v for k, v in self.headers.items() if k.lower() not in ("host", "content-length")}
+            req = urllib.request.Request(UPSTREAM + self.path, data=body, headers=h, method=self.command)
+            try:
+                with urllib.request.urlopen(req, timeout=120) as r:
+                    resp_body = r.read(); status = r.status; rh = dict(r.headers)
+            except urllib.error.HTTPError as e:
+                resp_body = e.read(); status = e.code; rh = dict(e.headers)
+            ct = rh.get("content-type", "")
+            if status == 200 and "application/json" in ct and self.path.startswith("/v1/messages"):
+                try:
+                    obj = json.loads(resp_body)
+                    patched = 0
+                    for blk in obj.get("content") or []:
+                        if isinstance(blk, dict) and blk.get("type") == "thinking" and "signature" not in blk:
+                            blk["signature"] = DUMMY_SIG; patched += 1
+                    if patched:
+                        resp_body = json.dumps(obj).encode("utf-8")
+                        rh["content-length"] = str(len(resp_body))
+                        print(f"judge-proxy: patched {patched} thinking block(s)", file=sys.stderr, flush=True)
+                except Exception as e:
+                    print(f"judge-proxy: patch failed: {e}", file=sys.stderr, flush=True)
+            self.send_response(status)
+            for k, v in rh.items():
+                if k.lower() not in ("transfer-encoding", "content-length", "connection"):
+                    self.send_header(k, v)
+            self.send_header("Content-Length", str(len(resp_body)))
+            self.end_headers(); self.wfile.write(resp_body)
+        def do_POST(self): self._proxy()
+        def do_GET(self): self._proxy()
+        def log_message(self, *a, **k): pass
+    class S(ThreadingMixIn, HTTPServer): daemon_threads = True
+    S(("0.0.0.0", 8802), H).serve_forever()
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: judge-proxy
+  namespace: pragent
+  labels:
+    app: judge-proxy
+spec:
+  nodeSelector:
+    kubernetes.io/hostname: kubernets
+  hostNetwork: true
+  dnsPolicy: ClusterFirstWithHostNet
+  restartPolicy: Always
+  containers:
+  - name: p
+    image: python:3.12-alpine
+    command: ["sh","-c","apk add --no-cache ca-certificates >/dev/null && python3 -u /etc/cfg/proxy.py"]
+    volumeMounts:
+    - {name: cfg, mountPath: /etc/cfg}
+    ports:
+    - {containerPort: 8802, hostPort: 8802}
+  volumes:
+  - name: cfg
+    configMap:
+      name: judge-proxy

pilot/README-eval.md

@@ -75,6 +75,103 @@ 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.
 
+### Item ids
+
+`{owner}__{repo}__pr{n}`. The obvious `{repo}#{pr}` cannot be used: items are
+routed as `/datasets/{id}/items/{item_id}`, so the `/` in `owner/repo` splits
+into extra path segments and everything after `#` is a fragment the browser
+never sends — the item is created fine by the API and then 404s when opened.
+Session ids elsewhere keep `{repo}#{pr}`; those are never path segments.
+
+### Filterable metadata
+
+The filter bar matches on `metadata` only — not on `input`, and not on the item
+id — so every facet worth slicing on is a flat, primitive key in `metadata`
+even where it duplicates `input`:
+
+| key | why it is there |
+| --- | --- |
+| `repo`, `owner`, `repo_name` | `owner` exists because a filter on the joined `repo` matches one repo, never a whole org |
+| `pr`, `head_sha` | jump from a filtered row back to the actual PR |
+| `finding_count`, `has_findings` | isolate the silent reviews, which are the interesting negatives |
+| `max_severity` | `"none"` rather than absent — an absent key matches no filter |
+| `reviews_run` | how churny the PR was; high values skew per-item averages |
+| `last_reviewed_at` / `_iso` | epoch sorts, ISO reads |
+| `labelled_by_human` | `false` everywhere today; the flag to filter on before trusting any of it |
+
+Nested objects and lists are deliberately absent: the filter bar cannot reach
+into them.
+
+`max_severity` is derived from `feedback.db`, whose `severity` column is
+re-parsed out of the rendered comment by `feedback_harvest._parse_severity` and
+defaults to `INFO` when its regex misses the badge. Trust the `severity_max`
+**score** (read from the model's structured output) over this facet.
+
+## Experiments
+
+`eval_experiment.py` links reviews that already ran into a dataset run, so the
+Experiments tab is populated without re-running anything. Runs are grouped by
+model — the comparison the pilot actually needs is the same PRs under a
+candidate model with `finding_rate` and `cost_per_finding` side by side. A new
+model produces a new run automatically on the next invocation.
+
+One trace per (run, item), the most recent: a PR re-reviewed on every push has
+many traces, and a run is one output per input.
+
+It uses `POST /api/public/dataset-run-items`, which is deprecated in favour of
+the SDK experiment runner and disappears in Langfuse v4. The deprecation notice
+exempts self-hosted v3 from the cutoff date, and this pilot is stdlib-only by
+design. Revisit when this deployment moves to v4.
+
+Coverage is bounded by the dataset, not by the traces: items only exist for PRs
+with a row in `feedback.db`, and a review that posted no comment leaves a trace
+but no row. That is why a run links fewer items than there are traces.
+
+## Evaluators: `eval_judges.py`
+
+Behaviour scores answer "how many, how severe, how much" — computable from data
+already in hand. Two things they cannot answer:
+
+- **Was the finding any good?** Specificity vs. hedge, generic advice vs.
+  fix-it-now advice — the difference between a useful review and one a
+  developer scrolls past.
+- **Did the summary match the findings?** Claiming "no issues" above two
+  criticals, or describing a problem in prose that never became a finding.
+
+These need a judge. `eval_judges.py` registers two `llm_as_judge` evaluators
+against the trace names this project emits (`pr-review`, `opencode-review`)
+and wires a sampling=1 rule per evaluator. Both run on every observation in a
+matching trace; the only observations in those traces are the review itself.
+
+| evaluator | output | what it answers |
+|---|---|---|
+| `finding_actionability` | NUMERIC 0–1 | How specific and fixable is each finding? |
+| `review_self_consistency` | BOOLEAN | Does the summary agree with the findings? |
+
+The judge is a different model from the reviewer (`kimi-k2.7-code` through the
+headroom hub). A model grading its own output agrees with itself for reasons
+that have nothing to do with quality. The judges are also asked only what they
+can answer from the review itself — never whether a finding is correct, since
+that needs the diff the trace does not carry.
+
+### Why the judge goes through `judge-proxy` (port 8802)
+
+The headroom hub in front of local Ollama returns Anthropic-format responses,
+but every `thinking` content block is missing the `signature` field real
+Claude emits. Langfuse's Zod schema requires it; the omission fails the
+evaluator preflight as `Invalid JSON response`. The `judge-proxy` pod sits in
+front of the hub on `100.74.17.70:8802` and patches every thinking block with
+a synthetic signature before forwarding the response. The model is unchanged;
+only the wire shape is fixed.
+
+```bash
+python3 pilot/eval_judges.py --dry-run   # show what would be created
+python3 pilot/eval_judges.py             # create the LLM connection, evaluators, rules
+```
+
+Idempotent: existing evaluators and rules are skipped, not duplicated. The
+connection is upserted on `provider` so re-runs return the same record.
+
 ## Running it
 
 ```bash
@@ -83,6 +180,10 @@ 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
+
+# link already-traced reviews into a dataset run per model
+python3 pilot/eval_experiment.py --dry-run
+python3 pilot/eval_experiment.py
 ```
 
 Both need `LANGFUSE_HOST`, `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY`. In

pilot/eval_bootstrap.py

@@ -44,6 +44,7 @@ import sqlite3
 import sys
 import urllib.error
 import urllib.request
+from datetime import datetime, timezone
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
@@ -111,6 +112,51 @@ def ensure_score_configs() -> dict:
 # 2. Dataset from recorded reviews
 # ---------------------------------------------------------------------------
 
+def item_id(repo: str, pr) -> str:
+    """A dataset-item id that survives being put in a URL path.
+
+    The obvious `{repo}#{pr}` is unusable: the UI routes items as
+    `/datasets/{id}/items/{item_id}`, so the `/` in `owner/repo` splits into
+    extra path segments and everything after the `#` is a fragment the browser
+    never sends. The item is created fine and then 404s when opened.
+
+    Session ids elsewhere keep the `{repo}#{pr}` form — those are never path
+    segments, and `feedback_scores` depends on that shape.
+    """
+    return f"{repo.replace('/', '__')}__pr{pr}"
+
+
+def _item_metadata(*, repo, pr, head_sha, reviews_run, last_seen, findings) -> dict:
+    """Filterable facets for one dataset item.
+
+    Kept flat and primitive: the filter bar matches a metadata key against a
+    literal, so a nested object or a list is not reachable from the UI.
+    """
+    owner, _, repo_name = str(repo).partition("/")
+    sevs = [str(f["severity"] or "").lower() for f in findings]
+    ranked = [s for s in sevs if s in eval_scores.SEVERITY_RANK]
+    return {
+        "repo": repo,
+        "owner": owner or repo,
+        "repo_name": repo_name or repo,
+        "pr": int(pr),
+        "head_sha": head_sha,
+        "reviews_run": reviews_run,
+        "last_reviewed_at": last_seen,
+        "last_reviewed_iso": datetime.fromtimestamp(last_seen, timezone.utc).isoformat(),
+        "finding_count": len(findings),
+        "has_findings": bool(findings),
+        # "none" rather than omitting the key: a filter for silent reviews needs
+        # something to match, and an absent key matches nothing.
+        "max_severity": (
+            max(ranked, key=lambda s: eval_scores.SEVERITY_RANK[s]) if ranked else "none"
+        ),
+        # Flags that this row is the reviewer's own past output, not a human
+        # judgement. Filter on it before anyone treats the dataset as truth.
+        "labelled_by_human": False,
+    }
+
+
 def read_review_items(db_path: str) -> list[dict]:
     """One dataset item per (repo, pr) the reviewer has run on.
 
@@ -140,7 +186,7 @@ def read_review_items(db_path: str) -> list[dict]:
             ).fetchall()
             items.append(
                 {
-                    "id": f'{row["repo"]}#{row["pr"]}',
+                    "id": item_id(row["repo"], row["pr"]),
                     "input": {
                         "repo": row["repo"],
                         "pr": int(row["pr"]),
@@ -150,14 +196,19 @@ def read_review_items(db_path: str) -> list[dict]:
                         "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,
-                    },
+                    # The UI's filter bar reads metadata and nothing else, so
+                    # anything worth slicing on is a top-level key here even
+                    # where it duplicates `input`. `owner` and `repo_name` are
+                    # split out because a filter on the joined `repo` can only
+                    # match one repo at a time, never a whole org.
+                    "metadata": _item_metadata(
+                        repo=row["repo"],
+                        pr=row["pr"],
+                        head_sha=row["head_sha"],
+                        reviews_run=int(row["reviews"]),
+                        last_seen=int(row["last_seen"]),
+                        findings=findings,
+                    ),
                 }
             )
         return items

pilot/eval_experiment.py

@@ -0,0 +1,212 @@
+#!/usr/bin/env python3
+"""pragent pilot — populate the Experiments tab from reviews already traced.
+
+An "experiment" in Langfuse is a dataset run: a set of (dataset item, trace)
+links under one run name. The Experiments tab then shows one row per item with
+its scores, and lets two runs be diffed side by side.
+
+Nothing here re-runs the reviewer. Every PR in `pragent-reviews` has already
+been reviewed, and each of those reviews left a trace carrying its findings,
+cost and scores. This links what exists, which is what makes the tab useful on
+day one instead of after the next N pushes.
+
+Runs are grouped by **model** by default, because that is the comparison the
+pilot actually needs to make: the same PRs reviewed by MiniMax vs whatever
+replaces it, with `finding_rate` and `cost_per_finding` side by side. Group by
+`none` for a single "all traces" run.
+
+One trace per (run, item) — the most recent. A PR re-reviewed on every push has
+many traces, and a dataset run is defined as one output per input; feeding it
+the other five would make the per-run averages meaningless.
+
+Note on the endpoint: `POST /api/public/dataset-run-items` is deprecated in
+favour of the SDK experiment runner / OTel ingestion, and disappears in
+Langfuse v4. This instance is self-hosted v3, which the deprecation notice
+explicitly exempts from the cutoff date, and the pilot is stdlib-only by
+design. Revisit when this deployment moves to v4.
+
+Usage:
+    LANGFUSE_HOST=... LANGFUSE_PUBLIC_KEY=... LANGFUSE_SECRET_KEY=... \\
+        python3 eval_experiment.py --dry-run
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.parse
+from collections import defaultdict
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import eval_bootstrap as eb  # noqa: E402
+
+TRACE_NAME = "pr-review"
+
+
+# ---------------------------------------------------------------------------
+# Reading what already exists
+# ---------------------------------------------------------------------------
+
+def fetch_traces(name: str = TRACE_NAME, limit: int = 100, max_pages: int = 50) -> list[dict]:
+    """Every review trace, newest first."""
+    out: list[dict] = []
+    for page in range(1, max_pages + 1):
+        q = urllib.parse.urlencode({"name": name, "limit": limit, "page": page})
+        st, body = eb._call("GET", f"/api/public/traces?{q}")
+        if st != 200 or not isinstance(body, dict):
+            raise SystemExit(f"listing traces failed: {st} {body}")
+        data = body.get("data") or []
+        out.extend(data)
+        meta = body.get("meta") or {}
+        if page * meta.get("limit", limit) >= meta.get("totalItems", 0):
+            break
+    return out
+
+
+def fetch_item_ids(dataset: str) -> set[str]:
+    """Ids present in the dataset, so runs never reference a missing item."""
+    ids: set[str] = set()
+    for page in range(1, 51):
+        q = urllib.parse.urlencode({"datasetName": dataset, "limit": 100, "page": page})
+        st, body = eb._call("GET", f"/api/public/dataset-items?{q}")
+        if st != 200 or not isinstance(body, dict):
+            raise SystemExit(f"listing dataset items failed: {st} {body}")
+        ids.update(i["id"] for i in body.get("data") or [])
+        meta = body.get("meta") or {}
+        if page * meta.get("limit", 100) >= meta.get("totalItems", 0):
+            break
+    return ids
+
+
+# ---------------------------------------------------------------------------
+# Grouping traces into runs
+# ---------------------------------------------------------------------------
+
+def trace_model(trace: dict) -> str:
+    """The model that produced a review, from its `model:` tag."""
+    for tag in trace.get("tags") or []:
+        if tag.startswith("model:"):
+            return tag[len("model:"):] or "unknown"
+    return "unknown"
+
+
+def trace_item_id(trace: dict) -> str | None:
+    """The dataset item a trace belongs to, or None if it is not a PR review."""
+    md = trace.get("metadata") or {}
+    repo, pr = md.get("repo"), md.get("pr")
+    if not repo or pr in (None, ""):
+        return None
+    return eb.item_id(str(repo), pr)
+
+
+def _sort_key(trace: dict):
+    return (trace.get("timestamp") or "", trace.get("id") or "")
+
+
+def plan_runs(traces: list[dict], known_items: set[str], group_by: str = "model") -> dict:
+    """Map run name -> {item id: trace}, keeping only the newest trace per item.
+
+    Traces whose PR is not in the dataset are dropped: `feedback.db` is the
+    source for both, but a review can be traced without its row landing (the
+    posting step can fail after the model ran), and a run item pointing at a
+    non-existent dataset item is rejected.
+    """
+    runs: dict[str, dict[str, dict]] = defaultdict(dict)
+    skipped_no_item, skipped_unknown = 0, 0
+    for tr in traces:
+        iid = trace_item_id(tr)
+        if iid is None:
+            skipped_unknown += 1
+            continue
+        if iid not in known_items:
+            skipped_no_item += 1
+            continue
+        run = "all-traces" if group_by == "none" else trace_model(tr)
+        prev = runs[run].get(iid)
+        if prev is None or _sort_key(tr) > _sort_key(prev):
+            runs[run][iid] = tr
+    return {
+        "runs": dict(runs),
+        "skipped_not_in_dataset": skipped_no_item,
+        "skipped_not_a_review": skipped_unknown,
+    }
+
+
+def run_name(prefix: str, key: str) -> str:
+    return f"{prefix}-{key}" if prefix else key
+
+
+# ---------------------------------------------------------------------------
+# Writing the runs
+# ---------------------------------------------------------------------------
+
+def create_run(name: str, items: dict[str, dict], description: str = "") -> dict:
+    """Link each (item, trace) pair into the named run. Idempotent per pair."""
+    created, failed = 0, []
+    for iid, tr in sorted(items.items()):
+        md = tr.get("metadata") or {}
+        body = {
+            "runName": name,
+            "runDescription": description,
+            "datasetItemId": iid,
+            "traceId": tr["id"],
+            "metadata": {
+                "model": trace_model(tr),
+                "engine": md.get("engine"),
+                "findings": md.get("findings"),
+                "duration_s": md.get("duration_s"),
+                "cost_basis": md.get("cost_basis"),
+                "linked_by": "eval_experiment.py",
+            },
+        }
+        st, resp = eb._call("POST", "/api/public/dataset-run-items", body)
+        if st in (200, 201):
+            created += 1
+        else:
+            failed.append({"item": iid, "status": st, "error": resp})
+    return {"run": name, "items_linked": created, "failed": failed}
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--dataset", default=eb.DATASET_NAME)
+    ap.add_argument("--group-by", choices=("model", "none"), default="model")
+    ap.add_argument("--prefix", default="baseline",
+                    help="run name prefix; '' for the bare group key")
+    ap.add_argument("--dry-run", action="store_true")
+    args = ap.parse_args(argv)
+
+    traces = fetch_traces()
+    items = fetch_item_ids(args.dataset)
+    plan = plan_runs(traces, items, group_by=args.group_by)
+
+    report = {
+        "traces_read": len(traces),
+        "dataset_items": len(items),
+        "skipped_not_in_dataset": plan["skipped_not_in_dataset"],
+        "skipped_not_a_review": plan["skipped_not_a_review"],
+        "runs": {},
+    }
+    for key, mapping in sorted(plan["runs"].items()):
+        name = run_name(args.prefix, key)
+        if args.dry_run:
+            report["runs"][name] = {"items_would_link": len(mapping)}
+            continue
+        report["runs"][name] = create_run(
+            name,
+            mapping,
+            description=(
+                "Reviews already run by the pilot, linked after the fact. "
+                "Scores come from the traces; expectedOutput is the reviewer's "
+                "own prior output, not human-verified ground truth."
+            ),
+        )
+    report["dry_run"] = args.dry_run
+    print(json.dumps(report, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pilot/eval_judges.py

@@ -208,24 +208,18 @@ def ensure_evaluators() -> dict:
 # ---------------------------------------------------------------------------
 
 def rule_body(name: str, judge_name: str, sampling: float) -> dict:
-    """POST /evaluation-rules shape for an LLM-as-judge trace rule.
+    """POST /evaluation-rules shape for an LLM-as-judge observation rule.
 
-    Target is `trace` rather than `observation` on purpose: the standard
-    `/api/public/ingestion` path that ships review traces here feeds only
-    the trace-upsert queue, and `evalService.createEvalJobs` only creates
-    jobs for `targetObject ∈ {TRACE, DATASET}`. Observation rules are
-    triggered exclusively from the OTel ingestion pipeline, which this
-    pilot does not use. A trace rule reads the trace's own input/output —
-    `langfuse_trace` already writes `_review_input`/`_review_output` onto
-    the trace body for exactly this reason.
-
-    Mapping is required at both the rule root (server validates it there)
-    and inside `evaluator` (the API echoes it back).
+    The judge is referenced by `name`+`scope`, not by id — ids name specific
+    versions, names name the evaluator across versions. Mapping is required at
+    both the rule root (the server validates it there) and inside `evaluator`
+    (the API echoes it back). Filter is on `traceName` because that is the only
+    stringOptions column the observation-rule schema exposes.
     """
     return {
         "name": name,
         "enabled": True,
-        "target": "trace",
+        "target": "observation",
         "sampling": sampling,
         "filter": [
             {"column": "traceName", "operator": "any of",

pilot/langfuse_trace.py

@@ -280,8 +280,8 @@ def build_batch(
         "timestamp": ts,
         "environment": env,
         "sessionId": f"{repo}#{index}",
-        "input": {"repo": repo, "pr": index, "sha": sha, "title": title},
-        "output": {"summary": summary[:2000], "findings": len(findings or [])},
+        "input": _review_input(repo, index, sha, title),
+        "output": _review_output(summary, findings),
         "metadata": metadata,
         "tags": tags,
     }
@@ -310,6 +310,11 @@ def build_batch(
             "usageDetails": _usage_details(usage),
             "metadata": metadata,
             "level": "DEFAULT",
+            # Repeated from the trace on purpose: an evaluator's variable
+            # mapping reads the *observation's* input/output, so a generation
+            # left blank cannot be judged at all.
+            "input": _review_input(repo, index, sha, title),
+            "output": _review_output(summary, findings),
         }
         if costs:
             gen_body["costDetails"] = costs
@@ -337,6 +342,46 @@ def build_batch(
     return events
 
 
+MAX_JUDGED_FINDINGS = 25
+_FIELD_CAP = 600
+
+
+def _review_input(repo: str, index, sha: str, title: str) -> dict:
+    return {"repo": repo, "pr": index, "sha": sha, "title": title}
+
+
+def _review_output(summary: str, findings) -> dict:
+    """What the reviewer actually said, in a shape an evaluator can read.
+
+    The findings themselves are included, not just their count. A judge given
+    only `{"summary": ..., "findings": 3}` can say nothing about whether those
+    three findings are specific, actionable, or consistent with the summary —
+    which is the whole question worth asking of a reviewer that has no ground
+    truth to check against.
+
+    Capped rather than complete: this rides in every ingestion batch, and a
+    review with 80 findings would push the payload past what is reasonable to
+    store per trace. `finding_count` stays exact so nothing reading the count
+    is misled by the cap.
+    """
+    items = list(findings or [])
+    return {
+        "summary": summary[:2000],
+        "finding_count": len(items),
+        "findings_truncated": len(items) > MAX_JUDGED_FINDINGS,
+        "findings": [
+            {
+                "path": f.get("path"),
+                "line": f.get("line"),
+                "severity": f.get("severity"),
+                "problem": str(f.get("problem") or "")[:_FIELD_CAP],
+                "fix": str(f.get("fix") or "")[:_FIELD_CAP],
+            }
+            for f in items[:MAX_JUDGED_FINDINGS]
+        ],
+    }
+
+
 def _score_events(*, cost_basis: str, **kwargs) -> list[dict]:
     """Deterministic scores for this review, or [] if the scorer is missing.
 

tests/pilot/test_eval_bootstrap.py

@@ -0,0 +1,158 @@
+"""Tests for the eval bootstrap's dataset-item construction."""
+import os
+import sqlite3
+import sys
+import urllib.parse
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))
+
+import eval_bootstrap as eb  # noqa: E402
+
+
+# --- item_id --------------------------------------------------------------
+
+def test_item_id_has_no_path_separator():
+    """A `/` would split the UI's item route into extra path segments."""
+    assert "/" not in eb.item_id("netcracker/interview", 29)
+
+
+def test_item_id_has_no_fragment_marker():
+    """Everything after a `#` is a fragment the browser never sends."""
+    assert "#" not in eb.item_id("netcracker/interview", 29)
+
+
+def test_item_id_survives_a_url_round_trip():
+    """The id must appear verbatim in a path, needing no percent-encoding."""
+    ident = eb.item_id("netcracker/interview", 29)
+    assert urllib.parse.quote(ident, safe="") == ident
+
+
+def test_item_id_keeps_repo_and_pr_readable():
+    assert eb.item_id("netcracker/interview", 29) == "netcracker__interview__pr29"
+
+
+def test_item_id_is_unique_per_pr():
+    assert eb.item_id("o/r", 1) != eb.item_id("o/r", 2)
+
+
+def test_item_id_is_unique_per_repo():
+    assert eb.item_id("o/one", 1) != eb.item_id("o/two", 1)
+
+
+def test_item_id_accepts_a_string_pr():
+    assert eb.item_id("o/r", "29") == eb.item_id("o/r", 29)
+
+
+# --- read_review_items ----------------------------------------------------
+
+def _db(tmp_path, rows, findings=()):
+    path = str(tmp_path / "feedback.db")
+    conn = sqlite3.connect(path)
+    conn.execute(
+        "CREATE TABLE review (repo TEXT, pr INTEGER, posted_at INTEGER, head_sha TEXT)"
+    )
+    conn.execute(
+        "CREATE TABLE inline_finding (repo TEXT, pr INTEGER, path TEXT, line INTEGER,"
+        " severity TEXT, problem TEXT, fix TEXT)"
+    )
+    conn.executemany("INSERT INTO review VALUES (?,?,?,?)", rows)
+    conn.executemany("INSERT INTO inline_finding VALUES (?,?,?,?,?,?,?)", findings)
+    conn.commit()
+    conn.close()
+    return path
+
+
+def test_items_use_url_safe_ids(tmp_path):
+    path = _db(tmp_path, [("netcracker/interview", 29, 100, "abc")])
+    items = eb.read_review_items(path)
+    assert [i["id"] for i in items] == ["netcracker__interview__pr29"]
+
+
+def test_item_input_keeps_the_real_repo_name(tmp_path):
+    """The id is mangled for the URL; the payload must stay faithful."""
+    path = _db(tmp_path, [("netcracker/interview", 29, 100, "abc")])
+    item = eb.read_review_items(path)[0]
+    assert item["input"]["repo"] == "netcracker/interview"
+    assert item["input"]["pr"] == 29
+
+
+def test_one_item_per_pr_not_per_review(tmp_path):
+    path = _db(
+        tmp_path,
+        [
+            ("o/r", 1, 100, "a"),
+            ("o/r", 1, 200, "b"),
+            ("o/r", 2, 300, "c"),
+        ],
+    )
+    items = eb.read_review_items(path)
+    assert [i["id"] for i in items] == ["o__r__pr1", "o__r__pr2"]
+    assert items[0]["metadata"]["reviews_run"] == 2
+
+
+def test_items_are_not_flagged_as_human_labelled(tmp_path):
+    path = _db(tmp_path, [("o/r", 1, 100, "a")])
+    assert eb.read_review_items(path)[0]["metadata"]["labelled_by_human"] is False
+
+
+# --- metadata facets ------------------------------------------------------
+
+def _md(findings=(), repo="netcracker/interview", pr=29):
+    return eb._item_metadata(
+        repo=repo, pr=pr, head_sha="abc", reviews_run=2, last_seen=1788189422,
+        findings=[{"severity": s} for s in findings],
+    )
+
+
+def test_metadata_carries_the_repo_for_filtering():
+    assert _md()["repo"] == "netcracker/interview"
+
+
+def test_metadata_splits_owner_from_repo_name():
+    """A filter on the joined repo can match one repo; owner matches an org."""
+    md = _md()
+    assert md["owner"] == "netcracker"
+    assert md["repo_name"] == "interview"
+
+
+def test_owner_falls_back_when_the_repo_is_unqualified():
+    md = _md(repo="standalone")
+    assert md["owner"] == "standalone"
+    assert md["repo_name"] == "standalone"
+
+
+def test_metadata_values_are_filterable_primitives():
+    """Nested objects and lists are not reachable from the filter bar."""
+    for key, value in _md(["high"]).items():
+        assert isinstance(value, (str, int, float, bool)), key
+
+
+def test_max_severity_is_the_worst_finding():
+    assert _md(["low", "critical", "medium"])["max_severity"] == "critical"
+
+
+def test_max_severity_is_none_not_absent_for_a_silent_review():
+    md = _md([])
+    assert md["max_severity"] == "none"
+    assert md["has_findings"] is False
+
+
+def test_unknown_severity_does_not_win_the_max():
+    assert _md(["banana", "low"])["max_severity"] == "low"
+
+
+def test_severity_comparison_ignores_case():
+    assert _md(["HIGH"])["max_severity"] == "high"
+
+
+def test_finding_count_matches_the_findings():
+    md = _md(["low", "low"])
+    assert md["finding_count"] == 2
+    assert md["has_findings"] is True
+
+
+def test_last_reviewed_is_exposed_both_ways():
+    """The epoch sorts; the ISO string is what a human reads in a filter."""
+    md = _md()
+    assert md["last_reviewed_at"] == 1788189422
+    assert md["last_reviewed_iso"].startswith("2026-08-31T")

tests/pilot/test_eval_experiment.py

@@ -0,0 +1,159 @@
+"""Tests for linking existing review traces into dataset runs."""
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))
+
+import eval_experiment as ex  # noqa: E402
+
+
+def trace(tid, repo="o/r", pr=1, model="M2", ts="2026-08-01T00:00:00Z", **md):
+    meta = {"repo": repo, "pr": pr}
+    meta.update(md)
+    return {
+        "id": tid,
+        "timestamp": ts,
+        "tags": [f"model:{model}", "engine:opencode"],
+        "metadata": meta,
+    }
+
+
+# --- trace_model ----------------------------------------------------------
+
+def test_model_read_from_tag():
+    assert ex.trace_model(trace("t1", model="MiniMax-M2.7")) == "MiniMax-M2.7"
+
+
+def test_model_falls_back_when_untagged():
+    assert ex.trace_model({"tags": ["engine:opencode"]}) == "unknown"
+
+
+def test_model_falls_back_when_tags_absent():
+    assert ex.trace_model({}) == "unknown"
+
+
+# --- trace_item_id --------------------------------------------------------
+
+def test_item_id_matches_the_bootstrap_scheme():
+    assert ex.trace_item_id(trace("t1", repo="netcracker/interview", pr=29)) == \
+        "netcracker__interview__pr29"
+
+
+def test_trace_without_repo_is_not_an_item():
+    assert ex.trace_item_id({"metadata": {"pr": 1}}) is None
+
+
+def test_trace_without_pr_is_not_an_item():
+    assert ex.trace_item_id({"metadata": {"repo": "o/r"}}) is None
+
+
+def test_trace_without_metadata_is_not_an_item():
+    assert ex.trace_item_id({}) is None
+
+
+# --- plan_runs ------------------------------------------------------------
+
+ITEMS = {"o__r__pr1", "o__r__pr2"}
+
+
+def test_traces_group_by_model():
+    plan = ex.plan_runs(
+        [trace("a", pr=1, model="x"), trace("b", pr=2, model="y")], ITEMS
+    )
+    assert set(plan["runs"]) == {"x", "y"}
+
+
+def test_group_by_none_collapses_to_one_run():
+    plan = ex.plan_runs(
+        [trace("a", pr=1, model="x"), trace("b", pr=2, model="y")],
+        ITEMS,
+        group_by="none",
+    )
+    assert list(plan["runs"]) == ["all-traces"]
+
+
+def test_only_the_newest_trace_per_item_is_kept():
+    """A re-reviewed PR has many traces; a run takes one output per input."""
+    plan = ex.plan_runs(
+        [
+            trace("old", pr=1, ts="2026-08-01T00:00:00Z"),
+            trace("new", pr=1, ts="2026-08-09T00:00:00Z"),
+        ],
+        ITEMS,
+    )
+    assert plan["runs"]["M2"]["o__r__pr1"]["id"] == "new"
+
+
+def test_newest_wins_regardless_of_input_order():
+    older = trace("old", pr=1, ts="2026-08-01T00:00:00Z")
+    newer = trace("new", pr=1, ts="2026-08-09T00:00:00Z")
+    for order in ([older, newer], [newer, older]):
+        plan = ex.plan_runs(order, ITEMS)
+        assert plan["runs"]["M2"]["o__r__pr1"]["id"] == "new"
+
+
+def test_trace_for_a_pr_outside_the_dataset_is_skipped():
+    plan = ex.plan_runs([trace("a", pr=99)], ITEMS)
+    assert plan["runs"] == {}
+    assert plan["skipped_not_in_dataset"] == 1
+
+
+def test_non_review_trace_is_counted_separately():
+    plan = ex.plan_runs([{"id": "x", "metadata": {}}], ITEMS)
+    assert plan["skipped_not_a_review"] == 1
+    assert plan["skipped_not_in_dataset"] == 0
+
+
+def test_same_pr_different_models_lands_in_both_runs():
+    plan = ex.plan_runs([trace("a", pr=1, model="x"), trace("b", pr=1, model="y")], ITEMS)
+    assert plan["runs"]["x"]["o__r__pr1"]["id"] == "a"
+    assert plan["runs"]["y"]["o__r__pr1"]["id"] == "b"
+
+
+# --- run_name -------------------------------------------------------------
+
+def test_run_name_prefixed():
+    assert ex.run_name("baseline", "MiniMax-M2.7") == "baseline-MiniMax-M2.7"
+
+
+def test_empty_prefix_leaves_the_key_bare():
+    assert ex.run_name("", "MiniMax-M2.7") == "MiniMax-M2.7"
+
+
+# --- create_run -----------------------------------------------------------
+
+def test_create_run_posts_one_item_per_pair(monkeypatch):
+    calls = []
+
+    def fake_call(method, path, body=None, timeout=20.0):
+        calls.append((method, path, body))
+        return 201, {}
+
+    monkeypatch.setattr(ex.eb, "_call", fake_call)
+    res = ex.create_run("run-1", {"o__r__pr1": trace("t1"), "o__r__pr2": trace("t2", pr=2)})
+    assert res["items_linked"] == 2
+    assert res["failed"] == []
+    assert {c[1] for c in calls} == {"/api/public/dataset-run-items"}
+    assert {c[2]["runName"] for c in calls} == {"run-1"}
+
+
+def test_create_run_links_the_trace_to_the_item(monkeypatch):
+    seen = {}
+
+    def fake_call(method, path, body=None, timeout=20.0):
+        seen.update(body)
+        return 201, {}
+
+    monkeypatch.setattr(ex.eb, "_call", fake_call)
+    ex.create_run("run-1", {"o__r__pr1": trace("t1")})
+    assert seen["datasetItemId"] == "o__r__pr1"
+    assert seen["traceId"] == "t1"
+    assert seen["metadata"]["model"] == "M2"
+
+
+def test_create_run_reports_rejected_items(monkeypatch):
+    monkeypatch.setattr(ex.eb, "_call", lambda *a, **k: (400, "nope"))
+    res = ex.create_run("run-1", {"o__r__pr1": trace("t1")})
+    assert res["items_linked"] == 0
+    assert res["failed"][0]["item"] == "o__r__pr1"
+    assert res["failed"][0]["status"] == 400

tests/pilot/test_eval_judges.py

@@ -11,16 +11,10 @@ import eval_judges as ej  # noqa: E402
 
 # --- rule_body ------------------------------------------------------------
 
-def test_rule_body_targets_traces():
-    """Trace target matches the path `/api/public/ingestion` triggers.
-
-    Observation rules only fire from the OTel ingestion pipeline; this
-    pilot uses standard ingestion, so its jobs only come from
-    `evalService.createEvalJobs` and that dispatcher handles
-    `targetObject ∈ {TRACE, DATASET}`.
-    """
+def test_rule_body_targets_observations():
+    """Trace-level rules wouldn't see observation input/output."""
     body = ej.rule_body("rule-x", "finding_actionability", 1.0)
-    assert body["target"] == "trace"
+    assert body["target"] == "observation"
     assert body["enabled"] is True