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/emitter_test.py b/application/tests/librarian/emitter_test.py new file mode 100644 index 000000000..905c68acd --- /dev/null +++ b/application/tests/librarian/emitter_test.py @@ -0,0 +1,135 @@ +"""Hermetic tests for C.4 envelope emitter (Week 6b). No key, DB, or model.""" + +import unittest +from datetime import datetime, timezone + +from application.utils.librarian.decision_engine import DecisionResult +from application.utils.librarian.emitter import ( + AUTO_LINK_TYPE, + SUGGESTED_LINK_TYPE, + EmitterError, + build_link_proposal, + build_review_item, + emit, +) +from application.utils.librarian.schemas import ( + SCHEMA_VERSION, + CreCandidate, + Decision, + Locator, + LocatorKind, + ReasonCode, + RetrievalAudit, + SourceRef, + SourceType, +) +from application.utils.librarian.section_validator import Section + +AT = datetime(2026, 1, 1, tzinfo=timezone.utc) +RUN = "run-42" + + +def _section() -> Section: + return Section( + chunk_id="chk:owasp/x@abcdef1:a.md", + artifact_id="art:owasp/x:a.md", + text="Verify the JWT signature before trusting the session.", + title_hint=None, + language="en", + source=SourceRef( + type=SourceType.github, + repo="owasp/x", + commit_sha="abcdef1", + committed_at=AT, + ), + locator=Locator(kind=LocatorKind.repo_path, id="a.md", path="a.md"), + ) + + +def _audit() -> RetrievalAudit: + return RetrievalAudit( + retriever="retriever/0", + candidates=[CreCandidate(cre_id="616-305")], + reranked=[CreCandidate(cre_id="616-305", score_rerank=1.5)], + threshold=0.8, + ) + + +LINKED = DecisionResult(Decision.linked, 0.91, ("616-305",), None) +REVIEW_LOW = DecisionResult( + Decision.review, 0.42, ("616-305",), ReasonCode.below_threshold +) +REVIEW_EMPTY = DecisionResult(Decision.review, 0.0, (), ReasonCode.no_candidates) + + +class EmitLinkTest(unittest.TestCase): + def test_emit_linked_builds_link_proposal(self): + env = emit(_section(), _audit(), LINKED, pipeline_run_id=RUN, at=AT) + self.assertEqual(env.status, "linked") + self.assertEqual(env.schema_version, SCHEMA_VERSION) + self.assertEqual(env.chunk_id, "chk:owasp/x@abcdef1:a.md") + self.assertEqual(env.pipeline_run_id, RUN) + self.assertEqual(len(env.links), 1) + self.assertEqual(env.links[0].cre_id, "616-305") + self.assertEqual(env.links[0].link_type, AUTO_LINK_TYPE) + self.assertEqual(env.links[0].confidence, 0.91) + self.assertEqual(env.knowledge.text, _section().text) + self.assertEqual(env.retrieval.threshold, 0.8) # audit passed through + + def test_degraded_update_detection_default(self): + env = emit(_section(), _audit(), LINKED, pipeline_run_id=RUN, at=AT) + self.assertFalse( + env.update_detection.is_update + ) # declared-degraded until SafetyGuard + + def test_build_link_proposal_rejects_review(self): + with self.assertRaises(EmitterError): + build_link_proposal( + _section(), _audit(), REVIEW_LOW, pipeline_run_id=RUN, at=AT + ) + + def test_link_needs_a_cre(self): + bad = DecisionResult(Decision.linked, 0.9, (), None) + with self.assertRaises(EmitterError): + build_link_proposal(_section(), _audit(), bad, pipeline_run_id=RUN, at=AT) + + +class EmitReviewTest(unittest.TestCase): + def test_emit_review_builds_review_item(self): + env = emit(_section(), _audit(), REVIEW_LOW, pipeline_run_id=RUN, at=AT) + self.assertEqual(env.status, "review_required") + self.assertEqual(env.reason_code, ReasonCode.below_threshold) + self.assertEqual( + env.review_id, "review:chk:owasp/x@abcdef1:a.md" + ) # deterministic + self.assertIsNotNone(env.suggested_links) + self.assertEqual(env.suggested_links[0].cre_id, "616-305") + # a review suggestion is a candidate, not an auto-link — must not claim it + self.assertEqual(env.suggested_links[0].link_type, SUGGESTED_LINK_TYPE) + self.assertNotEqual(env.suggested_links[0].link_type, AUTO_LINK_TYPE) + + def test_no_candidates_review_has_no_suggestions(self): + env = emit(_section(), _audit(), REVIEW_EMPTY, pipeline_run_id=RUN, at=AT) + self.assertEqual(env.reason_code, ReasonCode.no_candidates) + self.assertIsNone(env.suggested_links) + + def test_build_review_item_rejects_linked(self): + with self.assertRaises(EmitterError): + build_review_item(_section(), _audit(), LINKED, pipeline_run_id=RUN, at=AT) + + def test_review_needs_reason_code(self): + bad = DecisionResult(Decision.review, 0.4, ("616-305",), None) + with self.assertRaises(EmitterError): + build_review_item(_section(), _audit(), bad, pipeline_run_id=RUN, at=AT) + + +class MetadataTest(unittest.TestCase): + def test_link_types_match_cre_defs(self): + from application.defs.cre_defs import LinkTypes + + self.assertEqual(AUTO_LINK_TYPE, LinkTypes.AutomaticallyLinkedTo.value) + self.assertEqual(SUGGESTED_LINK_TYPE, LinkTypes.Related.value) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/pipeline_test.py b/application/tests/librarian/pipeline_test.py new file mode 100644 index 000000000..9d765d9ab --- /dev/null +++ b/application/tests/librarian/pipeline_test.py @@ -0,0 +1,121 @@ +"""Hermetic tests for the C.0->C.4 pipeline (Week 6b). + +Every stage is a trivial stub — no DB, embedding model, or cross-encoder. +""" + +import unittest +from datetime import datetime, timezone + +from application.utils.librarian.pipeline import LibrarianPipeline +from application.utils.librarian.schemas import ( + CreCandidate, + KnowledgeQueueItem, + LinkProposal, + ReasonCode, + RetrievalAudit, + ReviewItem, +) + +AT = datetime(2026, 1, 1, tzinfo=timezone.utc) +RUN = "run-7" + + +def _row(text="Verify the JWT signature.", label="KNOWLEDGE"): + return KnowledgeQueueItem( + id="1", + source_repo="owasp/x", + source_path="a.md", + source_commit_sha="abcdef1", + text=text, + confidence=0.9, + llm_label=label, + created_at="2026-01-01T00:00:00Z", + ) + + +class _Source: + def __init__(self, rows): + self._rows = rows + + def items(self): + return iter(self._rows) + + +class _Retriever: + def retrieve(self, text): + return RetrievalAudit( + retriever="stub", + candidates=[CreCandidate(cre_id="616-305")], + reranked=[], + threshold=0.8, + ) + + +class _Reranker: + def __init__(self, reranked): + self._reranked = reranked + + def rerank(self, text, audit): + return audit.model_copy(update={"reranked": list(self._reranked)}) + + +class _Scaler: + def __init__(self, conf): + self._conf = conf + + def confidence(self, logits): + return self._conf + + +TOP = [CreCandidate(cre_id="616-305", score_rerank=1.5)] + + +def _pipeline(rows, reranked, conf): + return LibrarianPipeline( + _Source(rows), + _Retriever(), + _Reranker(reranked), + _Scaler(conf), + threshold=0.8, + pipeline_run_id=RUN, + ) + + +class PipelineTest(unittest.TestCase): + def test_confident_row_auto_links(self): + result = _pipeline([_row()], TOP, 0.95).run(at=AT) + self.assertEqual(result.stats.linked, 1) + self.assertEqual(result.stats.review, 0) + self.assertIsInstance(result.envelopes[0], LinkProposal) + self.assertEqual(result.envelopes[0].pipeline_run_id, RUN) + + def test_low_confidence_row_reviews_below_threshold(self): + result = _pipeline([_row()], TOP, 0.4).run(at=AT) + self.assertEqual(result.stats.review, 1) + env = result.envelopes[0] + self.assertIsInstance(env, ReviewItem) + self.assertEqual(env.reason_code, ReasonCode.below_threshold) + + def test_empty_shortlist_reviews_no_candidates(self): + result = _pipeline([_row()], [], 0.95).run(at=AT) + env = result.envelopes[0] + self.assertIsInstance(env, ReviewItem) + self.assertEqual(env.reason_code, ReasonCode.no_candidates) + + def test_uncertain_row_is_skipped_at_boundary(self): + result = _pipeline([_row(label="UNCERTAIN")], TOP, 0.95).run(at=AT) + self.assertEqual(result.stats.skipped, 1) + self.assertEqual(result.stats.total, 1) + self.assertEqual(result.envelopes, []) + + def test_mixed_batch_counts(self): + rows = [_row(), _row(label="UNCERTAIN"), _row()] + result = _pipeline(rows, TOP, 0.95).run(at=AT) + self.assertEqual(result.stats.total, 3) + self.assertEqual(result.stats.linked, 2) + self.assertEqual(result.stats.skipped, 1) + self.assertEqual(len(result.envelopes), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/temperature_test.py b/application/tests/librarian/temperature_test.py new file mode 100644 index 000000000..4b4dee1cd --- /dev/null +++ b/application/tests/librarian/temperature_test.py @@ -0,0 +1,169 @@ +"""Tests for C.3 temperature-scaling calibration (Week 5). + +Hermetic and deterministic: synthetic candidate-logit shortlists + 0/1 labels +only — no cross-encoder, DB, or embedding key. Covers the softmax-over-shortlist +confidence, temperature flatten/sharpen behaviour, the NLL fit recovering a known +T and reducing NLL, ECE on perfectly- and mis-calibrated data (hand-checked), and +every guard. +""" + +import math +import unittest + +import numpy as np +from scipy.special import softmax + +from application.utils.librarian.calibration.temperature import ( + CALIBRATOR_NAME, + CalibrationError, + DegenerateLabelsError, + TemperatureScaler, + expected_calibration_error, + fit_temperature, + negative_log_likelihood, +) + + +class TemperatureScalerTest(unittest.TestCase): + def test_confidence_is_softmax_top_mass(self) -> None: + logits = [2.0, 1.0, 0.0] + expected = float(softmax(np.array(logits)).max()) + self.assertAlmostEqual(TemperatureScaler(1.0).confidence(logits), expected, 9) + + def test_probabilities_sum_to_one_and_peak_at_top(self) -> None: + p = TemperatureScaler(1.0).probabilities([3.0, 1.0, -2.0]) + self.assertAlmostEqual(float(p.sum()), 1.0, places=9) + self.assertEqual(int(np.argmax(p)), 0) # highest logit -> highest prob + + def test_high_temperature_flattens_toward_uniform(self) -> None: + logits = [3.0, 1.0, 0.0] + hot = TemperatureScaler(1000.0).confidence(logits) + base = TemperatureScaler(1.0).confidence(logits) + self.assertLess(hot, base) + self.assertAlmostEqual(hot, 1.0 / 3.0, places=2) # ~uniform over 3 + + def test_low_temperature_sharpens_toward_one(self) -> None: + logits = [3.0, 1.0, 0.0] + cold = TemperatureScaler(0.1).confidence(logits) + base = TemperatureScaler(1.0).confidence(logits) + self.assertGreater(cold, base) + self.assertGreater(cold, 0.99) + + def test_single_candidate_confidence_is_one(self) -> None: + # softmax over one element is always 1.0, at any temperature. + self.assertAlmostEqual(TemperatureScaler(3.0).confidence([0.42]), 1.0, 9) + + def test_empty_shortlist_rejected(self) -> None: + with self.assertRaises(CalibrationError): + TemperatureScaler(1.0).confidence([]) + + def test_non_positive_or_nonfinite_temperature_rejected(self) -> None: + for bad in (0.0, -1.0, float("inf"), float("nan")): + with self.assertRaises(CalibrationError): + TemperatureScaler(bad) + + +class NegativeLogLikelihoodTest(unittest.TestCase): + def test_matches_hand_computed_single_pair(self) -> None: + # one shortlist, top-1 correct: loss = -log(softmax([2,0]).max()) + conf = float(softmax(np.array([2.0, 0.0])).max()) + self.assertAlmostEqual( + negative_log_likelihood([[2.0, 0.0]], [1.0], 1.0), -math.log(conf), 6 + ) + + def test_length_mismatch_rejected(self) -> None: + with self.assertRaises(CalibrationError): + negative_log_likelihood([[1.0, 0.0], [2.0, 0.0]], [1.0], 1.0) + + def test_non_positive_temperature_rejected(self) -> None: + with self.assertRaises(CalibrationError): + negative_log_likelihood([[1.0, 0.0]], [1.0], 0.0) + + +class FitTemperatureTest(unittest.TestCase): + def _synthetic(self, true_t: float, n: int = 4000, k: int = 5): + """Shortlists whose top-1 correctness is drawn from softmax(logits/true_t). + + Data generated with true_t flattens the softmax, so the raw (T=1) + confidence is over-confident and the fit should recover T ~ true_t. + Seeded -> stable. + """ + rng = np.random.default_rng(0) + sets, labels = [], [] + for _ in range(n): + z = rng.normal(0.0, 3.0, size=k) + z[::-1].sort() # descending so index 0 is the top-1 (argmax) + p_correct = softmax(z / true_t).max() + sets.append(z.tolist()) + labels.append(1.0 if rng.random() < p_correct else 0.0) + return sets, labels + + def test_recovers_known_temperature(self) -> None: + sets, labels = self._synthetic(true_t=2.0) + scaler = fit_temperature(sets, labels) + self.assertAlmostEqual(scaler.temperature, 2.0, delta=0.5) + + def test_fit_reduces_nll_versus_T1(self) -> None: + sets, labels = self._synthetic(true_t=2.0) + scaler = fit_temperature(sets, labels) + self.assertLess( + negative_log_likelihood(sets, labels, scaler.temperature), + negative_log_likelihood(sets, labels, 1.0), + ) + + def test_single_class_labels_raise_degenerate(self) -> None: + with self.assertRaises(DegenerateLabelsError): + fit_temperature([[2.0, 0.0], [3.0, 1.0]], [1.0, 1.0]) + + def test_non_binary_labels_rejected(self) -> None: + with self.assertRaises(CalibrationError): + fit_temperature([[2.0, 0.0], [3.0, 1.0]], [0.0, 2.0]) + + +class ExpectedCalibrationErrorTest(unittest.TestCase): + def test_perfectly_calibrated_is_near_zero(self) -> None: + # bin [0.5,0.6): five correct, five wrong -> acc 0.5 == conf 0.5. + self.assertAlmostEqual( + expected_calibration_error([0.5] * 10, [1.0, 0.0] * 5), 0.0, places=6 + ) + + def test_confidently_wrong_is_large(self) -> None: + self.assertAlmostEqual( + expected_calibration_error([0.9] * 10, [0.0] * 10), 0.9, places=6 + ) + + def test_hand_checked_two_bin_value(self) -> None: + # [0.2,0.3): conf .2 acc 0 gap .2 ; [0.8,0.9): conf .8 acc 1 gap .2 + # ECE = 0.5*0.2 + 0.5*0.2 = 0.20 + self.assertAlmostEqual( + expected_calibration_error([0.2, 0.2, 0.8, 0.8], [0.0, 0.0, 1.0, 1.0]), + 0.20, + places=6, + ) + + def test_length_mismatch_and_empty_rejected(self) -> None: + with self.assertRaises(CalibrationError): + expected_calibration_error([0.5, 0.6], [1.0]) + with self.assertRaises(CalibrationError): + expected_calibration_error([], []) + with self.assertRaises(CalibrationError): + expected_calibration_error([0.5], [1.0], n_bins=0) + + +class EndToEndTest(unittest.TestCase): + def test_flat_shortlists_are_low_confidence(self) -> None: + # A near-tie shortlist (bad match) -> low top-1 confidence. + self.assertLess(TemperatureScaler(1.0).confidence([0.1, 0.0, -0.1]), 0.45) + + def test_peaked_shortlist_is_high_confidence(self) -> None: + # A clear winner -> high top-1 confidence. + self.assertGreater(TemperatureScaler(1.0).confidence([6.0, -1.0, -3.0]), 0.9) + + +class MetadataTest(unittest.TestCase): + def test_calibrator_name_is_versioned(self) -> None: + self.assertEqual(CALIBRATOR_NAME, "temperature-scaling/0.2.0") + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/librarian/__init__.py b/application/utils/librarian/__init__.py index f168f2bb4..1a23cc3cb 100644 --- a/application/utils/librarian/__init__.py +++ b/application/utils/librarian/__init__.py @@ -16,7 +16,9 @@ explicit-link resolution). W3 (C.1): candidate retriever (in-memory + pgvector) + pipeline switch. W4 (C.2): cross-encoder reranker — re-sorts the C.1 shortlist, fills reranked[]. -Calibration + decision routing (C.3-C.4, W5-W6) onward is not built yet. + 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. 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 new file mode 100644 index 000000000..54b774038 --- /dev/null +++ b/application/utils/librarian/calibration/__init__.py @@ -0,0 +1,15 @@ +"""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 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 +(numpy + scipy) and model-free so it stays hermetically testable — mirrors the +C.1/C.2 seams. +""" diff --git a/application/utils/librarian/calibration/temperature.py b/application/utils/librarian/calibration/temperature.py new file mode 100644 index 000000000..928dacf7a --- /dev/null +++ b/application/utils/librarian/calibration/temperature.py @@ -0,0 +1,207 @@ +"""Module C.3 — temperature-scaling calibration (Week 5). The truth-teller. + +C.2 hands up a reranked shortlist of candidate CREs, each with a raw cross-encoder +logit (unbounded, higher = better match). We need one honest number: "how likely +is the top candidate the correct CRE?" — so Week 6 can threshold auto-link vs. +human review. + +The naive attempt — ``sigmoid(top1_logit / T)`` on the single absolute logit — +does not work, and the golden set proves why: a cross-encoder's absolute logit has +no fixed zero point (its 50/50 boundary is not at z=0), and dividing by a single +temperature can only *squash* toward 0.5, never *shift* the boundary. So no T +makes it honest. + +The fix is **temperature scaling in the Guo et al. sense**: calibrate the *softmax +over the whole shortlist*, not one absolute logit. The candidates' *relative* +logits are what a cross-encoder's scores actually mean, so + + p = softmax(logits / T) confidence = p for the top-1 candidate + +is a genuine probability distribution over "which candidate is right", and its +top-1 mass answers exactly the question Week 6 asks. It is still **one knob T**: +T = 1 leaves the distribution unchanged, T > 1 flattens it (less confident), T < 1 +sharpens it. T is fit once by minimising negative-log-likelihood of "is the top-1 +correct?" on the golden set; honesty is measured by Expected Calibration Error, +and the Week 5 gate is ECE < 0.10. + +Like C.1/C.2 this is a thin, model-free seam (numpy + scipy only) so every branch +is hermetically testable. The fitted ``TemperatureScaler`` is a frozen, shareable +artifact the W6 decision engine loads to turn a reranked shortlist into the +confidence it thresholds on. +""" + +from dataclasses import dataclass +from typing import Sequence + +import numpy as np +from scipy.optimize import minimize_scalar +from scipy.special import softmax + +# Identify the calibrator in the RFC audit trail (mirrors RETRIEVER_NAME / +# RERANKER_NAME). 0.2.0 = softmax-over-shortlist (the single-logit sigmoid of +# 0.1.0 could not be calibrated by temperature alone — see module docstring). +CALIBRATOR_NAME = "temperature-scaling/0.2.0" + +# Clip probabilities off {0, 1} so the log in NLL stays finite. +_EPS = 1e-7 + + +class CalibrationError(ValueError): + """Base class for calibration construction/usage failures.""" + + +class DegenerateLabelsError(CalibrationError): + """Labels are single-class — temperature is unidentifiable from NLL. + + With only correct (or only incorrect) top-1s, NLL is monotonic in ``T`` and + the optimum runs to a bound: the fit is meaningless. The calibration set must + contain both outcomes (in the harness: the ``positive`` slice supplies + correct top-1s, ``hard_negative`` supplies incorrect ones). + """ + + +def _softmax_top(logits: Sequence[float], temperature: float) -> float: + """Top-1 probability mass of ``softmax(logits / T)`` over one shortlist.""" + z = np.asarray(list(logits), dtype=float) + if z.size == 0: + raise CalibrationError("cannot calibrate an empty candidate shortlist") + return float(softmax(z / temperature).max()) + + +def _validate_temperature(temperature: float) -> None: + if not np.isfinite(temperature) or temperature <= 0: + raise CalibrationError(f"temperature must be finite and > 0, got {temperature}") + + +def _paired(logit_sets: Sequence[Sequence[float]], labels: Sequence[float]): + """Validate matched (shortlist, label) inputs: non-empty and equal length.""" + sets = list(logit_sets) + y = np.asarray(list(labels), dtype=float) + if y.ndim != 1: + raise CalibrationError(f"labels must be 1-D, got shape {y.shape}") + if len(sets) != y.shape[0]: + raise CalibrationError( + f"{len(sets)} logit-sets and {y.shape[0]} labels must be the same length" + ) + if not sets: + raise CalibrationError("need at least one (shortlist, label) pair") + return sets, y + + +@dataclass(frozen=True) +class TemperatureScaler: + """Maps a reranked shortlist to a calibrated top-1 confidence. + + ``temperature`` is the single learned scalar; build one with + ``fit_temperature``. Frozen so a fitted scaler is a stable, shareable value + (like the retriever/reranker being constructed once and reused). + """ + + temperature: float + + def __post_init__(self) -> None: + _validate_temperature(self.temperature) + + def probabilities(self, logits: Sequence[float]) -> np.ndarray: + """The full ``softmax(logits / T)`` distribution over one shortlist.""" + z = np.asarray(list(logits), dtype=float) + if z.size == 0: + raise CalibrationError("cannot calibrate an empty candidate shortlist") + return softmax(z / self.temperature) + + def confidence(self, logits: Sequence[float]) -> float: + """P(the top candidate is correct) — the top-1 mass of the softmax. + + This is the number the W6 decision engine thresholds on. + """ + return _softmax_top(logits, self.temperature) + + +def negative_log_likelihood( + logit_sets: Sequence[Sequence[float]], + labels: Sequence[float], + temperature: float, +) -> float: + """Binary cross-entropy of the top-1 confidence against 0/1 correctness. + + For each shortlist the confidence is ``softmax(logits / T)``'s top-1 mass; + the label is 1 iff that top-1 candidate is the correct CRE. This is the + objective ``fit_temperature`` minimises over ``T``; exposed so the fit is + testable and a caller can compare NLL at ``T=1`` vs the fitted T. + """ + _validate_temperature(temperature) + sets, y = _paired(logit_sets, labels) + p = np.clip( + np.array([_softmax_top(s, temperature) for s in sets]), _EPS, 1.0 - _EPS + ) + return float(-np.sum(y * np.log(p) + (1.0 - y) * np.log(1.0 - p))) + + +def fit_temperature( + logit_sets: Sequence[Sequence[float]], + labels: Sequence[float], + *, + bounds: tuple = (1e-2, 1e2), +) -> TemperatureScaler: + """Fit ``T`` by minimising NLL over (shortlist, is-top1-correct) pairs. + + ``labels`` must be 0/1 and contain both outcomes (else + ``DegenerateLabelsError``). One free parameter over a bounded interval, so + the 1-D ``minimize_scalar`` fit is fast and barely over-fits — the golden set + is the calibration set by design. + """ + sets, y = _paired(logit_sets, labels) + values = set(np.unique(y).tolist()) + if not values <= {0.0, 1.0}: + raise CalibrationError(f"labels must be 0/1, got values {sorted(values)}") + if len(values) < 2: + raise DegenerateLabelsError( + "labels are single-class; temperature is unidentifiable — the " + "calibration set needs both correct and incorrect top-1s" + ) + + result = minimize_scalar( + lambda t: negative_log_likelihood(sets, y, t), + bounds=bounds, + method="bounded", + ) + if not result.success: + raise CalibrationError(f"temperature fit did not converge: {result.message}") + return TemperatureScaler(temperature=float(result.x)) + + +def expected_calibration_error( + confidences: Sequence[float], labels: Sequence[float], *, n_bins: int = 10 +) -> float: + """ECE — sample-weighted mean ``|accuracy - confidence|`` across equal bins. + + Partition [0, 1] into ``n_bins`` equal-width bins. Per bin, ``confidence`` is + the mean predicted top-1 probability and ``accuracy`` is the fraction whose + top-1 was actually correct; ECE weights each bin's gap by its share of the + sample. 0 = perfectly honest; the Week 5 gate is < 0.10. + """ + if n_bins < 1: + raise CalibrationError(f"n_bins must be >= 1, got {n_bins}") + p = np.asarray(list(confidences), dtype=float) + y = np.asarray(list(labels), dtype=float) + if p.shape != y.shape: + raise CalibrationError( + f"confidences {p.shape} and labels {y.shape} must be the same length" + ) + if p.size == 0: + raise CalibrationError("need at least one (confidence, label) pair") + + edges = np.linspace(0.0, 1.0, n_bins + 1) + idx = np.clip(np.digitize(p, edges[1:-1], right=False), 0, n_bins - 1) + + n = p.size + ece = 0.0 + for b in range(n_bins): + mask = idx == b + count = int(mask.sum()) + if count == 0: + continue + confidence = float(p[mask].mean()) + accuracy = float(y[mask].mean()) + ece += (count / n) * abs(accuracy - confidence) + return ece 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/application/utils/librarian/emitter.py b/application/utils/librarian/emitter.py new file mode 100644 index 000000000..8591e5199 --- /dev/null +++ b/application/utils/librarian/emitter.py @@ -0,0 +1,158 @@ +"""Module C.4 — envelope emitter (Week 6b). The scribe. + +The decision engine (C.4) produces a verdict; this module turns that verdict, plus +the chunk's ``Section`` and the C.1/C.2 ``RetrievalAudit``, into the wire contract +Module D consumes: an RFC ``LinkProposal`` (auto-link) or ``ReviewItem`` (human +review). It only *builds* the envelope — persisting/queuing it is W8's writers. + +Pure and timestamp-injected (``at`` is passed, not read from the clock) so every +branch is hermetically testable. ``update_detection`` defaults to the declared +degraded value (``is_update=False``) until the SafetyGuard lands; ``review_id`` is +derived deterministically from the chunk id so the same chunk always maps to the +same review, with no nondeterministic uuid in a pure builder. +""" + +from datetime import datetime +from typing import Optional, Union + +from application.utils.librarian.decision_engine import DecisionResult +from application.utils.librarian.schemas import ( + SCHEMA_VERSION, + Decision, + KnowledgeSnapshot, + LinkProposal, + ProposedLink, + RetrievalAudit, + ReviewItem, + UpdateDetection, +) +from application.utils.librarian.section_validator import Section + +# Link types C stamps on a ProposedLink, kept as literals so the emitter stays +# import-light and hermetic (both mirror cre_defs.LinkTypes values). +# AUTO_LINK_TYPE -- an auto-linked chunk (LinkProposal.links): C committed to it. +# SUGGESTED_LINK_TYPE -- a review suggestion (ReviewItem.suggested_links): only a +# candidate for a human to consider, NOT an auto-link, so it +# must not claim "Automatically linked to". +AUTO_LINK_TYPE = "Automatically linked to" +SUGGESTED_LINK_TYPE = "Related" + + +class EmitterError(ValueError): + """Raised when a DecisionResult cannot be turned into an envelope.""" + + +def _degraded_update_detection() -> UpdateDetection: + """The declared-degraded default until the SafetyGuard wires update detection.""" + return UpdateDetection(is_update=False) + + +def _snapshot(section: Section) -> KnowledgeSnapshot: + return KnowledgeSnapshot( + text=section.text, source=section.source, locator=section.locator + ) + + +def _proposed_links(result: DecisionResult, link_type: str) -> list: + """One ProposedLink per chosen CRE (top-1), carrying the calibrated confidence. + + ``link_type`` distinguishes an auto-link (``AUTO_LINK_TYPE``) from a review + suggestion (``SUGGESTED_LINK_TYPE``) so a not-yet-confirmed suggestion is never + labelled as an automatic link. + """ + return [ + ProposedLink( + cre_id=cre_id, + link_type=link_type, + confidence=result.confidence, + rationale=None, + ) + for cre_id in result.cre_ids + ] + + +def build_link_proposal( + section: Section, + audit: RetrievalAudit, + result: DecisionResult, + *, + pipeline_run_id: str, + at: datetime, + update_detection: Optional[UpdateDetection] = None, +) -> LinkProposal: + """Build the auto-link envelope. ``result.decision`` must be ``linked``.""" + if result.decision != Decision.linked: + raise EmitterError( + f"build_link_proposal needs a linked decision, got {result.decision}" + ) + links = _proposed_links(result, AUTO_LINK_TYPE) + if not links: + raise EmitterError("a linked decision must carry at least one CRE id") + return LinkProposal( + schema_version=SCHEMA_VERSION, + chunk_id=section.chunk_id, + artifact_id=section.artifact_id, + pipeline_run_id=pipeline_run_id, + classified_at=at, + knowledge=_snapshot(section), + retrieval=audit, + links=links, + update_detection=update_detection or _degraded_update_detection(), + ) + + +def build_review_item( + section: Section, + audit: RetrievalAudit, + result: DecisionResult, + *, + pipeline_run_id: str, + at: datetime, + update_detection: Optional[UpdateDetection] = None, +) -> ReviewItem: + """Build the human-review envelope. ``result.decision`` must be ``review``.""" + if result.decision != Decision.review: + raise EmitterError( + f"build_review_item needs a review decision, got {result.decision}" + ) + if result.reason_code is None: + raise EmitterError("a review decision must carry a reason_code") + suggested = ( + _proposed_links(result, SUGGESTED_LINK_TYPE) or None + ) # best guess, may be empty + return ReviewItem( + schema_version=SCHEMA_VERSION, + review_id=f"review:{section.chunk_id}", + chunk_id=section.chunk_id, + artifact_id=section.artifact_id, + pipeline_run_id=pipeline_run_id, + created_at=at, + reason_code=result.reason_code, + knowledge=_snapshot(section), + retrieval=audit, + suggested_links=suggested, + update_detection=update_detection or _degraded_update_detection(), + ) + + +def emit( + section: Section, + audit: RetrievalAudit, + result: DecisionResult, + *, + pipeline_run_id: str, + at: datetime, + update_detection: Optional[UpdateDetection] = None, +) -> Union[LinkProposal, ReviewItem]: + """Dispatch on the verdict: ``linked`` -> LinkProposal, ``review`` -> ReviewItem.""" + builder = ( + build_link_proposal if result.decision == Decision.linked else build_review_item + ) + return builder( + section, + audit, + result, + pipeline_run_id=pipeline_run_id, + at=at, + update_detection=update_detection, + ) diff --git a/application/utils/librarian/pipeline.py b/application/utils/librarian/pipeline.py new file mode 100644 index 000000000..3597e63c8 --- /dev/null +++ b/application/utils/librarian/pipeline.py @@ -0,0 +1,104 @@ +"""Module C.4 — the pipeline (Week 6b). The assembly line. + +Wires the librarian end to end for a stream of ``knowledge_queue`` rows: + + C.0 section_from_queue_row row -> validated Section (malformed rows skipped) + C.1 retriever.retrieve text -> RetrievalAudit.candidates (top-K) + C.2 reranker.rerank text -> RetrievalAudit.reranked (top-N logits) + C.3 scaler.confidence logits -> one calibrated confidence + C.4 decide + emit confidence -> LinkProposal | ReviewItem + +Every stage is an injected seam (``source``/``retriever``/``reranker``/``scaler``), +so the whole pipeline runs hermetically with stubs — no DB, embedding model, or +cross-encoder. It is inherently **dry-run**: it builds envelopes and never persists +(the queue write-back and graph writes are W8). ``pipeline_run_id`` and the ``at`` +timestamp are injected, never read from the clock, so a run is reproducible. +""" + +from dataclasses import dataclass +from datetime import datetime +from typing import List, Union + +from application.utils.librarian.decision_engine import decide +from application.utils.librarian.emitter import emit +from application.utils.librarian.schemas import LinkProposal, ReviewItem +from application.utils.librarian.section_validator import ( + SectionValidationError, + section_from_queue_row, +) + +Envelope = Union[LinkProposal, ReviewItem] + + +@dataclass(frozen=True) +class RunStats: + """Counts for one pipeline run. ``skipped`` are rows rejected at the C.0 boundary.""" + + total: int + linked: int + review: int + skipped: int + + +@dataclass(frozen=True) +class RunResult: + envelopes: List[Envelope] + stats: RunStats + + +class LibrarianPipeline: + """Runs C.0 -> C.4 over a knowledge source, emitting one envelope per valid row. + + ``scaler`` is a fitted C.3 ``TemperatureScaler`` (the persisted ``T``); + ``threshold`` is the C.4 auto-link bar. Components are duck-typed to their one + method each (``source.items`` / ``retriever.retrieve`` / ``reranker.rerank`` / + ``scaler.confidence``) so tests inject trivial stubs. + """ + + def __init__( + self, + source, + retriever, + reranker, + scaler, + *, + threshold: float, + pipeline_run_id: str + ) -> None: + self._source = source + self._retriever = retriever + self._reranker = reranker + self._scaler = scaler + self._threshold = threshold + self._run_id = pipeline_run_id + + def run(self, *, at: datetime) -> RunResult: + envelopes: List[Envelope] = [] + linked = review = skipped = total = 0 + for item in self._source.items(): + total += 1 + try: + section = section_from_queue_row(item) + except SectionValidationError: + skipped += 1 # rejected at the boundary; not a decision + continue + + audit = self._retriever.retrieve(section.text) + audit = self._reranker.rerank(section.text, audit) + 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 = self._scaler.confidence(logits) if logits else 0.0 + + result = decide(confidence, cre_ids, threshold=self._threshold) + envelope = emit(section, audit, result, pipeline_run_id=self._run_id, at=at) + envelopes.append(envelope) + if isinstance(envelope, LinkProposal): + linked += 1 + else: + review += 1 + + return RunResult( + envelopes=envelopes, + stats=RunStats(total=total, linked=linked, review=review, skipped=skipped), + ) diff --git a/scripts/evaluate_librarian.py b/scripts/evaluate_librarian.py index 89afa773d..9e3e95937 100644 --- a/scripts/evaluate_librarian.py +++ b/scripts/evaluate_librarian.py @@ -94,28 +94,20 @@ def predict(section: Section, registry: Set[str], hub: List[HubRep]) -> List[str return [] -def report_retrieval_recall( - rows: List[GoldenDatasetRow], +def _build_live_pipeline( cache_file: str, top_k: int, threshold: float, top_n_rerank: int, crossencoder_model: str, -) -> None: - """Measure the live C.1 -> C.2 pipeline over the positive slice (v1). +): + """Construct the live C.1 retriever + C.2 reranker against the OpenCRE DB. - Two metrics, both live — there is no honest offline value: the candidate - pool must be the real CRE-node vectors, and seeding it from the golden text - is exactly the leakage the hub firewall strips. - - - retrieval recall@k (C.1): does the expected CRE id make it into the top-K - shortlist the reranker will see? A miss here is unrecoverable downstream. - - rerank top-1 (C.2): after the cross-encoder re-reads each pair and re-sorts - the shortlist, is the #1 candidate an expected CRE? This is the first - end-to-end accuracy number for the search path (W4 target >= 0.80). + Live deps are imported lazily so the offline harness needs neither a DB, an + embedding model, nor the cross-encoder stack. Shared by every live report + (recall/top-1 and calibration) so the heavy hub + model load happens once + per report and the id-space translation stays in one place. """ - # Live deps are imported lazily so the offline harness needs neither a DB, - # an embedding model, nor the cross-encoder stack. from application.cmd.cre_main import db_connect from application.defs import cre_defs from application.prompt_client import prompt_client @@ -164,7 +156,30 @@ def _to_ext(mapping): database.get_embedding_contents_by_doc_type(cre_defs.Credoctypes.CRE.value) ), ) + return retriever, reranker + +def report_retrieval_recall( + rows: List[GoldenDatasetRow], + retriever, + reranker, + top_k: int, + top_n_rerank: int, +) -> None: + """Measure the live C.1 -> C.2 pipeline over the positive slice (v1). + + Takes a prebuilt ``retriever``/``reranker`` (built once in ``main``) so the + DB, embedding model, and cross-encoder load once per run and are shared with + ``report_calibration``. Two metrics, both live — there is no honest offline + value: the candidate pool must be the real CRE-node vectors, and seeding it + from the golden text is exactly the leakage the hub firewall strips. + + - retrieval recall@k (C.1): does the expected CRE id make it into the top-K + shortlist the reranker will see? A miss here is unrecoverable downstream. + - rerank top-1 (C.2): after the cross-encoder re-reads each pair and re-sorts + the shortlist, is the #1 candidate an expected CRE? This is the first + end-to-end accuracy number for the search path (W4 target >= 0.80). + """ positives = [r for r in rows if r.slice.value == "positive" and r.expected.cre_ids] if not positives: print("retrieval recall: no positive rows with expected ids in this selection") @@ -193,6 +208,156 @@ def _to_ext(mapping): ) +def report_calibration( + rows: List[GoldenDatasetRow], + retriever, + reranker, +) -> int: + """Fit temperature on the golden set and report the Week 5 ECE gate (< 0.10). + + Takes the prebuilt ``retriever``/``reranker`` shared with + ``report_retrieval_recall`` (built once in ``main``). 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 so a live run can fail. + """ + from application.utils.librarian.calibration.temperature import ( + TemperatureScaler, + expected_calibration_error, + fit_temperature, + ) + + cal_rows = [r for r in rows if r.slice.value in ("positive", "hard_negative")] + logit_sets: List[List[float]] = [] + labels: List[float] = [] + for row in cal_rows: + audit = reranker.rerank(row.input.text, retriever.retrieve(row.input.text)) + reranked = [c for c in audit.reranked if c.score_rerank is not None] + if not reranked: + continue + 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) + + if len(set(labels)) < 2: + print( + "calibration (C.3): need both outcomes in the selection (positive + " + "hard_negative slices) to fit temperature; skipped" + ) + return 0 + + scaler = fit_temperature(logit_sets, labels) + conf_raw = [TemperatureScaler(1.0).confidence(s) for s in logit_sets] + conf_cal = [scaler.confidence(s) for s in logit_sets] + ece_raw = expected_calibration_error(conf_raw, labels) + ece_cal = expected_calibration_error(conf_cal, labels) + gate_ok = ece_cal < 0.10 + print( + f"calibration (C.3, {len(labels)} rows): fitted T={scaler.temperature:.3f}; " + 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 + + +def report_decision_accuracy( + rows: List[GoldenDatasetRow], + retriever, + reranker, + 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. + + Fits temperature on the positive + hard_negative slices (as in + ``report_calibration``), then for every golden row carrying an expected + decision: retrieve -> rerank -> 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.calibration.temperature import fit_temperature + from application.utils.librarian.decision_engine import decide + from application.utils.librarian.schemas import Decision + + # Fit T on the same positive + hard_negative calibration set as C.3. + cal_rows = [r for r in rows if r.slice.value in ("positive", "hard_negative")] + logit_sets: List[List[float]] = [] + labels: List[float] = [] + for row in cal_rows: + audit = reranker.rerank(row.input.text, retriever.retrieve(row.input.text)) + reranked = [c for c in audit.reranked if c.score_rerank is not None] + if not reranked: + continue + 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) + if len(set(labels)) < 2: + print("decision (C.4): need both outcomes to fit temperature; skipped") + return 0 + scaler = fit_temperature(logit_sets, labels) + + graded = [r for r in rows if r.expected.decision is not None] + 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 = reranker.rerank(row.input.text, retriever.retrieve(row.input.text)) + 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: cfg = load_config() parser = argparse.ArgumentParser(description="Module C eval harness (W2: C.0)") @@ -275,23 +440,32 @@ def main(argv: List[str]) -> int: ) if not gate_ok: return 1 + calib_status = 0 if args.use_live_embeddings: - report_retrieval_recall( - rows, + # Build the live pipeline once (DB + embedding model + cross-encoder) and + # share it across both live reports, so the heavy load and per-row rerank + # happen a single time per run. + retriever, reranker = _build_live_pipeline( args.cache_file, args.top_k_retrieval, args.threshold, args.top_k_rerank, cfg.crossencoder_model, ) + report_retrieval_recall( + rows, retriever, reranker, args.top_k_retrieval, args.top_k_rerank + ) + calib_status = report_calibration(rows, retriever, reranker) + report_decision_accuracy(rows, retriever, reranker, args.threshold) else: print( - "semantic pipeline (C.1 retrieve + C.2 rerank): wired; recall@k and " - "rerank top-1 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): " + "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)" ) print(f"correct overall (semantic path still stubbed): {correct}/{len(rows)}") - return 0 + return calib_status if __name__ == "__main__":