Add frozen shared-service speech retrieval and negative-control evaluator

tools/speech-eval/test_score_native_validation.py

@@ -0,0 +1,50 @@
+import copy
+import unittest
+
+from score_native_validation import score
+
+
+class ScoringTests(unittest.TestCase):
+    def inputs(self):
+        fixture = {"assets": [{"id": "speech-en", "reference_keywords": ["equal"]}, {"id": "speech-de", "reference_keywords": ["gleich"]}],
+                   "queries": [{"id": "positive", "language": "en", "text": "equal dignity", "expected": ["speech-en", "speech-de"]}, {"id": "absent", "language": "en", "text": "salary", "expected": []}]}
+        sources = [{"id": id, "windows": [{"pcm_sha256": id}]} for id in ("speech-en", "speech-de", "music-control", "silence-control")]
+        calls = [{"pcm_sha256": source["id"], "expected_source_windows": [{"asset": source["id"]}], "transcription": {"passages": []}} for source in sources]
+        core = {"sources": sources, "native_calls": calls, "comparison_eligible": True, "published_speech": [], "rows": []}
+        for query in fixture["queries"]:
+            core["rows"].append({"query": query, "strict_point_score_eligible": False,
+                                 "hits": [{"asset": "speech-de", "result": {"match_kind": "semantic_speech", "snippet": "gleich", "source": {"kind": "timestamp_ms", "number": 5000}}}]})
+        return {"complete": True, "strict_point_score_eligible": False, "core": core, "seams": {"sources": [], "native_calls": [], "comparison_eligible": True}}, fixture
+
+    def test_no_answer_return_is_failure_not_positive_recall(self):
+        report, fixture = self.inputs()
+        result = score(report, fixture)
+        self.assertEqual(result["top1"], 1)
+        self.assertEqual(result["negative_returned_any"], 1)
+        self.assertFalse(result["strict_point_score_eligible"])
+
+    def test_dropped_native_window_invalidates_comparison(self):
+        report, fixture = self.inputs()
+        report["core"]["native_calls"].pop()
+        result = score(report, fixture)
+        self.assertFalse(result["comparison_eligible"])
+        self.assertIsNone(result["top1"])
+        self.assertIsNone(result["mrr"])
+
+    def test_silence_hallucination_is_retained_failure(self):
+        report, fixture = self.inputs()
+        report["core"]["native_calls"][-1]["transcription"]["passages"] = [{"text": "fake speech"}]
+        self.assertFalse(score(report, fixture)["negative_gate_acceptance"])
+
+    def test_frozen_query_mutation_is_rejected(self):
+        report, fixture = self.inputs()
+        report = copy.deepcopy(report)
+        report["core"]["rows"][0]["query"]["text"] = "changed"
+        with self.assertRaises(ValueError):
+            score(report, fixture)
+
+    def test_duplicate_query_is_rejected(self):
+        report, fixture = self.inputs()
+        report["core"]["rows"][1] = report["core"]["rows"][0]
+        with self.assertRaises(ValueError):
+            score(report, fixture)