2026-08-31 18:34:03 +00:00
10 changed files with 1258 additions and 11 deletions
@@ -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."
 }
@@ -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
@@ -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
@@ -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": {
+                    # The UI's filter bar reads metadata and nothing else, so
-                        "reviews_run": int(row["reviews"]),
+                    # anything worth slicing on is a top-level key here even
-                        "last_reviewed_at": int(row["last_seen"]),
+                    # where it duplicates `input`. `owner` and `repo_name` are
-                        # Flags that this row is the reviewer's own past output,
+                    # split out because a filter on the joined `repo` can only
-                        # not a human judgement. Filter on it before anyone
+                    # match one repo at a time, never a whole org.
-                        # treats the dataset as ground truth.
+                    "metadata": _item_metadata(
-                        "labelled_by_human": False,
+                        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
@@ -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())
@@ -0,0 +1,308 @@
 #!/usr/bin/env python3
 """pragent pilot — LLM-as-a-judge evaluators for the reviewer.
 The deterministic scorers in `eval_scores.py` measure *behaviour*: how many
 findings, how severe, how much they cost. None of them can say whether a
 finding was any good. With no human labels in `feedback.db`, a judge is the
 only thing that can — so these two ask the questions that need no ground truth,
 only the review itself:
  `finding_actionability` — is each finding concrete enough to act on? A
    reviewer that says "consider improving error handling" at file level is
    indistinguishable from a useful one by finding count alone. This is the
    failure mode a cheap model degrades into first.
  `review_self_consistency` — does the summary agree with the findings it
    posted? Claiming "no issues found" above a list of two criticals, or
    describing a problem in prose that never became a finding, is a defect the
    reviewer can commit entirely on its own.
 Neither judge is asked whether a finding is *correct*. That needs the diff,
 which these traces do not carry, and a judge asked to rule on correctness from
 a summary alone will confabulate. Accuracy stays an open question until humans
 start labelling — which is what `feedback_scores.py` is there to capture.
 **The judge is a different model from the reviewer.** The reviewer runs
 MiniMax-M2.7; the judge runs kimi-k2.7-code through the same headroom hub. A
 model grading its own output agrees with itself for reasons that have nothing
 to do with quality.
 Evaluators score *observations*, and their variable mapping reads the
 observation's own input/output — which is why `langfuse_trace` now writes the
 review onto the generation and not just onto the trace.
 Usage:
    LANGFUSE_HOST=... LANGFUSE_PUBLIC_KEY=... LANGFUSE_SECRET_KEY=... \\
        python3 eval_judges.py --dry-run
 """
 from __future__ import annotations
 import argparse
 import json
 import os
 import sys
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 import eval_bootstrap as eb  # noqa: E402
 # The headroom hub in front of the local Ollama, plus a small pass-through
 # proxy (`judge-proxy` on 8802) that patches every `thinking` content block
 # to carry the `signature` field Langfuse's Anthropic adapter requires. The
 # underlying model is kimi-k2.7-code through the hub on 8790; the proxy fixes
 # the shape so Mastra's Zod parse stops failing.
 JUDGE_PROVIDER = "headroom-ollama"
 JUDGE_BASE_URL = os.environ.get("PRAGENT_JUDGE_BASE_URL", "http://100.74.17.70:8802")
 JUDGE_API_KEY = os.environ.get("PRAGENT_JUDGE_API_KEY", "ollama")
 JUDGE_MODEL = os.environ.get("PRAGENT_JUDGE_MODEL", "kimi-k2.7-code:cloud")
 # The trace names this project emits (`pr-review` on the trace, `opencode-review`
 # on the generation). Filter on `traceName` rather than observation `name` — the
 # observation-rule schema only exposes `traceName` as a stringOptions column, and
 # every observation inside these traces is the review itself, so the narrowness
 # is the same.
 REVIEW_TRACE_NAMES = ["pr-review", "opencode-review"]
 def _model_config() -> dict:
    return {"provider": JUDGE_PROVIDER, "model": JUDGE_MODEL}
 JUDGES = [
    {
        "name": "finding_actionability",
        "prompt": (
            "You are auditing the output of an automated code reviewer.\n\n"
            "PR under review:\n{{input}}\n\n"
            "What the reviewer produced:\n{{output}}\n\n"
            "Rate how ACTIONABLE the findings are, from 0 to 1. A finding is "
            "actionable when a developer could act on it without asking a "
            "follow-up question: it points at a specific location, names a "
            "concrete problem, and proposes a fix that could be applied.\n\n"
            "Score 1.0 when every finding is specific and fixable. Score around "
            "0.5 when findings identify a real area but leave the developer to "
            "work out what to change. Score near 0.0 when findings are generic "
            "advice that would apply to almost any pull request.\n\n"
            "Judge only specificity and actionability. You cannot see the diff, "
            "so do NOT attempt to judge whether a finding is factually correct, "
            "and do not penalise a finding for being one you cannot verify.\n\n"
            "If the reviewer reported no findings at all, return 1.0 and say in "
            "your reasoning that there was nothing to judge — a silent review is "
            "measured by finding_rate, not here."
        ),
        "outputDefinition": {
            "dataType": "NUMERIC",
            "minValue": 0,
            "maxValue": 1,
            "reasoning": {
                "description": (
                    "Name the least actionable finding and say what it would "
                    "need in order to be acted on."
                )
            },
            "score": {"description": "0 = generic advice, 1 = every finding is specific and fixable."},
        },
    },
    {
        "name": "review_self_consistency",
        "prompt": (
            "You are auditing the output of an automated code reviewer.\n\n"
            "PR under review:\n{{input}}\n\n"
            "What the reviewer produced:\n{{output}}\n\n"
            "The output contains a prose `summary` and a list of `findings`. "
            "Decide whether the summary is CONSISTENT with the findings.\n\n"
            "Inconsistent means, for example: the summary says no issues were "
            "found while findings are listed; the summary describes a problem "
            "that never became a finding; the summary characterises the severity "
            "of the findings in a way the findings themselves contradict; or the "
            "summary refers to files that appear in no finding and in no part of "
            "the PR description.\n\n"
            "A summary that adds context beyond the findings is NOT inconsistent "
            "as long as nothing in it contradicts them. A review that found "
            "nothing and says so is consistent.\n\n"
            "You cannot see the diff. Judge the summary against the findings and "
            "the PR title only — never against what you imagine the code does."
        ),
        "outputDefinition": {
            "dataType": "BOOLEAN",
            "reasoning": {
                "description": "Quote the part of the summary that conflicts with the findings, if any."
            },
            "score": {"description": "true = summary agrees with the findings, false = it contradicts them."},
        },
    },
 ]
 # Both judges read the observation's own input/output.
 MAPPING = [
    {"variable": "input", "source": "input"},
    {"variable": "output", "source": "output"},
 ]
 # ---------------------------------------------------------------------------
 # LLM connection
 # ---------------------------------------------------------------------------
 def ensure_llm_connection() -> dict:
    """Point the project at the judge model. Upserted on `provider`."""
    body = {
        "provider": JUDGE_PROVIDER,
        "adapter": "anthropic",
        "baseURL": JUDGE_BASE_URL,
        "secretKey": JUDGE_API_KEY,
        "customModels": [JUDGE_MODEL],
        # The hub serves two local models and none of Anthropic's, so the
        # default catalogue would be a list of models that all fail on use.
        "withDefaultModels": False,
    }
    st, resp = eb._call("PUT", "/api/public/llm-connections", body)
    return {"status": st, "ok": st in (200, 201), "provider": JUDGE_PROVIDER,
            "error": None if st in (200, 201) else resp}
 # ---------------------------------------------------------------------------
 # Evaluators
 # ---------------------------------------------------------------------------
 def existing_evaluators() -> dict[str, str]:
    """name -> id for evaluators already in the project."""
    out: dict[str, str] = {}
    st, body = eb._call("GET", "/api/public/unstable/evaluators?limit=100")
    if st == 200 and isinstance(body, dict):
        for ev in body.get("data") or []:
            out[ev.get("name")] = ev.get("id")
    return out
 def ensure_evaluators() -> dict:
    """Create each judge if no version exists for the name yet.
    POST /evaluators with a name that already exists creates a new version, not
    a no-op — re-running this script would pile up versions until the page
    listing them is unreadable. Skip when an evaluator of that name is present.
    """
    created, skipped, failed = {}, [], []
    existing = set(existing_evaluators())
    for judge in JUDGES:
        if judge["name"] in existing:
            skipped.append(judge["name"])
            continue
        body = {
            "type": "llm_as_judge",
            "name": judge["name"],
            "prompt": judge["prompt"],
            "outputDefinition": judge["outputDefinition"],
            "modelConfig": _model_config(),
        }
        st, resp = eb._call("POST", "/api/public/unstable/evaluators", body, timeout=60.0)
        if st in (200, 201) and isinstance(resp, dict):
            created[judge["name"]] = resp.get("id")
        else:
            failed.append({"name": judge["name"], "status": st, "error": resp})
    return {"created": created, "skipped": skipped, "failed": failed}
 # ---------------------------------------------------------------------------
 # Rules — what gets judged, and how often
 # ---------------------------------------------------------------------------
 def rule_body(name: str, judge_name: str, sampling: float) -> dict:
    """POST /evaluation-rules shape for an LLM-as-judge observation rule.
    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": "observation",
        "sampling": sampling,
        "filter": [
            {"column": "traceName", "operator": "any of",
             "value": REVIEW_TRACE_NAMES, "type": "stringOptions"},
        ],
        "evaluator": {
            "name": judge_name,
            "scope": "project",
            "variableMapping": MAPPING,
        },
        "mapping": MAPPING,
    }
 def ensure_rules(evaluator_ids: dict[str, str], sampling: float) -> dict:
    """Idempotent: existing rules with the same name are skipped, not duplicated.
    The API has no `name`-keyed upsert; the convention is to POST once and
    re-run the script to verify the response. A duplicate POST raises 409.
    """
    created, failed, skipped = [], [], []
    existing = existing_rule_names()
    for name, eid in evaluator_ids.items():
        if not eid:
            continue
        rule_name = f"{name}-on-reviews"
        if rule_name in existing:
            skipped.append(name)
            continue
        st, resp = eb._call(
            "POST", "/api/public/unstable/evaluation-rules",
            rule_body(rule_name, name, sampling), timeout=60.0,
        )
        if st in (200, 201):
            created.append(name)
        else:
            failed.append({"rule": name, "status": st, "error": resp})
    return {"created": created, "failed": failed, "skipped": skipped}
 def existing_rule_names() -> set[str]:
    """Names of observation-target rules already in the project."""
    out: set[str] = set()
    st, body = eb._call("GET", "/api/public/unstable/evaluation-rules?limit=100")
    if st == 200 and isinstance(body, dict):
        for r in body.get("data") or []:
            if r.get("target") == "observation":
                out.add(r.get("name"))
    return out
 def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--sampling", type=float, default=1.0,
                    help="fraction of matching observations to judge (default: all)")
    ap.add_argument("--skip-connection", action="store_true")
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args(argv)
    if args.dry_run:
        print(json.dumps({
            "would_connect": {"provider": JUDGE_PROVIDER, "baseURL": JUDGE_BASE_URL,
                              "model": JUDGE_MODEL},
            "would_create": [j["name"] for j in JUDGES],
            "existing_evaluators": sorted(existing_evaluators()),
            "sampling": args.sampling,
        }, indent=2))
        return 0
    report = {}
    if not args.skip_connection:
        report["llm_connection"] = ensure_llm_connection()
    report["evaluators"] = ensure_evaluators()
    ids = dict(report["evaluators"]["created"])
    # Fall back to whatever is already registered, so a re-run still wires rules.
    for name, eid in existing_evaluators().items():
        ids.setdefault(name, eid)
    report["rules"] = ensure_rules(
        {j["name"]: ids.get(j["name"]) for j in JUDGES}, args.sampling
    )
    print(json.dumps(report, indent=2))
    return 0
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -280,8 +280,8 @@ def build_batch(
        "timestamp": ts,
        "environment": env,
        "sessionId": f"{repo}#{index}",
-        "input": {"repo": repo, "pr": index, "sha": sha, "title": title},
+        "input": _review_input(repo, index, sha, title),
-        "output": {"summary": summary[:2000], "findings": len(findings or [])},
+        "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.
@@ -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")
@@ -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
@@ -0,0 +1,126 @@
 """Tests for the LLM-as-judge evaluator bootstrap."""
 import os
 import sys
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))
 import eval_judges as ej  # noqa: E402
 # --- rule_body ------------------------------------------------------------
 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"] == "observation"
    assert body["enabled"] is True
 def test_rule_body_filters_on_trace_name():
    """`name` isn't a stringOptions column; only `traceName` is."""
    body = ej.rule_body("rule-x", "finding_actionability", 1.0)
    f = body["filter"][0]
    assert f["column"] == "traceName"
    assert f["operator"] == "any of"
    assert f["type"] == "stringOptions"
    assert "pr-review" in f["value"]
 def test_rule_body_references_evaluator_by_name():
    """Ids are version-specific; rules must name the evaluator across versions."""
    body = ej.rule_body("rule-x", "finding_actionability", 1.0)
    assert body["evaluator"]["name"] == "finding_actionability"
    assert body["evaluator"]["scope"] == "project"
 def test_rule_body_maps_input_and_output():
    """Both judges read the observation's own input/output."""
    body = ej.rule_body("rule-x", "any", 1.0)
    sources = {m["source"] for m in body["mapping"]}
    assert sources == {"input", "output"}
 def test_rule_body_carries_mapping_at_both_levels():
    """The server validates `mapping` at the rule root and echoes it on the evaluator."""
    body = ej.rule_body("rule-x", "any", 1.0)
    assert body["mapping"]
    assert body["evaluator"]["variableMapping"] == body["mapping"]
 def test_rule_body_passes_sampling_through():
    assert ej.rule_body("r", "any", 0.25)["sampling"] == 0.25
 # --- ensure_evaluators idempotency ---------------------------------------
 def test_ensure_evaluators_skips_existing(monkeypatch):
    seen = []
    def fake_call(method, path, body=None, timeout=20.0):
        seen.append(path)
        return 200, {}
    monkeypatch.setattr(ej.eb, "_call", fake_call)
    monkeypatch.setattr(ej, "existing_evaluators",
                        lambda: {"finding_actionability": "id-1", "review_self_consistency": "id-2"})
    res = ej.ensure_evaluators()
    assert res["created"] == {}
    assert sorted(res["skipped"]) == ["finding_actionability", "review_self_consistency"]
    assert res["failed"] == []
    assert seen == []
 def test_ensure_evaluators_records_failures(monkeypatch):
    def fake_call(method, path, body=None, timeout=20.0):
        return 422, "boom"
    monkeypatch.setattr(ej.eb, "_call", fake_call)
    monkeypatch.setattr(ej, "existing_evaluators", lambda: {})
    res = ej.ensure_evaluators()
    assert res["created"] == {}
    assert res["failed"][0]["status"] == 422
 # --- ensure_rules idempotency --------------------------------------------
 def test_ensure_rules_skips_existing(monkeypatch):
    calls = []
    monkeypatch.setattr(ej.eb, "_call",
                        lambda *a, **k: calls.append(a) or (200, {}))
    monkeypatch.setattr(ej, "existing_evaluators",
                        lambda: {"finding_actionability": "id-1",
                                 "review_self_consistency": "id-2"})
    monkeypatch.setattr(ej, "existing_rule_names",
                        lambda: {"finding_actionability-on-reviews",
                                 "review_self_consistency-on-reviews"})
    res = ej.ensure_rules({"finding_actionability": "id-1",
                           "review_self_consistency": "id-2"}, 1.0)
    assert res["created"] == []
    assert sorted(res["skipped"]) == ["finding_actionability", "review_self_consistency"]
    assert calls == []
 def test_ensure_rules_creates_when_missing(monkeypatch):
    calls = []
    monkeypatch.setattr(ej.eb, "_call",
                        lambda *a, **k: calls.append(a) or (201, {}))
    monkeypatch.setattr(ej, "existing_rule_names", lambda: set())
    res = ej.ensure_rules({"finding_actionability": "id-1"}, 1.0)
    assert res["created"] == ["finding_actionability"]
    assert calls[0][0] == "POST"
    assert calls[0][1] == "/api/public/unstable/evaluation-rules"
 # --- judge shape ----------------------------------------------------------
 def test_judges_have_required_keys():
    for j in ej.JUDGES:
        assert j["prompt"]
        assert j["outputDefinition"]["dataType"] in ("NUMERIC", "BOOLEAN", "CATEGORICAL")
 def test_default_base_url_points_at_the_thinking_patch_proxy():
    """`8802` is the judge-proxy that adds a `signature` to thinking blocks."""
    assert "8802" in ej.JUDGE_BASE_URL