Freeze calibration-derived document and code relevance hypothesis

tools/relevance-eval/analyze_calibration.py

@@ -0,0 +1,68 @@
+"""Inspect only the frozen calibration baseline; no held-out inputs accepted."""
+import argparse
+import hashlib
+import json
+import math
+from pathlib import Path
+
+
+BASELINE_SHA256 = "56f42c523527dd748918bb8f17be6bd7818158227e692fccaf387d915f9d0a94"
+
+
+def analyze(raw, policy):
+    if hashlib.sha256(raw).hexdigest() != BASELINE_SHA256:
+        raise ValueError("Expected exact frozen calibration baseline")
+    report = json.loads(raw)
+    if not report["complete"] or report["split"] != "calibration" or report["held_out_evaluated"]:
+        raise ValueError("Only complete calibration evidence is accepted")
+    if report["fixture_sha256"] != policy["fixture_sha256"]:
+        raise ValueError("Fixture differs from frozen hypothesis")
+    threshold = policy["semantic_cosine_minimum"]
+    if not math.isfinite(threshold) or not -1 <= threshold <= 1:
+        raise ValueError("Invalid cosine minimum")
+    scopes = {}
+    rows = []
+    for row in report["rows"]:
+        if row["query"]["split"] != "calibration":
+            raise ValueError("Non-calibration query in report")
+        label = row["query"]["classification"]
+        if label not in ("positive", "absent_artifact"):
+            raise ValueError("Unsupported relevance label")
+        score = row["semantic_raw"][0]["score"]
+        if not math.isfinite(score) or not -1 <= score <= 1:
+            raise ValueError("Invalid raw semantic cosine")
+        accepted = score >= threshold
+        group = scopes.setdefault(row["scope"], {
+            "positive_scores": [], "absent_scores": [],
+            "retained_positive_top_results": 0, "absent_queries_with_retained_neighbors": 0,
+        })
+        group["positive_scores" if label == "positive" else "absent_scores"].append(score)
+        if accepted:
+            group["retained_positive_top_results" if label == "positive"
+                  else "absent_queries_with_retained_neighbors"] += 1
+        rows.append({"id": row["query"]["id"], "scope": row["scope"], "label": label,
+                     "top_cosine": score, "accepted_by_hypothesis": accepted})
+    for group in scopes.values():
+        positive = group.pop("positive_scores")
+        absent = group.pop("absent_scores")
+        group.update(positive_requests=len(positive), absent_requests=len(absent),
+                     minimum_positive_top_cosine=min(positive),
+                     maximum_absent_top_cosine=max(absent),
+                     observed_separation=min(positive) - max(absent))
+    return {"policy_id": policy["id"], "calibration_only": True,
+            "held_out_evaluated": False, "production_policy_changed": False,
+            "scopes": scopes, "rows": rows}
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("report", type=Path)
+    parser.add_argument("output", type=Path)
+    args = parser.parse_args()
+    policy_path = Path(__file__).parent / "policies/cosine-v1.json"
+    policy_bytes = policy_path.read_bytes()
+    result = analyze(args.report.read_bytes(), json.loads(policy_bytes))
+    result["policy_sha256"] = hashlib.sha256(policy_bytes).hexdigest()
+    with args.output.open("x", encoding="utf-8") as stream:
+        json.dump(result, stream, indent=2)
+    print(json.dumps(result["scopes"], indent=2))