week_3: Module C C.1 — candidate retriever (in-memory + pgvector) + pipeline switch

The semantic search step. For sections with no explicit CRE id, embed the
text and cosine-rank the CRE vector hub, returning the top-K (default 20)
candidates as an RFC RetrievalAudit (reranked[] empty until W4).

Two interchangeable backends behind one retrieve() seam, selected by
CRE_LIBRARIAN_RETRIEVER_BACKEND:
- CandidateRetriever: in-memory sklearn cosine (SQLite/CI/harness)
- PgVectorRetriever: Postgres-side <=> cosine over embedding_vec
build_retriever() is the factory. Reuses OpenCRE's embedding stack
(PromptHandler.get_text_embeddings, db.get_embeddings_by_doc_type).

Dim gate fails loudly on empty/ragged hubs and query/hub width mismatch.
CLI switch (--run_librarian / --librarian_dry_run) runs the pipeline
dry-run; harness --use_live_embeddings measures retrieval recall@k.

The pgvector schema migration is a W8 deliverable (validated against real
Postgres there); W3 ships only the code path, unit-tested via a fake
connection. 82 librarian tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: PRAteek-singHWY

.env.example

@@ -45,3 +45,20 @@ VERTEX_EMBED_CONTENT_MODEL=gemini-embedding-001
 # Spreadsheet Auth
 
 OpenCRE_gspread_Auth=path/to/credentials.json
+
+# Module C — The Librarian (defaults match the OIE design doc; see
+# application/utils/librarian/config_loader.py for validation)
+
+# Retrieval backend: in_memory (sklearn cosine; SQLite dev/CI/harness) or
+# pgvector (Postgres vector column — pending prod extension + mentor OK).
+CRE_LIBRARIAN_RETRIEVER_BACKEND=in_memory
+# Candidate shortlist size produced by C.1 and reranked by C.2 (W4).
+CRE_LIBRARIAN_TOP_K_RETRIEVAL=20
+CRE_LIBRARIAN_TOP_K_RERANK=5
+# Cross-encoder reranker model (W4), pinned.
+CRE_LIBRARIAN_CROSSENCODER_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2
+# Auto-link confidence threshold; below this -> human review (W5).
+CRE_LIBRARIAN_LINK_THRESHOLD=0.8
+CRE_LIBRARIAN_BATCH_SIZE=32
+CRE_LIBRARIAN_ECE_TARGET=0.10
+CRE_LIBRARIAN_CONFORMAL_ALPHA=0.10

application/cmd/cre_main.py

@@ -962,6 +962,12 @@ def run(args: argparse.Namespace) -> None:  # pragma: no cover
         )
     if args.upstream_sync:
         download_graph_from_upstream(args.cache_file)
+    if args.run_librarian or args.librarian_dry_run:
+        run_librarian(
+            cache_file=args.cache_file,
+            dry_run=args.librarian_dry_run or not args.run_librarian,
+            source_jsonl=args.librarian_source,
+        )
 
 
 def ai_client_init(database: db.Node_collection):
@@ -1008,6 +1014,114 @@ def generate_embeddings(db_url: str) -> None:
     prompt_client.PromptHandler(database, load_all_embeddings=True)
 
 
