feat(eval): LLM-as-judge evaluators for finding actionability and review self-consistency

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.

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.