feat(oie): B2 eval harness and CRE neighborhood scorer

application/tests/db_url_test.py

@@ -0,0 +1,31 @@
+import unittest
+
+from application.utils.db_url import redact_db_url
+
+
+class RedactDbUrlTest(unittest.TestCase):
+    def test_masks_password_and_keeps_rest(self) -> None:
+        self.assertEqual(
+            redact_db_url("postgresql://cre:s3cret@127.0.0.1:5432/cre?sslmode=require"),
+            "postgresql://cre:***@127.0.0.1:5432/cre?sslmode=require",
+        )
+
+    def test_url_without_password_is_unchanged(self) -> None:
+        for url in (
+            "postgresql://127.0.0.1/cre",
+            "postgresql://cre@127.0.0.1/cre",
+            "sqlite:///tmp/x.sqlite",
+            "standards_cache.sqlite",
+            "",
+        ):
+            self.assertEqual(redact_db_url(url), url)
+
+    def test_password_with_special_characters(self) -> None:
+        self.assertEqual(
+            redact_db_url("postgres://u:p%40ss:word@db.example.com/x"),
+            "postgres://u:***@db.example.com/x",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

application/tests/librarian/aix_llm_link_seed_test.py

@@ -106,5 +106,43 @@ def retrieve(self, text, *, allowlist=None):
         self.assertEqual(audit.candidates[0].cre_id, "uuid-direct")
 
 
+class GoldRebuildHelpersTest(unittest.TestCase):
+    def test_build_gold_rows_marks_gaps(self) -> None:
+        # Import from script path via runpy-style relative — keep pure helper copy.
+        import importlib.util
+        from pathlib import Path
+
+        path = (
+            Path(__file__).resolve().parents[3]
+            / "scripts"
+            / "oie_owasp_eval"
+            / "rebuild_b2_llm_gold_from_hub.py"
+        )
+        spec = importlib.util.spec_from_file_location("rebuild_b2_llm_gold", path)
+        assert spec and spec.loader
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+
+        hub = {
+            "LLM01": ["686-110", "012-625"],
+            "LLM03": ["701-654"],
+        }
+        existing = {
+            "LLM01": {
+                "section": "Prompt Injection",
+                "hyperlink": "https://example.test/llm01",
+            }
+        }
+        rows, gaps = mod.build_gold_rows(hub, existing=existing)
+        self.assertEqual(len(rows), 10)
+        llm01 = next(r for r in rows if r["section_id"] == "LLM01")
+        self.assertEqual(llm01["cre_ids"], ["012-625", "686-110"])
+        self.assertEqual(llm01["hyperlink"], "https://example.test/llm01")
+        self.assertIn("LLM06", gaps)
+        self.assertIn("LLM09", gaps)
+        empty = next(r for r in rows if r["section_id"] == "LLM06")
+        self.assertEqual(empty["cre_ids"], [])
+
+
 if __name__ == "__main__":
     unittest.main()

application/tests/oie_b2_gold_loader_test.py

@@ -0,0 +1,73 @@
+import importlib.util
+import unittest
+from pathlib import Path
+
+from application.utils import mapping_fixtures
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _load_run_b2():
+    path = ROOT / "scripts" / "oie_owasp_eval" / "run_b2_pr_mappings.py"
+    spec = importlib.util.spec_from_file_location("oie_run_b2_pr_mappings", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestOieB2GoldLoader(unittest.TestCase):
+    def test_shared_harnesses_load_from_mapping_fixtures(self) -> None:
+        mod = _load_run_b2()
+        shared = 0
+        for harness in mod.HARNESSES:
+            gold = mod.load_gold(harness)
+            self.assertIsInstance(gold, list)
+            self.assertGreater(len(gold), 0)
+            if harness.get("local_gold"):
+                continue
+            shared += 1
+            self.assertEqual(
+                gold,
+                mapping_fixtures.load_owasp_mapping_fixture(harness["fixture_name"]),
+            )
+        self.assertGreaterEqual(shared, 6)
+
+
+class TestOieB2SourceIdentityPrefix(unittest.TestCase):
+    def test_k8s_years_emit_standard_and_version(self) -> None:
+        mod = _load_run_b2()
+        h22 = next(
+            h
+            for h in mod.HARNESSES
+            if h.get("fixture_name") == "owasp_kubernetes_top10_2022"
+        )
+        h25 = next(
+            h
+            for h in mod.HARNESSES
+            if h.get("fixture_name") == "owasp_kubernetes_top10_2025"
+        )
+        self.assertEqual(mod.harness_year(h22), 2022)
+        self.assertEqual(mod.harness_year(h25), 2025)
+        row = {"section_id": "K06", "section": "Broken Authentication Mechanisms"}
+        prefix = mod.source_identity_prefix(row, h22)
+        self.assertIn("Standard: OWASP Kubernetes Top 10 2022", prefix)
+        self.assertIn("Version: 2022", prefix)
+        self.assertIn("Section-ID: K06", prefix)
+        # Version before narrative so year_from_text does not pick body years.
+        self.assertLess(prefix.index("Version:"), prefix.index("Section-ID:"))
+
+    def test_strip_and_rewrite_keeps_body(self) -> None:
+        mod = _load_run_b2()
+        harness = {
+            "label": "OWASP Kubernetes Top 10 2025",
+            "fixture_name": "owasp_kubernetes_top10_2025",
+        }
+        row = {"section_id": "K01", "section": "Insecure Workload Configurations"}
+        old = "Source: K01\nSection: Insecure Workload Configurations\nSection-ID: K01\n\nBODY\n"
+        body = mod.strip_source_identity_prefix(old)
+        self.assertEqual(body.strip(), "BODY")
+        rewritten = mod.source_identity_prefix(row, harness) + body
+        self.assertTrue(rewritten.startswith("Standard:"))
+        self.assertIn("Version: 2025", rewritten)
+        self.assertIn("BODY", rewritten)

application/tests/oie_b2_hop_distance_test.py

@@ -0,0 +1,126 @@
+"""Unit tests for B2 CRE hop-distance classification (no DB)."""
+
+from __future__ import annotations
+
+import importlib.util
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _load_hop():
+    import sys
+
+    path = ROOT / "scripts" / "oie_owasp_eval" / "hop_distance.py"
+    name = "oie_hop_distance"
+    if name in sys.modules:
+        return sys.modules[name]
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestOieB2HopDistance(unittest.TestCase):
+    def setUp(self) -> None:
+        self.hop = _load_hop()
+        # Graph:
+        #   parent --Contains--> child --Related--> cousin
+        #   island (known, disconnected)
+        edges = [
+            self.hop.TypedEdge("parent", "child", "CONTAINS_DOWN"),
+            self.hop.TypedEdge("child", "parent", "CONTAINS_UP"),
+            self.hop.TypedEdge("child", "cousin", "RELATED"),
+            self.hop.TypedEdge("cousin", "child", "RELATED"),
+        ]
+        self.adj = self.hop.CreAdjacency(
+            known_ids=frozenset({"parent", "child", "cousin", "island"}),
+            edges=edges,
+        )
+
+    def test_exact(self) -> None:
+        row = self.hop.classify_section(
+            gold=["child"], predicted=["child", "999-999"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "exact")
+        self.assertEqual(row["hop_count"], 0)
+
+    def test_one_hop_related(self) -> None:
+        row = self.hop.classify_section(
+            gold=["child"], predicted=["cousin"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "one_hop_related")
+        self.assertEqual(row["hop_count"], 1)
+        self.assertEqual(row["path_penalty"], 2)
+
+    def test_one_hop_contains_down(self) -> None:
+        row = self.hop.classify_section(
+            gold=["child"], predicted=["parent"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "one_hop_contains")
+        self.assertEqual(row["hop_count"], 1)
+        # parent -> child is CONTAINS_DOWN from parent? Wait: predicted=parent, gold=child
+        # path parent -> child is CONTAINS_DOWN, penalty 1
+        self.assertEqual(row["path_penalty"], 1)
+
+    def test_one_hop_contains_up(self) -> None:
+        row = self.hop.classify_section(
+            gold=["parent"], predicted=["child"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "one_hop_contains")
+        self.assertEqual(row["hop_count"], 1)
+        self.assertEqual(row["path_penalty"], 2)  # CONTAINS_UP
+
+    def test_multi_hop(self) -> None:
+        row = self.hop.classify_section(
+            gold=["parent"], predicted=["cousin"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "multi_hop")
+        self.assertEqual(row["hop_count"], 2)
+
+    def test_hallucination_invented_only(self) -> None:
+        row = self.hop.classify_section(
+            gold=["child"], predicted=["zzz-001", "zzz-002"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "hallucination")
+        self.assertEqual(sorted(row["invented"]), ["zzz-001", "zzz-002"])
+
+    def test_irrelevant_known_disconnected(self) -> None:
+        row = self.hop.classify_section(
+            gold=["child"], predicted=["island"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "irrelevant")
+
+    def test_mixed_known_beats_invented(self) -> None:
+        row = self.hop.classify_section(
+            gold=["child"], predicted=["zzz-001", "cousin"], adj=self.adj
+        )
+        self.assertEqual(row["bucket"], "one_hop_related")
+        self.assertIn("zzz-001", row["invented"])
+
+    def test_aggregate_percentages(self) -> None:
+        details = [
+            {"gold": ["child"], "predicted": ["child"], "aligned": True},
+            {"gold": ["child"], "predicted": ["cousin"], "aligned": True},
+            {"gold": ["child"], "predicted": ["parent"], "aligned": True},
+            {"gold": ["parent"], "predicted": ["cousin"], "aligned": True},
+            {"gold": ["child"], "predicted": ["zzz-9"], "aligned": True},
+            {"gold": ["child"], "predicted": ["island"], "aligned": True},
+        ]
+        summary = self.hop.score_details(details, self.adj)
+        self.assertEqual(summary["scorable"], 6)
+        self.assertEqual(summary["counts"]["exact"], 1)
+        self.assertEqual(summary["counts"]["one_hop_related"], 1)
+        self.assertEqual(summary["counts"]["one_hop_contains"], 1)
+        self.assertEqual(summary["counts"]["multi_hop"], 1)
+        self.assertEqual(summary["counts"]["hallucination"], 1)
+        self.assertEqual(summary["counts"]["irrelevant"], 1)
+        self.assertAlmostEqual(summary["rates"]["exact"], 1 / 6, places=4)
+        self.assertAlmostEqual(summary["rates"]["distance1_hit"], 3 / 6, places=4)
+
+
+if __name__ == "__main__":
+    unittest.main()

application/tests/oie_eval_scripts_test.py

@@ -0,0 +1,78 @@
+import importlib.util
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[2]
+EVAL_DIR = ROOT / "scripts" / "oie_owasp_eval"
+
+
+def _load(name: str):
+    spec = importlib.util.spec_from_file_location(
+        f"oie_eval_{name}", EVAL_DIR / f"{name}.py"
+    )
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class MatchGoldTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.mod = _load("score_b1_opencre_api")
+        self.gold = {
+            "authentication cheat sheet": {"cre-authn"},
+            "multifactor authentication cheat sheet": {"cre-mfa"},
+        }
+
+    def test_fuzzy_match_bleeds_into_longer_sheet_name(self) -> None:
+        got = self.mod.match_gold({"authentication cheat sheet"}, self.gold)
+        self.assertEqual(got, {"cre-authn", "cre-mfa"})
+
+    def test_exact_match_keeps_sheets_separate(self) -> None:
+        got = self.mod.match_gold({"authentication cheat sheet"}, self.gold, exact=True)
+        self.assertEqual(got, {"cre-authn"})
+
+
+class ClaimComboTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.mod = _load("run_exhaustive_switch_grid")
+        self.tmp = tempfile.TemporaryDirectory()
+        grid = Path(self.tmp.name)
+        patcher_grid = patch.object(self.mod, "GRID_DIR", grid)
+        patcher_locks = patch.object(self.mod, "LOCKS_DIR", grid / ".locks")
+        patcher_grid.start()
+        patcher_locks.start()
+        self.addCleanup(patcher_grid.stop)
+        self.addCleanup(patcher_locks.stop)
+        self.addCleanup(self.tmp.cleanup)
+        self.grid = grid
+
+    def _write_ok(self, combo_id: str) -> None:
+        (self.grid / f"{combo_id}.json").write_text(
+            json.dumps({"status": "ok", "headline": {"hits": 1}})
+        )
+
+    def test_completed_combo_is_skipped_unless_forced(self) -> None:
+        self._write_ok("c1")
+        self.assertFalse(self.mod.claim_combo("c1"))
+        self.assertTrue(self.mod.claim_combo("c1", force=True))
+
+    def test_live_claim_blocks_second_worker(self) -> None:
+        self.assertTrue(self.mod.claim_combo("c2"))
+        self.assertFalse(self.mod.claim_combo("c2"))
+        self.assertFalse(self.mod.claim_combo("c2", force=True))
+
+    def test_claim_left_by_dead_worker_is_reclaimed(self) -> None:
+        lock = self.grid / ".locks" / "c3"
+        lock.mkdir(parents=True)
+        (lock / "pid").write_text("999999999\n")
+        self.assertTrue(self.mod.claim_combo("c3"))
+        self.assertEqual((lock / "pid").read_text().strip(), str(os.getpid()))
+
+
+if __name__ == "__main__":
+    unittest.main()

application/utils/db_url.py

@@ -0,0 +1,22 @@
+"""Helpers for handling database URLs in logs and reports."""
+
+from __future__ import annotations
+
+from urllib.parse import urlsplit, urlunsplit
+
+
+def redact_db_url(url: str) -> str:
+    """Replace the password in ``scheme://user:password@host/db`` with ``***``.
+
+    Anything that is not a URL with credentials (file paths, bare names) is
+    returned unchanged.
+    """
+    try:
+        parts = urlsplit(url)
+    except ValueError:
+        return "<unparseable database url>"
+    userinfo, sep, hostport = parts.netloc.rpartition("@")
+    if not sep or ":" not in userinfo:
+        return url
+    user = userinfo.split(":", 1)[0]
+    return urlunsplit(parts._replace(netloc=f"{user}:***@{hostport}"))