+_DEFAULT_LIBRARIAN_SOURCE = os.path.join(
+    os.path.dirname(os.path.dirname(__file__)),
+    "tests",
+    "librarian",
+    "fixtures",
+    "sample_knowledge_queue.jsonl",
+)
+
+
+def run_librarian(
+    cache_file: str, dry_run: bool = True, source_jsonl: Optional[str] = None
+) -> None:
+    """Module C entrypoint — the pipeline switch (W3).
+
+    For each knowledge-queue section: try the deterministic explicit-CRE fast
+    path (C.0.5); on no/ambiguous reference, run the semantic retriever (C.1)
+    and log the top-K candidate shortlist. The cross-encoder rerank (C.2, W4),
+    decision/threshold routing (C.3-C.4, W5) and graph writes (W8) are not
+    built yet, so this is dry-run only: it never writes a link. ``--run_librarian``
+    without writes behaves identically and warns.
+    """
+    from application.utils.librarian.candidate_retriever import (
+        CandidatePool,
+        RetrieverBackend,
+        build_retriever,
+    )
+    from application.utils.librarian.config_loader import load_config
+    from application.utils.librarian.explicit_link_resolver import (
+        ResolutionOutcome,
+        resolve,
+    )
+    from application.utils.librarian.knowledge_source import FixtureKnowledgeSource
+    from application.utils.librarian.section_validator import (
+        SectionValidationError,
+        section_from_queue_row,
+    )
+
+    if not dry_run:
+        logger.warning(
+            "the Librarian cannot write links yet (DecisionEngine + graph write "
+            "land W8); running in dry-run mode"
+        )
+
+    cfg = load_config()
+    database = db_connect(path=cache_file)
+    ph = prompt_client.PromptHandler(database=database)
+
+    backend = RetrieverBackend(cfg.retriever_backend)
+    # The CRE ids present in the hub are exactly the known ids the explicit
+    # resolver may auto-link to (W2 seeded this from the golden set; here it is
+    # the real DB-backed registry).
+    cre_embeddings = database.get_embeddings_by_doc_type(defs.Credoctypes.CRE.value)
+    known_ids = set(cre_embeddings.keys())
+    # in_memory loads the hub matrix; pgvector ranks in the DB over the
+    # embedding_vec column (no in-RAM pool). Both honor the same retrieve().
+    pool = (
+        CandidatePool.from_mapping(cre_embeddings)
+        if backend is RetrieverBackend.in_memory
+        else None
+    )
+    retriever = build_retriever(
+        backend,
+        embed_fn=ph.get_text_embeddings,
+        top_k=cfg.top_k_retrieval,
+        threshold=cfg.link_threshold,
+        pool=pool,
+        connection=database.session.connection(),
+    )
+
+    source = FixtureKnowledgeSource(source_jsonl or _DEFAULT_LIBRARIAN_SOURCE)
+    sections = explicit = semantic = rejected = 0
+    for item in source.items():
+        try:
+            section = section_from_queue_row(item)
+        except SectionValidationError as exc:
+            rejected += 1
+            logger.warning("section rejected at C.0 boundary: %s", exc)
+            continue
+        sections += 1
+
+        resolution = resolve(section.text, known_ids)
+        if resolution.outcome == ResolutionOutcome.resolved:
+            explicit += 1
+            logger.info("[explicit] %s -> %s", section.chunk_id, resolution.cre_ids[0])
+            continue
+
+        semantic += 1
+        audit = retriever.retrieve(section.text)
+        top = ", ".join(
+            f"{c.cre_id}:{c.score_vector:.3f}" for c in audit.candidates[:5]
+        )
+        logger.info(
+            "[semantic] %s -> %d candidates (top5: %s)",
+            section.chunk_id,
+            len(audit.candidates),
+            top or "<none>",
+        )
+
+    logger.info(
+        "librarian dry-run complete: %d sections (%d explicit, %d semantic), "
+        "%d rejected at boundary",
+        sections,
+        explicit,
+        semantic,
+        rejected,
+    )
+
+
 def regenerate_embeddings(db_url: str) -> None:
     """Wipe all embedding rows, then rebuild (CRE + every node type) like ``--generate_embeddings``."""
     database = db_connect(path=db_url)

application/tests/librarian/candidate_retriever_test.py

