feat(oie): report B2 gold-mapping provenance

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:

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,
     }