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

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

application/tests/librarian/pipeline_test.py

@@ -0,0 +1,210 @@
+"""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(result.stats.errored, 0)
+        self.assertEqual(len(result.envelopes), 2)
+
+
+class PipelineErrorContainmentTest(unittest.TestCase):
+    """A failing row must cost that row only, never the envelopes already built.
+
+    These stages are stubs today but become live DB / embedding / cross-encoder
+    calls in W8, so the containment is asserted at each seam that can raise.
+    """
+
+    def _run_with(self, failing_component, rows):
+        """Build a pipeline whose one named component raises on every call."""
+        parts = {
+            "retriever": _Retriever(),
+            "reranker": _Reranker(TOP),
+            "scaler": _Scaler(0.95),
+        }
+        parts[failing_component] = failing_component_stub(failing_component)
+        return LibrarianPipeline(
+            _Source(rows),
+            parts["retriever"],
+            parts["reranker"],
+            parts["scaler"],
+            threshold=0.8,
+            pipeline_run_id=RUN,
+        ).run(at=AT)
+
+    def test_retriever_failure_is_contained(self):
+        result = self._run_with("retriever", [_row()])
+        self.assertEqual(result.stats.errored, 1)
+        self.assertEqual(result.stats.total, 1)
+        self.assertEqual(result.envelopes, [])
+
+    def test_reranker_failure_is_contained(self):
+        result = self._run_with("reranker", [_row()])
+        self.assertEqual(result.stats.errored, 1)
+        self.assertEqual(result.envelopes, [])
+
+    def test_scaler_failure_is_contained(self):
+        result = self._run_with("scaler", [_row()])
+        self.assertEqual(result.stats.errored, 1)
+        self.assertEqual(result.envelopes, [])
+
+    def test_one_bad_row_does_not_discard_the_good_ones(self):
+        # The reranker fails only on the middle row's text.
+        class _FlakyReranker:
+            def rerank(self, text, audit):
+                if "boom" in text:
+                    raise RuntimeError("cross-encoder blew up on this pair")
+                return audit.model_copy(update={"reranked": list(TOP)})
+
+        rows = [_row(), _row(text="boom goes the model"), _row()]
+        result = LibrarianPipeline(
+            _Source(rows),
+            _Retriever(),
+            _FlakyReranker(),
+            _Scaler(0.95),
+            threshold=0.8,
+            pipeline_run_id=RUN,
+        ).run(at=AT)
+
+        self.assertEqual(result.stats.total, 3)
+        self.assertEqual(result.stats.errored, 1)
+        self.assertEqual(result.stats.linked, 2)
+        self.assertEqual(len(result.envelopes), 2)
+
+    def test_errored_is_counted_separately_from_skipped(self):
+        rows = [_row(label="UNCERTAIN"), _row()]
+        result = self._run_with("retriever", rows)
+        # The UNCERTAIN row is a clean boundary refusal; the other is a fault.
+        self.assertEqual(result.stats.skipped, 1)
+        self.assertEqual(result.stats.errored, 1)
+
+
+def failing_component_stub(kind):
+    """A stub whose single method always raises, for the given seam."""
+
+    class _Boom:
+        def retrieve(self, text):
+            raise RuntimeError("retriever down")
+
+        def rerank(self, text, audit):
+            raise RuntimeError("reranker down")
+
+        def confidence(self, logits):
+            raise RuntimeError("scaler down")
+
+    assert kind in ("retriever", "reranker", "scaler")
+    return _Boom()
+
+
+if __name__ == "__main__":
+    unittest.main()

application/utils/librarian/__init__.py

@@ -20,8 +20,10 @@
              an honest probability (fit by NLL on the golden set, gated ECE < 0.10).
   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.
+  W6b (C.4): envelope emitter — builds the RFC LinkProposal / ReviewItem from a
+             DecisionResult — plus the C.0->C.4 pipeline glue that runs a batch
+             of queue rows end to end (dry-run: nothing is persisted).
+The live queue drain and the graph / review-queue writers (W8) are not built yet.
 
 Vendored RFC JSON schemas live under ``_rfc_schemas/``. They are pinned to
 upstream/owasp-graph @ 2b1437987768d5ed20fe9ee721ab9a898c4b84af (PR #734).