week_2: Module C C.0 — section validator (input boundary) + ExplicitLinkResolver

The C.0 deterministic input boundary, per the proposal's W2/W3 'data
preparation layer' (named validator, not normalizer: the RFC assigns text
normalization to Module A; C validates and adapts, never transforms text).

- section_validator.py: typed-error validation of both upstream shapes
  (B's knowledge_queue row + RFC KnowledgeItem envelope) into an internal
  Section; synthesizes RFC identity fields (chunk_id/artifact_id/source/
  locator) from B's reduced row; strips volatile audit metadata.
- explicit_link_resolver.py: deterministic ddd-ddd fast path, no ML.
  Fail-safe: only a single known reference auto-links; unknown or
  conflicting references route to review.
- evaluate_librarian.py: harness now runs every golden row through C.0,
  prints per-slice validation pass rate, and gates the explicit slice at
  100% resolver correctness (exit 1 on regression). Gate: PASS 5/5.
- Table-driven tests for every rejection class and resolver outcome.
  mypy --strict clean, black clean, 66 tests green.

Author: PRAteek-singHWY

application/tests/librarian/explicit_link_resolver_test.py

@@ -0,0 +1,83 @@
+"""Tests for the C.0.5 deterministic explicit-CRE fast path.
+
+Covers extraction (pattern boundaries, ordering, dedup) and resolution
+(the fail-safe rule: only a single known reference auto-links; unknown or
+conflicting references must fall through to review).
+"""
+
+import unittest
+
+from application.utils.librarian.explicit_link_resolver import (
+    Resolution,
+    ResolutionOutcome,
+    extract_cre_refs,
+    resolve,
+)
+
+KNOWN = {"027-555", "123-456", "764-507"}
+
+
+class ExtractCreRefsTest(unittest.TestCase):
+    def test_extraction_table(self) -> None:
+        cases = [
+            ("plain reference", "Per CRE 027-555, verify passwords.", ["027-555"]),
+            (
+                "inside an opencre url",
+                "See https://opencre.org/cre/123-456 for details.",
+                ["123-456"],
+            ),
+            (
+                "multiple distinct, in order",
+                "Maps to 123-456 and also 027-555.",
+                ["123-456", "027-555"],
+            ),
+            ("repeated reference deduped", "027-555 then 027-555 again.", ["027-555"]),
+            ("no reference", "Use MFA for all administrative access.", []),
+            ("too many leading digits", "CVE-2024-1234-567 is unrelated.", []),
+            ("too many trailing digits", "Item 027-5555 is not a CRE id.", []),
+            ("punctuation boundary", "(see CRE 764-507).", ["764-507"]),
+        ]
+        for name, text, expected in cases:
+            with self.subTest(name):
+                self.assertEqual(extract_cre_refs(text), expected)
+
+
+class ResolveTest(unittest.TestCase):
+    def test_single_known_reference_resolves(self) -> None:
+        resolution = resolve("Per CRE 027-555, verify passwords.", KNOWN)
+        self.assertEqual(
+            resolution,
+            Resolution(ResolutionOutcome.resolved, ("027-555",), ()),
+        )
+
+    def test_no_reference_falls_through_to_semantic_path(self) -> None:
+        resolution = resolve("Use MFA for all administrative access.", KNOWN)
+        self.assertEqual(resolution.outcome, ResolutionOutcome.no_reference)
+        self.assertEqual(resolution.cre_ids, ())
+
+    def test_unknown_reference_never_auto_links(self) -> None:
+        resolution = resolve("Per CRE 999-999, do the thing.", KNOWN)
+        self.assertEqual(resolution.outcome, ResolutionOutcome.unknown_reference)
+        self.assertEqual(resolution.cre_ids, ())
+        self.assertEqual(resolution.unknown_refs, ("999-999",))
+
+    def test_mixed_known_and_unknown_routes_to_review(self) -> None:
+        resolution = resolve("Maps to 027-555 and 999-999.", KNOWN)
+        self.assertEqual(resolution.outcome, ResolutionOutcome.unknown_reference)
+        # The known id survives as a review suggestion, but must not auto-link.
+        self.assertEqual(resolution.cre_ids, ("027-555",))
+        self.assertEqual(resolution.unknown_refs, ("999-999",))
+
+    def test_conflicting_known_references_route_to_review(self) -> None:
+        resolution = resolve("Maps to 123-456 and also 027-555.", KNOWN)
+        self.assertEqual(resolution.outcome, ResolutionOutcome.conflicting_references)
+        self.assertEqual(resolution.cre_ids, ("123-456", "027-555"))
+
+    def test_repeated_single_reference_still_resolves(self) -> None:
+        resolution = resolve("027-555 is cited twice: 027-555.", KNOWN)
+        self.assertEqual(resolution.outcome, ResolutionOutcome.resolved)
+        self.assertEqual(resolution.cre_ids, ("027-555",))
+
+
+if __name__ == "__main__":
+    unittest.main()

application/tests/librarian/section_validator_test.py

@@ -0,0 +1,182 @@
+"""Tests for the C.0 input boundary (section_validator).
+
+Table-driven over every rejection class plus the happy paths for both
+upstream shapes (knowledge_queue row and RFC KnowledgeItem envelope).
+Asserts the boundary never leaks a raw Pydantic ValidationError.
+"""
+
+import unittest
+
+from pydantic import ValidationError
+
+from application.utils.librarian.section_validator import (
+    EmptyTextError,
+    MalformedKnowledgeItemError,
+    NotKnowledgeError,
+    Section,
+    SectionValidationError,
+    UnsupportedLanguageError,
+    section_from_knowledge_item,
+    section_from_queue_row,
+)
+
+
+def valid_queue_row(**overrides) -> dict:
+    row = {
+        "id": "4a8c1b2e-1d2f-4e3a-9b4c-5d6e7f8a9b0c",
+        "source_repo": "OWASP/ASVS",
+        "source_path": "4.0/en/0x11-V2-Authentication.md",
+        "source_commit_sha": "abc123def456789012345678901234567890abcd",
+        "text": "Verify that user-set passwords are at least 12 characters long.",
+        "confidence": 0.93,
+        "llm_label": "KNOWLEDGE",
+        "llm_reasoning": "clear security requirement",
+        "created_at": "2026-05-25T02:25:00Z",
+        "consumed_at": None,
+    }
+    row.update(overrides)
+    return row
+
+
+def valid_knowledge_item(**overrides) -> dict:
+    item = {
+        "schema_version": "0.2.0",
+        "chunk_id": "chk:art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md:0",
+        "artifact_id": "art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md",
+        "event_id": "evt-001",
+        "pipeline_run_id": "20260601T020000Z",
+        "filtered_at": "2026-06-01T02:10:00Z",
+        "status": "accepted",
+        "source": {
+            "type": "github",
+            "repo": "OWASP/ASVS",
+            "commit_sha": "abc123def456789012345678901234567890abcd",
+            "committed_at": "2026-06-01T01:00:00Z",
+        },
+        "locator": {
+            "kind": "repo_path",
+            "id": "4.0/en/0x11-V2-Authentication.md",
+            "path": "4.0/en/0x11-V2-Authentication.md",
+        },
+        "content": {
+            "text": "Verify that user-set passwords are at least 12 characters long.",
+            "title_hint": "Password length",
+            "language": "en",
+        },
+        "filter": {
+            "stages": [{"name": "llm_relevance", "passed": True}],
+            "is_security_knowledge": True,
+            "confidence": 0.93,
+        },
+    }
+    item.update(overrides)
+    return item
+
+
+class QueueRowBoundaryTest(unittest.TestCase):
+    def test_valid_row_builds_section_with_synthesized_identity(self) -> None:
+        section = section_from_queue_row(valid_queue_row())
+        self.assertIsInstance(section, Section)
+        self.assertEqual(
+            section.chunk_id,
+            "chk:OWASP/ASVS@abc123def456789012345678901234567890abcd:"
+            "4.0/en/0x11-V2-Authentication.md",
+        )
+        self.assertEqual(
+            section.artifact_id, "art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md"
+        )
+        self.assertEqual(section.source.repo, "OWASP/ASVS")
+        self.assertEqual(section.source.committed_at, "2026-05-25T02:25:00Z")
+        self.assertEqual(section.locator.path, "4.0/en/0x11-V2-Authentication.md")
+        self.assertEqual(section.language, "en")
+
+    def test_volatile_metadata_not_carried_into_section(self) -> None:
+        section = section_from_queue_row(
+            valid_queue_row(llm_reasoning="audit-only rationale")
+        )
+        self.assertFalse(hasattr(section, "llm_reasoning"))
+        self.assertFalse(hasattr(section, "confidence"))
+
+    def test_rejection_table(self) -> None:
+        cases = [
+            ("empty text", valid_queue_row(text=""), EmptyTextError),
+            ("whitespace text", valid_queue_row(text="  \n\t "), EmptyTextError),
+            ("noise label", valid_queue_row(llm_label="NOISE"), NotKnowledgeError),
+            (
+                "uncertain label",
+                valid_queue_row(llm_label="UNCERTAIN"),
+                NotKnowledgeError,
+            ),
+            (
+                "missing field",
+                {k: v for k, v in valid_queue_row().items() if k != "source_repo"},
+                MalformedKnowledgeItemError,
+            ),
+            (
+                "wrong type",
+                valid_queue_row(confidence="very sure"),
+                MalformedKnowledgeItemError,
+            ),
+            ("not a mapping", "just a string", MalformedKnowledgeItemError),
+        ]
+        for name, row, expected_error in cases:
+            with self.subTest(name):
+                with self.assertRaises(expected_error):
+                    section_from_queue_row(row)
+
+    def test_never_leaks_raw_pydantic_error(self) -> None:
+        try:
+            section_from_queue_row({"id": "x"})
+        except SectionValidationError as exc:
+            self.assertNotIsInstance(exc, ValidationError)
+            self.assertIsInstance(exc.__cause__, ValidationError)
+        else:
+            self.fail("expected SectionValidationError")
+
+
+class KnowledgeItemBoundaryTest(unittest.TestCase):
+    def test_valid_item_builds_section(self) -> None:
+        section = section_from_knowledge_item(valid_knowledge_item())
+        self.assertEqual(
+            section.chunk_id, "chk:art:OWASP/ASVS:4.0/en/0x11-V2-Authentication.md:0"
+        )
+        self.assertEqual(section.title_hint, "Password length")
+        self.assertEqual(section.language, "en")
+
+    def test_missing_language_defaults_to_english(self) -> None:
+        item = valid_knowledge_item()
+        del item["content"]["language"]
+        self.assertEqual(section_from_knowledge_item(item).language, "en")
+
+    def test_regional_english_variant_is_accepted(self) -> None:
+        item = valid_knowledge_item()
+        item["content"]["language"] = "en-GB"
+        self.assertEqual(section_from_knowledge_item(item).language, "en-GB")
+
+    def test_rejection_table(self) -> None:
+        rejected = valid_knowledge_item(
+            status="rejected",
+            content=None,
+            rejection={"reason_code": "NOT_SECURITY"},
+        )
+        unsupported_lang = valid_knowledge_item()
+        unsupported_lang["content"]["language"] = "fr"
+        whitespace_text = valid_knowledge_item()
+        whitespace_text["content"]["text"] = "   "
+        malformed = valid_knowledge_item()
+        del malformed["source"]
+
+        cases = [
+            ("status rejected", rejected, NotKnowledgeError),
+            ("unsupported language", unsupported_lang, UnsupportedLanguageError),
+            ("whitespace text", whitespace_text, EmptyTextError),
+            ("missing source", malformed, MalformedKnowledgeItemError),
+        ]
+        for name, item, expected_error in cases:
+            with self.subTest(name):
+                with self.assertRaises(expected_error):
+                    section_from_knowledge_item(item)
+
+
+if __name__ == "__main__":
+    unittest.main()

application/utils/librarian/explicit_link_resolver.py

@@ -0,0 +1,73 @@
+"""Module C.0.5 — deterministic explicit-CRE fast path (Week 2). No ML.
+
+If a section's text already cites a CRE id (``ddd-ddd``, plain or inside an
+``opencre.org/cre/<id>`` link), resolve it directly against the set of known
+CRE ids and bypass retrieval entirely. The fail-safe rule from the proposal:
+unknown or mutually conflicting references never auto-link — they fall
+through to human review; only a single, known reference resolves.
+
+Gate (PR 2): 100% correctness on the explicit golden-dataset slice. Any
+regression here is a merge blocker.
+
+The known-id set is injected (``Container[str]``) so this module stays
+dependency-free: the harness seeds it from the golden dataset today; the
+DB-backed registry of real ``cre.external_id`` values arrives with the
+retriever (W3).
+"""
+
+import re
+from dataclasses import dataclass
+from enum import Enum
+from typing import Container, List, Tuple
+
+# Word boundaries keep e.g. "1234-567" and "027-5555" from partially matching.
+CRE_ID_RE = re.compile(r"\b\d{3}-\d{3}\b")
+
+
+class ResolutionOutcome(str, Enum):
+    # No CRE reference in the text — continue to the semantic path (C.1+).
+    no_reference = "no_reference"
+    # Exactly one known CRE id — deterministic auto-link, skip retrieval.
+    resolved = "resolved"
+    # Reference(s) found but none/some are known CRE ids — route to review.
+    unknown_reference = "unknown_reference"
+    # Multiple distinct known ids in one section — ambiguous, route to review.
+    conflicting_references = "conflicting_references"
+
+
+@dataclass(frozen=True)
+class Resolution:
+    outcome: ResolutionOutcome
+    # Known ids, deduped, in order of first appearance. Non-empty iff some
+    # reference resolved; for conflicting outcomes these become the
+    # ReviewItem's suggested_links.
+    cre_ids: Tuple[str, ...]
+    # References that matched the CRE-id pattern but are not known ids.
+    unknown_refs: Tuple[str, ...]
+
+
+def extract_cre_refs(text: str) -> List[str]:
+    """All CRE-id-shaped references in the text, deduped, in order."""
+    seen = set()
+    refs = []
+    for match in CRE_ID_RE.findall(text):
+        if match not in seen:
+            seen.add(match)
+            refs.append(match)
+    return refs
+
+
+def resolve(text: str, known_cre_ids: Container[str]) -> Resolution:
+    """Deterministically resolve explicit CRE references in one section."""
+    refs = extract_cre_refs(text)
+    if not refs:
+        return Resolution(ResolutionOutcome.no_reference, (), ())
+
+    known = tuple(ref for ref in refs if ref in known_cre_ids)
+    unknown = tuple(ref for ref in refs if ref not in known_cre_ids)
+
+    if unknown:
+        return Resolution(ResolutionOutcome.unknown_reference, known, unknown)
+    if len(known) > 1:
+        return Resolution(ResolutionOutcome.conflicting_references, known, ())
+    return Resolution(ResolutionOutcome.resolved, known, ())