@@ -0,0 +1,213 @@
+"""Tests for the C.1 semantic candidate retriever (Week 3).
+
+Hermetic: the embedding function and the CRE vector hub are injected as
+controlled vectors, so cosine ordering, top-K truncation, the dim gate, and
+the RetrievalAudit shape are all assertable without an LLM or a DB.
+"""
+
+import unittest
+
+from application.utils.librarian.candidate_retriever import (
+    PGVECTOR_RETRIEVER_NAME,
+    RETRIEVER_NAME,
+    CandidatePool,
+    CandidateRetriever,
+    DimensionMismatchError,
+    EmptyPoolError,
+    PgVectorRetriever,
+    RetrieverBackend,
+    RetrieverError,
+    build_retriever,
+    to_pgvector_literal,
+)
+
+# A controlled hub. Query [1,0,0] -> cosine a=1.0, c=0.707, b=0.0 -> order a,c,b.
+HUB = {
+    "111-111": [1.0, 0.0, 0.0],  # "a"
+    "222-222": [0.0, 1.0, 0.0],  # "b"
+    "333-333": [1.0, 1.0, 0.0],  # "c"
+}
+
+# Deterministic embedder: text -> a fixed query vector.
+_VECTORS = {
+    "about-a": [1.0, 0.0, 0.0],
+    "about-b": [0.0, 1.0, 0.0],
+    "wrong-width": [1.0, 0.0],
+}
+
+
+def fake_embed(text):
+    return _VECTORS[text]
+
+
+def make_retriever(top_k=20, threshold=0.8, cre_names=None):
+    return CandidateRetriever(
+        embed_fn=fake_embed,
+        pool=CandidatePool.from_mapping(HUB),
+        top_k=top_k,
+        threshold=threshold,
+        cre_names=cre_names,
+    )
+
+
+class CandidatePoolTest(unittest.TestCase):
+    def test_from_mapping_builds_aligned_matrix(self) -> None:
+        pool = CandidatePool.from_mapping(HUB)
+        self.assertEqual(pool.dim, 3)
+        self.assertEqual(set(pool.cre_ids), set(HUB))
+        self.assertEqual(pool.matrix.shape, (3, 3))
+
+    def test_empty_pool_rejected(self) -> None:
+        with self.assertRaises(EmptyPoolError):
+            CandidatePool.from_mapping({})
+
+    def test_ragged_vectors_rejected(self) -> None:
+        with self.assertRaises(DimensionMismatchError):
+            CandidatePool.from_mapping({"a": [1.0, 0.0], "b": [1.0]})
+
+    def test_zero_width_vectors_rejected(self) -> None:
+        with self.assertRaises(DimensionMismatchError):
+            CandidatePool.from_mapping({"a": [], "b": []})
+
+
+class RetrieveTest(unittest.TestCase):
+    def test_candidates_rank_ordered_by_cosine(self) -> None:
+        audit = make_retriever().retrieve("about-a")
+        self.assertEqual(
+            [c.cre_id for c in audit.candidates],
+            ["111-111", "333-333", "222-222"],
+        )
+        # score_vector is the cosine; populated and descending.
+        scores = [c.score_vector for c in audit.candidates]
+        self.assertAlmostEqual(scores[0], 1.0)
+        self.assertEqual(scores, sorted(scores, reverse=True))
+
+    def test_top_k_truncates_and_caps_at_pool_size(self) -> None:
+        # top_k larger than the hub returns the whole hub, no error.
+        self.assertEqual(
+            len(make_retriever(top_k=20).retrieve("about-a").candidates), 3
+        )
+        # top_k smaller than the hub returns exactly the best k.
+        top1 = make_retriever(top_k=1).retrieve("about-a")
+        self.assertEqual([c.cre_id for c in top1.candidates], ["111-111"])
+
+    def test_audit_shape_for_w3(self) -> None:
+        audit = make_retriever(threshold=0.8).retrieve("about-b")
+        self.assertEqual(audit.retriever, RETRIEVER_NAME)
+        self.assertEqual(audit.reranked, [])  # cross-encoder lands W4
+        self.assertEqual(audit.threshold, 0.8)
+        # The closest to [0,1,0] is b, then c (0.707), then a (0.0).
+        self.assertEqual(audit.candidates[0].cre_id, "222-222")
+
+    def test_cre_names_populated_when_provided(self) -> None:
+        audit = make_retriever(cre_names={"111-111": "Authentication"}).retrieve(
+            "about-a"
+        )
+        self.assertEqual(audit.candidates[0].cre_name, "Authentication")
+        # Unnamed candidates stay None, never KeyError.
+        self.assertIsNone(audit.candidates[-1].cre_name)
+
+    def test_query_dim_mismatch_is_caught(self) -> None:
+        with self.assertRaises(DimensionMismatchError):
+            make_retriever().retrieve("wrong-width")
+
+
+class ConstructionTest(unittest.TestCase):
+    def test_non_positive_top_k_rejected(self) -> None:
+        with self.assertRaises(RetrieverError):
+            make_retriever(top_k=0)
+
+
+# --- pgvector backend (hermetic: the DB is a fake recording connection) ---
+
+
+class _FakeRow:
+    def __init__(self, cre_id, score):
+        self.cre_id = cre_id
+        self.score = score
+
+
+class _FakeResult:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def fetchall(self):
+        return self._rows
+
+
+class _FakeConnection:
+    """Records the last execute() call and returns canned, pre-ordered rows."""
+
+    def __init__(self, rows):
+        self._rows = rows
+        self.last_sql = None
+        self.last_params = None
+
+    def execute(self, sql, params):
+        self.last_sql = str(sql)
+        self.last_params = params
+        return _FakeResult(self._rows)
+
+
+class PgVectorRetrieverTest(unittest.TestCase):
+    def test_pgvector_literal_format(self) -> None:
+        self.assertEqual(to_pgvector_literal([1, 2.5, 0]), "[1.0,2.5,0.0]")
+
+    def test_parses_db_rows_in_order(self) -> None:
+        # The DB does the ranking; the retriever preserves ORDER BY order.
+        conn = _FakeConnection([_FakeRow("111-111", 0.97), _FakeRow("333-333", 0.71)])
+        retriever = PgVectorRetriever(
+            embed_fn=fake_embed, connection=conn, top_k=20, threshold=0.8
+        )
+        audit = retriever.retrieve("about-a")
+        self.assertEqual(audit.retriever, PGVECTOR_RETRIEVER_NAME)
+        self.assertEqual([c.cre_id for c in audit.candidates], ["111-111", "333-333"])
+        self.assertAlmostEqual(audit.candidates[0].score_vector, 0.97)
+        self.assertEqual(audit.reranked, [])
+
+    def test_binds_query_vector_doctype_and_limit(self) -> None:
+        conn = _FakeConnection([])
+        PgVectorRetriever(
+            embed_fn=fake_embed, connection=conn, top_k=7, threshold=0.8
+        ).retrieve("about-a")
+        self.assertEqual(conn.last_params["q"], "[1.0,0.0,0.0]")
+        self.assertEqual(conn.last_params["doc_type"], "CRE")
+        self.assertEqual(conn.last_params["k"], 7)
+        # Cosine via the <=> operator, scored as similarity (1 - distance).
+        self.assertIn("<=>", conn.last_sql)
+
+
+class BuildRetrieverTest(unittest.TestCase):
+    def test_in_memory_requires_pool(self) -> None:
+        with self.assertRaises(RetrieverError):
+            build_retriever(
+                RetrieverBackend.in_memory, fake_embed, top_k=20, threshold=0.8
+            )
+
+    def test_pgvector_requires_connection(self) -> None:
+        with self.assertRaises(RetrieverError):
+            build_retriever(
+                RetrieverBackend.pgvector, fake_embed, top_k=20, threshold=0.8
+            )
+
+    def test_factory_routes_to_each_backend(self) -> None:
+        in_mem = build_retriever(
+            RetrieverBackend.in_memory,
+            fake_embed,
+            top_k=20,
+            threshold=0.8,
+            pool=CandidatePool.from_mapping(HUB),
+        )
+        self.assertIsInstance(in_mem, CandidateRetriever)
+        pg = build_retriever(
+            RetrieverBackend.pgvector,
+            fake_embed,
+            top_k=20,
+            threshold=0.8,
+            connection=_FakeConnection([]),
+        )
+        self.assertIsInstance(pg, PgVectorRetriever)
+
+
+if __name__ == "__main__":
+    unittest.main()