diff --git a/application/tests/librarian/dataset_test.py b/application/tests/librarian/dataset_test.py index d7ee3284d..bb497b719 100644 --- a/application/tests/librarian/dataset_test.py +++ b/application/tests/librarian/dataset_test.py @@ -111,10 +111,19 @@ class TestLoadDatasetRejectsDuplicateIds(unittest.TestCase): """ def _load_harness(self): + # The harness is a standalone script, not an importable module, so it is + # loaded by path. spec_from_file_location returns None when the path is + # not there, and a spec without a loader is possible too. The harness is + # committed, so either case is a real breakage rather than an absent + # optional dependency — raise with the path instead of skipping (a skip + # would report success for a harness that never ran) and instead of an + # AttributeError on None, which names neither the file nor the cause. import importlib.util path = os.path.join(_REPO_ROOT, "scripts", "evaluate_librarian.py") spec = importlib.util.spec_from_file_location("evaluate_librarian", path) + if spec is None or spec.loader is None: + raise ImportError(f"cannot load the eval harness from {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module diff --git a/application/tests/librarian/decision_engine_test.py b/application/tests/librarian/decision_engine_test.py new file mode 100644 index 000000000..f9e1e630e --- /dev/null +++ b/application/tests/librarian/decision_engine_test.py @@ -0,0 +1,99 @@ +"""Hermetic tests for C.4 — the decision engine (Week 6). + +Table-driven over every (confidence, candidates, flag) combination the rule can +see, plus reason-code precedence and the input guards. No key, DB, or model. +""" + +import dataclasses +import math +import unittest + +from application.utils.librarian.decision_engine import ( + ENGINE_NAME, + DecisionError, + DecisionResult, + decide, +) +from application.utils.librarian.schemas import Decision, ReasonCode + +TAU = 0.8 +CANDS = ("616-305", "764-507", "611-909") + + +class DecideTest(unittest.TestCase): + def test_links_when_confident_and_unflagged(self): + r = decide(0.95, CANDS, threshold=TAU) + self.assertEqual(r.decision, Decision.linked) + self.assertIsNone(r.reason_code) + self.assertEqual(r.cre_ids, ("616-305",)) # only the top-1 is linked + + def test_confidence_exactly_at_threshold_links(self): + # link iff confidence >= threshold — the boundary is inclusive. + r = decide(TAU, CANDS, threshold=TAU) + self.assertEqual(r.decision, Decision.linked) + self.assertIsNone(r.reason_code) + + def test_just_below_threshold_reviews(self): + r = decide(TAU - 1e-9, CANDS, threshold=TAU) + self.assertEqual(r.decision, Decision.review) + self.assertEqual(r.reason_code, ReasonCode.below_threshold) + self.assertEqual(r.cre_ids, ("616-305",)) # best-guess suggestion kept + + def test_no_candidates_reviews_even_when_confident(self): + r = decide(0.99, (), threshold=TAU) + self.assertEqual(r.decision, Decision.review) + self.assertEqual(r.reason_code, ReasonCode.no_candidates) + self.assertEqual(r.cre_ids, ()) # nothing to suggest + + def test_adversarial_flag_reviews_even_when_confident(self): + r = decide(0.99, CANDS, threshold=TAU, adversarial=True) + self.assertEqual(r.decision, Decision.review) + self.assertEqual(r.reason_code, ReasonCode.adversarial_flag) + + def test_update_ambiguous_flag_reviews_even_when_confident(self): + r = decide(0.99, CANDS, threshold=TAU, update_ambiguous=True) + self.assertEqual(r.decision, Decision.review) + self.assertEqual(r.reason_code, ReasonCode.update_ambiguous) + + def test_precedence_no_candidates_beats_everything(self): + # empty shortlist + a flag + high confidence -> still NO_CANDIDATES. + r = decide(0.99, (), threshold=TAU, adversarial=True, update_ambiguous=True) + self.assertEqual(r.reason_code, ReasonCode.no_candidates) + + def test_precedence_adversarial_beats_below_threshold(self): + r = decide(0.10, CANDS, threshold=TAU, adversarial=True) + self.assertEqual(r.reason_code, ReasonCode.adversarial_flag) + + def test_precedence_adversarial_beats_update_ambiguous(self): + r = decide(0.99, CANDS, threshold=TAU, adversarial=True, update_ambiguous=True) + self.assertEqual(r.reason_code, ReasonCode.adversarial_flag) + + def test_confidence_is_carried_through(self): + for conf in (0.0, 0.42, 0.8, 1.0): + self.assertEqual(decide(conf, CANDS, threshold=TAU).confidence, conf) + + +class GuardTest(unittest.TestCase): + def test_bad_threshold_rejected(self): + for bad in (-0.1, 1.1, math.nan, math.inf): + with self.assertRaises(DecisionError): + decide(0.5, CANDS, threshold=bad) + + def test_bad_confidence_rejected(self): + for bad in (-0.1, 1.1, math.nan, math.inf): + with self.assertRaises(DecisionError): + decide(bad, CANDS, threshold=TAU) + + +class ResultTest(unittest.TestCase): + def test_engine_name_is_versioned(self): + self.assertRegex(ENGINE_NAME, r"^decision-engine/\d+\.\d+\.\d+$") + + def test_result_is_frozen(self): + r = decide(0.95, CANDS, threshold=TAU) + with self.assertRaises(dataclasses.FrozenInstanceError): + r.confidence = 0.1 # type: ignore[misc] + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/evaluate_harness_test.py b/application/tests/librarian/evaluate_harness_test.py new file mode 100644 index 000000000..9c2c59356 --- /dev/null +++ b/application/tests/librarian/evaluate_harness_test.py @@ -0,0 +1,259 @@ +"""Hermetic tests for the live-report plumbing in ``scripts/evaluate_librarian.py``. + +The live reports (recall/top-1, the C.3 ECE gate, the C.4 decision accuracy) only +run under ``--use_live_embeddings``, which needs a populated DB, an embedding +model, and the cross-encoder — so nothing exercised their wiring. That is exactly +the code that has to share one retrieve+rerank pass and one fitted ``T`` across +three reports, so the sharing is asserted here against stub seams instead: + +- ``live_audits`` must call the pipeline once per row, never once per report. +- ``calibration_set`` must draw only the positive + hard_negative slices. +- ``report_calibration`` must hand back the fitted scaler, and must fail (status + 1, no scaler) on a degenerate set rather than reporting success. +- ``report_decision_accuracy`` must consume that scaler and the shared audits + without touching the retriever or reranker again. +""" + +import importlib.util +import os +import unittest +from typing import List, Optional + +from application.utils.librarian.schemas import CreCandidate, RetrievalAudit + +# The harness is a standalone script, not an importable package module. +_HARNESS_PATH = os.path.join( + os.path.dirname(__file__), "..", "..", "..", "scripts", "evaluate_librarian.py" +) +_spec = importlib.util.spec_from_file_location("evaluate_librarian", _HARNESS_PATH) +assert _spec and _spec.loader +harness = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(harness) + + +def _golden_row( + row_id: str, + slice_name: str, + text: str, + cre_ids: List[str], + reason_code: Optional[str] = None, +): + """Build a GoldenDatasetRow through the real validator, not a stub. + + ``expected.decision`` is required by the schema, and ``linked`` requires + ``cre_ids`` while ``review`` requires a ``reason_code``, so both are derived: + rows with expected ids are linked, rows without route to review below the bar. + """ + from application.utils.librarian.schemas import GoldenDatasetRow + + expected: dict = { + "decision": "linked" if cre_ids else "review", + "cre_ids": cre_ids or None, + } + if not cre_ids: + expected["reason_code"] = reason_code or "BELOW_THRESHOLD" + elif reason_code is not None: + expected["reason_code"] = reason_code + source_input: dict = {"text": text, "source_standard": "ASVS"} + if slice_name == "explicit": + # The schema ties the explicit slice to a cited CRE id. + source_input["explicit_cre_ref"] = (cre_ids or ["616-305"])[0] + return GoldenDatasetRow.model_validate( + { + "id": row_id, + "schema_version": "0.1.0", + "slice": slice_name, + "input": source_input, + "expected": expected, + "provenance": { + "section_path": f"{row_id}.md", + "ground_truth_source": "synthesised for the harness plumbing tests", + }, + } + ) + + +class CountingPipeline: + """Stub retriever+reranker that records how many passes it was asked for.""" + + def __init__(self, shortlists): + # shortlists: row text -> list of (cre_id, logit), best first + self._shortlists = shortlists + self.retrieve_calls = 0 + self.rerank_calls = 0 + + def retrieve(self, text: str) -> RetrievalAudit: + self.retrieve_calls += 1 + pairs = self._shortlists.get(text, []) + return RetrievalAudit( + retriever="stub/1.0.0", + candidates=[CreCandidate(cre_id=c, score_vector=0.5) for c, _ in pairs], + reranked=[], + threshold=0.0, + ) + + def rerank(self, text: str, audit: RetrievalAudit) -> RetrievalAudit: + self.rerank_calls += 1 + pairs = self._shortlists.get(text, []) + return audit.model_copy( + update={ + "reranked": [ + CreCandidate(cre_id=c, score_rerank=logit) for c, logit in pairs + ] + } + ) + + +class LiveAuditsTest(unittest.TestCase): + def test_pipeline_runs_once_per_row_not_once_per_report(self) -> None: + rows = [ + _golden_row("p1", "positive", "alpha", ["616-305"]), + _golden_row("n1", "hard_negative", "beta", []), + ] + pipe = CountingPipeline( + {"alpha": [("616-305", 4.0)], "beta": [("111-111", 3.0)]} + ) + + audits = harness.live_audits(rows, pipe, pipe) + + self.assertEqual(pipe.retrieve_calls, 2) + self.assertEqual(pipe.rerank_calls, 2) + self.assertEqual(set(audits), {"p1", "n1"}) + + # Three reports read the same audits; none of them may re-run the pipeline. + harness.report_retrieval_recall(rows, audits, 10, 5) + _status, scaler = harness.report_calibration(rows, audits) + self.assertIsNotNone(scaler) + harness.report_decision_accuracy(rows, audits, scaler, 0.80) + self.assertEqual(pipe.retrieve_calls, 2) + self.assertEqual(pipe.rerank_calls, 2) + + +class CalibrationSetTest(unittest.TestCase): + def test_draws_only_the_two_calibration_slices(self) -> None: + rows = [ + _golden_row("p1", "positive", "alpha", ["616-305"]), + _golden_row("n1", "hard_negative", "beta", []), + _golden_row("a1", "ambiguous", "gamma", ["616-305"]), + _golden_row("e1", "explicit", "delta", ["616-305"]), + ] + pipe = CountingPipeline( + { + "alpha": [("616-305", 4.0)], + "beta": [("111-111", 3.0)], + "gamma": [("616-305", 2.0)], + "delta": [("616-305", 1.0)], + } + ) + audits = harness.live_audits(rows, pipe, pipe) + + logit_sets, labels = harness.calibration_set(rows, audits) + + # ambiguous/explicit rows are audited but must not enter the fit. + self.assertEqual(len(logit_sets), 2) + self.assertEqual(sorted(labels), [0.0, 1.0]) + + def test_skips_rows_with_no_audit_and_empty_shortlists(self) -> None: + rows = [ + _golden_row("p1", "positive", "alpha", ["616-305"]), + _golden_row("p2", "positive", "empty", ["616-305"]), + _golden_row("n1", "hard_negative", "beta", []), + ] + pipe = CountingPipeline( + {"alpha": [("616-305", 4.0)], "beta": [("111-111", 3.0)]} + ) + # "empty" yields no candidates; p3 is never audited at all. + audits = harness.live_audits(rows, pipe, pipe) + + logit_sets, labels = harness.calibration_set(rows, audits) + self.assertEqual(len(logit_sets), 2) + self.assertEqual(len(labels), 2) + + +class ReportCalibrationTest(unittest.TestCase): + def test_returns_status_and_fitted_scaler(self) -> None: + rows = [ + _golden_row("p1", "positive", "alpha", ["616-305"]), + _golden_row("p2", "positive", "alpha2", ["616-305"]), + _golden_row("n1", "hard_negative", "beta", []), + _golden_row("n2", "hard_negative", "beta2", []), + ] + pipe = CountingPipeline( + { + "alpha": [("616-305", 5.0), ("999-999", 0.1)], + "alpha2": [("616-305", 4.0), ("999-999", 0.2)], + "beta": [("111-111", 3.0), ("222-222", 2.9)], + "beta2": [("111-111", 2.0), ("222-222", 1.9)], + } + ) + audits = harness.live_audits(rows, pipe, pipe) + + status, scaler = harness.report_calibration(rows, audits) + + self.assertIn(status, (0, 1)) # gate outcome depends on the stub logits + self.assertIsNotNone(scaler) + self.assertGreater(scaler.temperature, 0.0) + + def test_degenerate_set_fails_and_yields_no_scaler(self) -> None: + # Single-class labels: every top-1 is correct, so T is unidentifiable. + rows = [ + _golden_row("p1", "positive", "alpha", ["616-305"]), + _golden_row("p2", "positive", "alpha2", ["616-305"]), + ] + pipe = CountingPipeline( + {"alpha": [("616-305", 5.0)], "alpha2": [("616-305", 4.0)]} + ) + audits = harness.live_audits(rows, pipe, pipe) + + status, scaler = harness.report_calibration(rows, audits) + + self.assertEqual(status, 1, "a skipped gate must not report success") + self.assertIsNone(scaler) + + +class ReportDecisionAccuracyTest(unittest.TestCase): + def test_grades_expected_decision_rows_off_shared_audits(self) -> None: + from application.utils.librarian.calibration.temperature import ( + TemperatureScaler, + ) + + rows = [ + _golden_row("d1", "positive", "alpha", ["616-305"]), + _golden_row( + "d2", "hard_negative", "beta", [], reason_code="BELOW_THRESHOLD" + ), + ] + pipe = CountingPipeline( + { + # A dominant top-1 clears tau; a near-tie falls below it. + "alpha": [("616-305", 20.0), ("999-999", 0.0)], + "beta": [("111-111", 1.0), ("222-222", 0.99)], + } + ) + audits = harness.live_audits(rows, pipe, pipe) + before = (pipe.retrieve_calls, pipe.rerank_calls) + + status = harness.report_decision_accuracy( + rows, audits, TemperatureScaler(1.0), 0.80 + ) + + self.assertEqual(status, 0, "the C.4 report is informational, never a gate") + self.assertEqual((pipe.retrieve_calls, pipe.rerank_calls), before) + + def test_no_graded_rows_is_not_an_error(self) -> None: + from application.utils.librarian.calibration.temperature import ( + TemperatureScaler, + ) + + rows = [_golden_row("p1", "positive", "alpha", ["616-305"])] + pipe = CountingPipeline({"alpha": [("616-305", 4.0)]}) + audits = harness.live_audits(rows, pipe, pipe) + + status = harness.report_decision_accuracy( + rows, audits, TemperatureScaler(1.0), 0.80 + ) + self.assertEqual(status, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/librarian/__init__.py b/application/utils/librarian/__init__.py index 1a23cc3cb..869b807fa 100644 --- a/application/utils/librarian/__init__.py +++ b/application/utils/librarian/__init__.py @@ -18,7 +18,10 @@ W4 (C.2): cross-encoder reranker — re-sorts the C.1 shortlist, fills reranked[]. W5 (C.3): confidence calibration — temperature scaling maps a rerank logit to an honest probability (fit by NLL on the golden set, gated ECE < 0.10). -Decision routing (C.4, W6) onward is not built yet. + W6 (C.4): decision engine — thresholds the calibrated confidence to auto-link + (LinkProposal) or route to human review (ReviewItem), with a reason. +Envelope emitter + pipeline glue (C.4, W6b) and the queue/graph writers (W8) are +not built yet. Vendored RFC JSON schemas live under ``_rfc_schemas/``. They are pinned to upstream/owasp-graph @ 2b1437987768d5ed20fe9ee721ab9a898c4b84af (PR #734). diff --git a/application/utils/librarian/calibration/__init__.py b/application/utils/librarian/calibration/__init__.py index 305586a76..54b774038 100644 --- a/application/utils/librarian/calibration/__init__.py +++ b/application/utils/librarian/calibration/__init__.py @@ -1,10 +1,12 @@ """Module C.3 — confidence calibration (Week 5). C.2 (the cross-encoder, W4) emits a raw ranking logit per candidate — great for -ordering, meaningless as confidence (a +1.5 is not "82% sure"). C.3 turns that -logit into an honest probability via **temperature scaling**: ``p = sigmoid(z/T)`` -with a single scalar ``T`` fit by negative-log-likelihood on the golden set, and -proves the result honest with **ECE < 0.10**. +ordering, meaningless as confidence (a +1.5 is not "82% sure"). C.3 turns those +logits into an honest probability by calibrating the **softmax over the whole +shortlist**, ``p = softmax(logits / T)``, with the confidence being the top-1 +candidate's mass — a single scalar ``T`` fit by negative-log-likelihood on the +golden set. It proves the result honest with **ECE < 0.10**. (Calibrating the +single top-1 logit with ``sigmoid(z/T)`` cannot work; see ``temperature.py``.) The W6 decision engine thresholds that probability (auto-link vs. human review), so calibration is what makes the threshold trustworthy. Kept dependency-light diff --git a/application/utils/librarian/decision_engine.py b/application/utils/librarian/decision_engine.py new file mode 100644 index 000000000..c34a3ef6e --- /dev/null +++ b/application/utils/librarian/decision_engine.py @@ -0,0 +1,99 @@ +"""Module C.4 — the decision engine (Week 6). The gatekeeper. + +C.3 hands up one calibrated confidence: "how likely is the top reranked candidate +the correct CRE?" C.4 turns that honest number into an action — **auto-link** the +chunk into the OpenCRE graph, or **route it to a human** for review. That choice is +the accuracy gate of the whole pipeline, so the rule is deliberately small and total: + + - no candidate at all -> review (NO_CANDIDATES) + - a blocking safety flag -> review (ADVERSARIAL_FLAG / UPDATE_AMBIGUOUS) + - confidence below the threshold -> review (BELOW_THRESHOLD) + - otherwise -> auto-link the top-1 candidate + +Like C.1/C.2/C.3 this is a thin, model-free seam: a pure function of +``(confidence, candidates, flags, threshold)`` -> ``DecisionResult``. It does **not** +import the C.3 ``TemperatureScaler`` — it consumes the confidence that scaler already +produced — so it is hermetically testable and agnostic to how the number was made. +Turning a ``DecisionResult`` into the RFC ``LinkProposal`` / ``ReviewItem`` envelope +(which needs the full chunk context) is the emitter's job, wired in the pipeline. + +Reason-code precedence when several conditions hold at once: +``NO_CANDIDATES > ADVERSARIAL_FLAG > UPDATE_AMBIGUOUS > BELOW_THRESHOLD``. A safety +flag is surfaced to the reviewer ahead of a mere low-confidence note, because it is +the more important thing for a human to see; you cannot link nothing, so the empty +shortlist dominates everything. +""" + +import math +from dataclasses import dataclass +from typing import Optional, Sequence, Tuple + +from application.utils.librarian.schemas import Decision, ReasonCode + +# Identify the engine in the RFC audit trail (mirrors RETRIEVER_NAME / +# RERANKER_NAME / CALIBRATOR_NAME). +ENGINE_NAME = "decision-engine/0.1.0" + + +class DecisionError(ValueError): + """Raised on decision-engine misuse (bad threshold or confidence).""" + + +@dataclass(frozen=True) +class DecisionResult: + """The verdict for one chunk. Frozen so it is a stable, loggable value. + + ``cre_ids`` is the top-1 candidate: the CRE that gets linked when + ``decision == linked``, or the reviewer's best-guess suggestion when + ``decision == review`` (empty only when there were no candidates at all). + ``reason_code`` is set iff ``decision == review``. + """ + + decision: Decision + confidence: float + cre_ids: Tuple[str, ...] + reason_code: Optional[ReasonCode] = None + + +def _validate(confidence: float, threshold: float) -> None: + if not math.isfinite(threshold) or not 0.0 <= threshold <= 1.0: + raise DecisionError(f"threshold must be finite in [0, 1], got {threshold}") + if not math.isfinite(confidence) or not 0.0 <= confidence <= 1.0: + raise DecisionError(f"confidence must be finite in [0, 1], got {confidence}") + + +def decide( + confidence: float, + candidate_cre_ids: Sequence[str], + *, + threshold: float, + adversarial: bool = False, + update_ambiguous: bool = False, +) -> DecisionResult: + """Apply the auto-link rule to one chunk's calibrated confidence. + + ``candidate_cre_ids`` is the reranked shortlist, best first (may be empty). + ``threshold`` is the auto-link bar τ (``LibrarianConfig.link_threshold``); a + chunk links only when ``confidence >= threshold``. ``adversarial`` / + ``update_ambiguous`` are blocking flags from the SafetyGuard (both default + False until it is wired) — either one forces review regardless of confidence. + """ + _validate(confidence, threshold) + + top = tuple(candidate_cre_ids[:1]) + + if not candidate_cre_ids: + return DecisionResult(Decision.review, confidence, (), ReasonCode.no_candidates) + if adversarial: + return DecisionResult( + Decision.review, confidence, top, ReasonCode.adversarial_flag + ) + if update_ambiguous: + return DecisionResult( + Decision.review, confidence, top, ReasonCode.update_ambiguous + ) + if confidence < threshold: + return DecisionResult( + Decision.review, confidence, top, ReasonCode.below_threshold + ) + return DecisionResult(Decision.linked, confidence, top, None) diff --git a/scripts/evaluate_librarian.py b/scripts/evaluate_librarian.py index 560693a2c..66ee69b1f 100644 --- a/scripts/evaluate_librarian.py +++ b/scripts/evaluate_librarian.py @@ -1,8 +1,8 @@ #!/usr/bin/env python -"""Module C regression harness — Week 2: C.0 deterministic input boundary. +"""Module C regression harness — C.0 through C.4 over the golden set. -On top of the W1 skeleton (golden dataset + scorer + TRACT hub-firewall), -the harness now runs every golden row through the C.0 boundary: +On top of the W1 skeleton (golden dataset + scorer + TRACT hub-firewall), every +golden row runs through the C.0 boundary, and these two reports are offline: 1. SectionValidator — each row is adapted to a synthetic knowledge_queue row and must validate into an internal ``Section``; the harness prints the @@ -10,8 +10,19 @@ 2. ExplicitLinkResolver — sections citing a CRE id resolve deterministically (no ML); the explicit slice is gated at 100% correctness. -The semantic path (retriever W3, cross-encoder W4) is still stubbed: rows -without an explicit reference yield no predictions. +The semantic path needs ``--use_live_embeddings``, because there is no honest +offline value: the candidate pool must be the real CRE-node vectors, and seeding +it from golden text is the leakage the hub firewall exists to strip. Under that +flag the run retrieves and reranks each row once (``live_audits``) and three +reports share those shortlists: + +3. C.1 retrieval recall@k and C.2 rerank top-1 over the positive slice. +4. C.3 temperature calibration — fits one ``T`` and gates on ECE < 0.10. This is + the only live report that sets the exit status: a failed *or* skipped gate + returns nonzero, so a live run cannot pass without calibration having run. +5. C.4 decision accuracy — thresholds that same fitted ``T`` through ``decide()``. + Informational only, since the SafetyGuard flags are not wired until W8 and + tuning tau is the W7 experiment. """ import argparse @@ -19,7 +30,7 @@ import os import sys from collections import Counter -from typing import Any, Dict, List, Set +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple # Bootstrap project root onto sys.path so this runs as a standalone script. sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) @@ -38,6 +49,9 @@ section_from_queue_row, ) +if TYPE_CHECKING: # the live calibration deps are imported lazily below + from application.utils.librarian.calibration.temperature import TemperatureScaler + # Harness-only synthetic provenance: golden rows are not queue rows, so we # synthesize the minimum B-shaped row needed to exercise the C.0 boundary. _SYNTHETIC_SHA = "0" * 40 @@ -121,9 +135,9 @@ def _build_live_pipeline( Live deps are imported lazily so the offline harness needs neither a DB, an embedding model, nor the cross-encoder stack. Called once per run from - ``main`` and shared by every live report (recall/top-1 and calibration), so - the heavy hub + model load happens a single time and the id-space translation - stays in one place. + ``main`` and shared by every live report (recall/top-1, the C.3 ECE gate, and + the C.4 decision accuracy), so the heavy hub + model load happens a single + time and the id-space translation stays in one place. """ from application.cmd.cre_main import db_connect from application.defs import cre_defs @@ -249,34 +263,26 @@ def report_retrieval_recall( ) -def report_calibration( +def calibration_set( rows: List[GoldenDatasetRow], audits: Dict[str, Any], -) -> int: - """Fit temperature on the golden set and report the Week 5 ECE gate (< 0.10). +) -> Tuple[List[List[float]], List[float]]: + """The (shortlist, is-top1-correct) pairs temperature is fit on. - Reads the shared ``audits`` from ``live_audits``, the same shortlists - ``report_retrieval_recall`` scores, so the positive slice is not retrieved and - reranked a second time just to calibrate on it. Builds a - (shortlist, label) calibration set from the live C.1 -> C.2 pipeline over the - positive + hard_negative slices: each row's *reranked shortlist* of logits, - labelled 1 iff its top-1 candidate is an expected CRE (hard_negatives expect - none, so they contribute the 0 class). Both slices are needed so the fit sees - both outcomes (else it is degenerate). Confidence is the top-1 mass of - softmax(logits / T); prints ECE at T=1 vs the fitted T and PASS/FAIL on - ECE < 0.10. Returns 1 on a failed gate, and also on a degenerate calibration - set, so a live run can never exit 0 without the gate actually having run. - """ - from application.utils.librarian.calibration.temperature import ( - TemperatureScaler, - expected_calibration_error, - fit_temperature, - ) + Drawn from the positive + hard_negative slices: each row contributes its + *reranked shortlist* of logits, labelled 1 iff its top-1 candidate is an + expected CRE (hard_negatives expect none, so they supply the 0 class). Both + slices are needed or the fit is degenerate. - cal_rows = [r for r in rows if r.slice.value in ("positive", "hard_negative")] + Split out so the C.3 gate and the C.4 decision report derive the calibration + set exactly once from the same shared audits, rather than each rebuilding it + and fitting its own ``T`` off a separate rerank pass. + """ logit_sets: List[List[float]] = [] labels: List[float] = [] - for row in cal_rows: + for row in rows: + if row.slice.value not in ("positive", "hard_negative"): + continue audit = audits.get(row.id) if audit is None: continue @@ -286,6 +292,34 @@ def report_calibration( expected = set(row.expected.cre_ids or []) logit_sets.append([float(c.score_rerank) for c in reranked]) labels.append(1.0 if reranked[0].cre_id in expected else 0.0) + return logit_sets, labels + + +def report_calibration( + rows: List[GoldenDatasetRow], + audits: Dict[str, Any], +) -> Tuple[int, Optional["TemperatureScaler"]]: + """Fit temperature on the golden set and report the Week 5 ECE gate (< 0.10). + + Reads the shared ``audits`` from ``live_audits``, the same shortlists + ``report_retrieval_recall`` scores, so the positive slice is not retrieved and + reranked a second time just to calibrate on it. Confidence is the top-1 mass + of softmax(logits / T); prints ECE at T=1 vs the fitted T and PASS/FAIL on + ECE < 0.10. + + Returns ``(status, scaler)``. Status is 1 on a failed gate, and also on a + degenerate calibration set, so a live run can never exit 0 without the gate + actually having run. The fitted scaler is handed back (``None`` when the set + was degenerate) so the C.4 report thresholds on this same ``T`` instead of + fitting its own. + """ + from application.utils.librarian.calibration.temperature import ( + TemperatureScaler, + expected_calibration_error, + fit_temperature, + ) + + logit_sets, labels = calibration_set(rows, audits) if len(set(labels)) < 2: # Degenerate calibration set: single-class labels, or nothing left after @@ -298,7 +332,7 @@ def report_calibration( f"{len(labels)} row(s) covering {len(set(labels))} class(es); " "FAILED (gate did not run)" ) - return 1 + return 1, None scaler = fit_temperature(logit_sets, labels) conf_raw = [TemperatureScaler(1.0).confidence(s) for s in logit_sets] @@ -311,7 +345,85 @@ def report_calibration( f"ECE {ece_raw:.3f} (raw, T=1) -> {ece_cal:.3f} (calibrated); " f"gate ECE<0.10: {'PASS' if gate_ok else 'FAIL'}" ) - return 0 if gate_ok else 1 + return (0 if gate_ok else 1), scaler + + +def report_decision_accuracy( + rows: List[GoldenDatasetRow], + audits: Dict[str, Any], + scaler: "TemperatureScaler", + threshold: float, +) -> int: + """Run the full C.1 -> C.4 decision over the golden set and measure how often + ``decide()`` lands on the expected auto-link-vs-review call. + + Takes the ``scaler`` already fitted by ``report_calibration`` and the shared + ``audits``, so the calibration set is derived once and ``T`` is fit once per + run: this report re-uses both rather than rebuilding the set and fitting its + own ``T`` off a second rerank pass. For every golden row carrying an expected + decision: C.3 confidence (top-1 softmax mass) -> ``decide()`` at the auto-link + threshold. Reports the linked-vs-review accuracy (the meaningful C.4 number at + this fixed threshold) and, for expected-review rows, how often the + ``reason_code`` matches too. + + Informational — it does not fail the run: the SafetyGuard flags (adversarial / + update_ambiguous) are not wired yet, so ``decide()`` sees them as False here and + reason codes that depend on them lag until that lands; and tuning the threshold + itself is the Week 7 experiment, so hard-gating it now would be premature. + """ + from application.utils.librarian.decision_engine import decide + from application.utils.librarian.schemas import Decision + + graded = [r for r in rows if r.expected.decision is not None and r.id in audits] + if not graded: + print("decision (C.4): no rows with an expected decision in this selection") + return 0 + + dec_match = reason_match = 0 + link_total = link_correct = review_total = review_correct = 0 + for row in graded: + audit = audits[row.id] + reranked = [c for c in audit.reranked if c.score_rerank is not None] + logits = [float(c.score_rerank) for c in reranked] + cre_ids = [c.cre_id for c in reranked] + confidence = scaler.confidence(logits) if logits else 0.0 + result = decide(confidence, cre_ids, threshold=threshold) + matched = result.decision == row.expected.decision + if matched: + dec_match += 1 + if row.expected.decision == Decision.linked: + link_total += 1 + link_correct += matched + elif row.expected.decision == Decision.review: + review_total += 1 + review_correct += matched + if result.reason_code == row.expected.reason_code: + reason_match += 1 + + # Overall agreement plus the two directions split out, because a single + # accuracy hides the story at an untuned threshold: at tau=0.80 the softmax + # top-1 mass of a correct-but-close winner is often ~0.5, so many correct + # positives fall *below* the bar and route to review (the safe direction). + # Auto-link recall vs review recall makes that visible; W7 tunes tau. + n = len(graded) + print( + f"decision (C.4, {n} rows @ tau={threshold:.2f}): " + f"overall {dec_match}/{n} ({dec_match / n:.0%})" + ) + if link_total: + print( + f" auto-link recall (expected-linked rows): " + f"{link_correct}/{link_total} ({link_correct / link_total:.0%})" + ) + if review_total: + print( + f" review recall (expected-review rows): " + f"{review_correct}/{review_total} ({review_correct / review_total:.0%}); " + f"reason_code match {reason_match}/{review_total} " + f"({reason_match / review_total:.0%}) " + f"(SafetyGuard flags not wired — flag-based codes lag)" + ) + return 0 def main(argv: List[str]) -> int: @@ -331,8 +443,10 @@ def main(argv: List[str]) -> int: parser.add_argument( "--use_live_embeddings", action="store_true", - help="connect to the OpenCRE DB + embedding model and measure the live " - "C.1 retrieval recall@k and C.2 rerank top-1 over the positive slice " + help="connect to the OpenCRE DB + embedding model and run every live " + "report: C.1 retrieval recall@k and C.2 rerank top-1 over the positive " + "slice, the C.3 ECE gate (which sets a nonzero exit status when it fails " + "or cannot run), and the informational C.4 decision accuracy " "(needs an LLM + populated DB)", ) parser.add_argument( @@ -409,19 +523,33 @@ def main(argv: List[str]) -> int: args.top_k_rerank, cfg.crossencoder_model, ) + # Union of what the reports read: the calibration slices, plus any row + # carrying an expected decision — C.4 grades those and they are not + # confined to positive/hard_negative. audits = live_audits( - [r for r in rows if r.slice.value in ("positive", "hard_negative")], + [ + r + for r in rows + if r.slice.value in ("positive", "hard_negative") + or r.expected.decision is not None + ], retriever, reranker, ) report_retrieval_recall(rows, audits, args.top_k_retrieval, args.top_k_rerank) - calib_status = report_calibration(rows, audits) + calib_status, scaler = report_calibration(rows, audits) + if scaler is not None: + report_decision_accuracy(rows, audits, scaler, args.threshold) + else: + # No fitted T means no honest confidence for C.4 to threshold on. + # report_calibration has already failed the run. + print("decision (C.4): skipped — calibration produced no fitted T") else: print( - "semantic pipeline (C.1 retrieve + C.2 rerank) + calibration (C.3): " - "wired; recall@k, rerank top-1, and the ECE gate need " - "--use_live_embeddings (no CRE vectors offline — seeding from golden " - "text would be leakage)" + "semantic pipeline (C.1 retrieve + C.2 rerank) + calibration (C.3) + " + "decision (C.4): wired; recall@k, rerank top-1, the ECE gate, and the " + "decision accuracy all need --use_live_embeddings (no CRE vectors " + "offline — seeding from golden text would be leakage)" ) print(f"correct overall (semantic path still stubbed): {correct}/{len(rows)}") return calib_status