diff --git a/application/tests/oie_b2_gold_loader_test.py b/application/tests/oie_b2_gold_loader_test.py index 272206985..1412a5275 100644 --- a/application/tests/oie_b2_gold_loader_test.py +++ b/application/tests/oie_b2_gold_loader_test.py @@ -1,6 +1,7 @@ import importlib.util import unittest from pathlib import Path +from types import SimpleNamespace from unittest.mock import MagicMock, patch from application.utils import mapping_fixtures @@ -34,6 +35,148 @@ def test_shared_harnesses_load_from_mapping_fixtures(self) -> None: ) self.assertGreaterEqual(shared, 6) + def test_gold_evidence_preserves_related_expansion_provenance(self) -> None: + mod = _load_run_b2() + entries = mapping_fixtures.load_owasp_mapping_fixture("owasp_top10_2025") + a04 = next(entry for entry in entries if entry["section_id"] == "A04") + + evidence = mod.gold_evidence(a04) + + self.assertEqual(["170-772", "227-045"], evidence["gold_original"]) + self.assertIn("333-858", evidence["gold"]) + self.assertEqual("related_expand_orphan", evidence["gold_remap"]) + self.assertTrue(evidence["gold_is_expanded"]) + + def test_gold_evidence_keeps_unremapped_rows_auditable(self) -> None: + mod = _load_run_b2() + entries = mapping_fixtures.load_owasp_mapping_fixture("owasp_top10_2025") + a01 = next(entry for entry in entries if entry["section_id"] == "A01") + + evidence = mod.gold_evidence(a01) + + self.assertEqual(evidence["gold"], evidence["gold_original"]) + self.assertIsNone(evidence["gold_remap"]) + self.assertFalse(evidence["gold_is_expanded"]) + + def test_score_run_reports_gold_provenance_for_aligned_and_unaligned_rows( + self, + ) -> None: + mod = _load_run_b2() # type: ignore[no-untyped-call] + expanded_harness = { + "label": "Expanded fixture", + "pr": 1, + "fixture_name": "expanded", + } + direct_harness = { + "label": "Direct fixture", + "pr": 2, + "fixture_name": "direct", + } + expanded_row = { + "section_id": "E1", + "cre_ids": ["001-001", "002-002"], + "cre_ids_original": ["001-001"], + "gold_remap": "related_expand_orphan", + } + expanded_unaligned_row = { + "section_id": "E2", + "cre_ids": ["004-004", "005-005"], + "cre_ids_original": ["004-004"], + "gold_remap": "related_expand_orphan", + } + direct_row = {"section_id": "D1", "cre_ids": ["003-003"]} + decision = SimpleNamespace( + chunk_id="chunk-expanded", + envelope={ + "knowledge": {"locator": {"path": "b2/expanded/E1.txt"}}, + "retrieval": {"reranked": [{"cre_id": "uuid-expanded"}]}, + }, + ) + cre_query = MagicMock() + cre_query.all.return_value = [ + SimpleNamespace(id="uuid-expanded", external_id="002-002") + ] + decision_query = MagicMock() + decision_query.filter_by.return_value.all.return_value = [decision] + + with patch("application.cmd.cre_main.db_connect"), patch( + "application.sqla" + ) as sqla, patch.object( + mod, "HARNESSES", [expanded_harness, direct_harness] + ), patch.object( + mod, + "load_gold", + side_effect=[[expanded_row, expanded_unaligned_row], [direct_row]], + ): + sqla.session.query.side_effect = [cre_query, decision_query] + report = mod.score_run("run-1", "sqlite://") + + details = {row["section_id"]: row for row in report["details"]} + self.assertEqual( + { + "gold": ["001-001", "002-002"], + "gold_original": ["001-001"], + "gold_remap": "related_expand_orphan", + "gold_is_expanded": True, + }, + { + key: details["E1"][key] + for key in ( + "gold", + "gold_original", + "gold_remap", + "gold_is_expanded", + ) + }, + ) + self.assertTrue(details["E1"]["aligned"]) + self.assertTrue(details["E1"]["hit"]) + self.assertEqual(["002-002"], details["E1"]["predicted"]) + self.assertEqual( + { + "gold": ["004-004", "005-005"], + "gold_original": ["004-004"], + "gold_remap": "related_expand_orphan", + "gold_is_expanded": True, + }, + { + key: details["E2"][key] + for key in ( + "gold", + "gold_original", + "gold_remap", + "gold_is_expanded", + ) + }, + ) + self.assertFalse(details["E2"]["aligned"]) + self.assertFalse(details["E2"]["hit"]) + self.assertEqual([], details["E2"]["predicted"]) + self.assertEqual( + { + "gold": ["003-003"], + "gold_original": ["003-003"], + "gold_remap": None, + "gold_is_expanded": False, + }, + { + key: details["D1"][key] + for key in ( + "gold", + "gold_original", + "gold_remap", + "gold_is_expanded", + ) + }, + ) + self.assertFalse(details["D1"]["aligned"]) + self.assertFalse(details["D1"]["hit"]) + self.assertEqual([], details["D1"]["predicted"]) + self.assertEqual( + {"expanded_rows": 2, "by_remap": {"related_expand_orphan": 2}}, + report["gold_remap_summary"], + ) + class TestOieB2HarvesterLanggraphKwargs(unittest.TestCase): def test_inner_harvester_accepts_repos_yaml(self) -> None: diff --git a/scripts/oie_owasp_eval/run_b2_pr_mappings.py b/scripts/oie_owasp_eval/run_b2_pr_mappings.py index 6fce15563..555a33361 100644 --- a/scripts/oie_owasp_eval/run_b2_pr_mappings.py +++ b/scripts/oie_owasp_eval/run_b2_pr_mappings.py @@ -325,6 +325,26 @@ def load_gold(harness: Dict[str, Any]) -> List[Dict[str, Any]]: return load_owasp_mapping_fixture(str(harness["fixture_name"])) +def gold_evidence(row: Dict[str, Any]) -> Dict[str, Any]: + """Return score gold plus authored/remap provenance for one fixture row. + + ``gold`` remains the expanded answer key used by B2 scoring. Rows without + an explicit direct mapping report that same list as ``gold_original`` so + every detail record has a stable, auditable shape. + """ + gold = sorted({str(cre_id) for cre_id in row.get("cre_ids") or [] if cre_id}) + original_raw = row.get("cre_ids_original") + original = sorted({str(cre_id) for cre_id in original_raw or [] if cre_id}) or gold + remap = row.get("gold_remap") + remap_name = str(remap) if isinstance(remap, str) and remap else None + return { + "gold": gold, + "gold_original": original, + "gold_remap": remap_name, + "gold_is_expanded": bool(remap_name and set(gold) != set(original)), + } + + def top_k_cre_external_ids( envelope: Dict[str, Any], uuid_to_ext: Dict[str, str], @@ -649,6 +669,8 @@ def score_run(run_id: str, cache: str) -> Dict[str, Any]: details: List[Dict[str, Any]] = [] by_resource: Dict[str, Dict[str, int]] = {} hits = scorable = unaligned = 0 + remap_counts: Dict[str, int] = {} + expanded_rows = 0 for harness in HARNESSES: resource = source_stem(harness) @@ -659,9 +681,15 @@ def score_run(run_id: str, cache: str) -> Dict[str, Any]: ) for row in gold_rows: sid = str(row.get("section_id") or "").strip() - gold = {c for c in (row.get("cre_ids") or []) if c} + evidence = gold_evidence(row) + gold = set(evidence["gold"]) if not gold: continue + if evidence["gold_is_expanded"]: + expanded_rows += 1 + if evidence["gold_remap"]: + remap_name = evidence["gold_remap"] + remap_counts[remap_name] = remap_counts.get(remap_name, 0) + 1 key = f"{resource}::{sid}" suggested = pred.get(key) or set() if key not in pred: @@ -673,7 +701,7 @@ def score_run(run_id: str, cache: str) -> Dict[str, Any]: "pr": harness["pr"], "section_id": sid, "aligned": False, - "gold": sorted(gold), + **evidence, "predicted": [], "hit": False, } @@ -691,7 +719,7 @@ def score_run(run_id: str, cache: str) -> Dict[str, Any]: "pr": harness["pr"], "section_id": sid, "aligned": True, - "gold": sorted(gold), + **evidence, "predicted": sorted(suggested), "hit": hit, } @@ -720,6 +748,10 @@ def score_run(run_id: str, cache: str) -> Dict[str, Any]: "hits": hits, "accuracy": round(rate, 4), "pass_gt_60": rate > 0.60, + "gold_remap_summary": { + "expanded_rows": expanded_rows, + "by_remap": remap_counts, + }, "by_resource": by_resource, "details": details, }