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..9bddd7ee1 --- /dev/null +++ b/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() diff --git a/application/utils/librarian/__init__.py b/application/utils/librarian/__init__.py index 869b807fa..a787f9ac7 100644 --- a/application/utils/librarian/__init__.py +++ b/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). 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..8ecbec1ec --- /dev/null +++ b/application/utils/librarian/pipeline.py @@ -0,0 +1,178 @@ +"""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. +""" + +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Dict, Iterable, List, Protocol, Sequence, Union + +from application.utils.librarian.decision_engine import decide +from application.utils.librarian.emitter import emit +from application.utils.librarian.schemas import ( + KnowledgeQueueItem, + LinkProposal, + RetrievalAudit, + ReviewItem, +) +from application.utils.librarian.section_validator import ( + SectionValidationError, + section_from_queue_row, +) + +logger = logging.getLogger(__name__) + +Envelope = Union[LinkProposal, ReviewItem] + + +# The injected seams, as Protocols rather than bare duck-typing: each stage is +# structurally one method, so a stub only has to provide that method, while +# ``make mypy --strict`` can still check the call sites and every implementation +# the live wiring passes in (W8). + + +class KnowledgeSource(Protocol): + """C.0 input: yields ``knowledge_queue`` rows, validated or still raw dicts.""" + + def items(self) -> Iterable[Union[KnowledgeQueueItem, Dict[str, Any]]]: ... + + +class Retriever(Protocol): + """C.1: text -> shortlist of candidate CREs.""" + + def retrieve(self, text: str) -> RetrievalAudit: ... + + +class Reranker(Protocol): + """C.2: re-sorts a shortlist, filling ``reranked`` with cross-encoder logits.""" + + def rerank(self, text: str, audit: RetrievalAudit) -> RetrievalAudit: ... + + +class Scaler(Protocol): + """C.3: reranked logits -> one calibrated top-1 confidence.""" + + def confidence(self, logits: Sequence[float]) -> float: ... + + +@dataclass(frozen=True) +class RunStats: + """Counts for one pipeline run. + + ``skipped`` are rows rejected at the C.0 boundary (a clean, typed refusal); + ``errored`` are rows a later stage raised on, which are contained per row so + one bad row cannot discard the whole batch. The two are separate because a + boundary rejection is expected input hygiene while an error is a fault worth + investigating. + """ + + total: int + linked: int + review: int + skipped: int + errored: int = 0 + + +@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. Each component is structurally one + method (``KnowledgeSource`` / ``Retriever`` / ``Reranker`` / ``Scaler``), so + tests inject trivial stubs while the call sites stay type-checked. + + Rows are independent: a row rejected at the C.0 boundary counts as ``skipped`` + and a row whose later stages raise counts as ``errored``, and neither stops the + run. + """ + + def __init__( + self, + source: KnowledgeSource, + retriever: Retriever, + reranker: Reranker, + scaler: 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 = errored = 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 + + # Contain failures per row. With hermetic stubs nothing here raises, + # but these seams become live DB / embedding / cross-encoder calls in + # W8, where one timeout or one malformed candidate must not throw away + # every envelope the run has already built. + try: + 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 + ) + except Exception: + errored += 1 + logger.warning( + "librarian pipeline: chunk %s (artifact %s) failed after the C.0 " + "boundary; skipping this row", + section.chunk_id, + section.artifact_id, + exc_info=True, + ) + continue + + 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, + errored=errored, + ), + )