week_6b: Module C (The Librarian) — C.4 envelope emitter + C.0→C.4 pipeline glue

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()

application/tests/librarian/emitter_test.py

@@ -0,0 +1,130 @@
+"""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,
+    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")
+
+    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_auto_link_type_matches_cre_defs(self):
+        from application.defs.cre_defs import LinkTypes
+
+        self.assertEqual(AUTO_LINK_TYPE, LinkTypes.AutomaticallyLinkedTo.value)
+
+
+if __name__ == "__main__":
+    unittest.main()

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()