From a0418c539b5acfc806b5e28b89baca9542a8fbb4 Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sat, 24 Oct 2026 00:46:42 +0530
Subject: [PATCH 1/8] Add POST /rest/v1/link/suggest: the Librarian's shortlist
 as an API

Fixes #275.

Given a CRE Standard document, return the CREs most likely to be its link
targets, ranked. This is C.1 of the Librarian exposed directly, so a standard
author can ask "where does this section belong?" without running the OIE
pipeline, and without a second similarity stack growing beside the one that is
evaluated and gated in CI (the maintainer decision on #965).

Retrieval only, deliberately: C.2 and C.3 need a torch cross-encoder, which the
web process must not load (the same constraint #1161 respects). The score is
the retriever's cosine similarity, named as such. Automated mapping with a
confidence threshold stays with the worker-side mapping job.

factory.build_retriever_for is split out of build_components with no change
in behaviour, so the endpoint and the pipeline construct C.1 from one place.

Results always carry the public CRE id. The embeddings table keys vectors by
the CRE row's internal UUID, so both retriever backends return UUIDs; each
candidate is resolved through the graph (internal id first, external second)
and a stale embedding for a CRE the graph no longer has is returned with the
retriever's label rather than silently dropped.

top_k is 1 to 20, default 5, whole numbers only: int() would truncate 2.5 to
2 and read True as 1, and a mistyped limit should be refused, not reinterpreted.

Same auth and the same unavailable/rate-limit mapping as /rest/v1/completion,
since both call the paid embedding API.

26 hermetic tests: the service with a stub retriever against an in-memory
graph (including a UUID-keyed candidate resolving to "287-251"), and the real
Flask route with only the builder seam patched. Verified against the real CRE
pool in a local cache with a stored vector as the query: the issue's own ASVS
V2.9.2 example returns 287-251 first, then 585-408 Challenge nonce
cryptography. The paid-API call path is PromptHandler.get_text_embeddings,
unchanged and shared with chat.
---
 .../tests/librarian/link_suggest_test.py      | 320 ++++++++++++++++++
 application/utils/librarian/factory.py        |  62 +++-
 application/utils/librarian/link_suggest.py   | 190 +++++++++++
 application/web/web_main.py                   |  67 ++++
 4 files changed, 621 insertions(+), 18 deletions(-)
 create mode 100644 application/tests/librarian/link_suggest_test.py
 create mode 100644 application/utils/librarian/link_suggest.py

diff --git a/application/tests/librarian/link_suggest_test.py b/application/tests/librarian/link_suggest_test.py
new file mode 100644
index 000000000..e153d368e
--- /dev/null
+++ b/application/tests/librarian/link_suggest_test.py
@@ -0,0 +1,320 @@
+"""Tests for ``POST /rest/v1/link/suggest`` and the service behind it (#275).
+
+Hermetic: the retriever is a stub returning a fixed ``RetrievalAudit``, so no
+embedding call is made. The web tests go through the real Flask route with the
+builder seam patched, which is the only way to prove the route's validation,
+auth bypass and error mapping without a model or a key.
+"""
+
+from cre_logging import get_logger
+
+logger = get_logger(__name__)
+
+import os
+import unittest
+from unittest import mock
+
+from application import create_app, sqla
+from application.database import db
+from application.defs import cre_defs as defs
+from application.utils.librarian import link_suggest
+from application.utils.librarian.link_suggest import (
+    DEFAULT_TOP_K,
+    MAX_TOP_K,
+    InvalidStandardError,
+    clamp_top_k,
+    standard_query_text,
+    suggest_links,
+)
+from application.utils.librarian.schemas import CreCandidate, RetrievalAudit
+
+
+def _audit() -> RetrievalAudit:
+    return RetrievalAudit(
+        retriever="stub/1.0.0",
+        candidates=[
+            CreCandidate(
+                cre_id="287-251", cre_name="from retriever", score_vector=0.91234
+            ),
+            CreCandidate(cre_id="000-001", score_vector=0.5),
+        ],
+        reranked=[],
+        threshold=0.8,
+    )
+
+
+class _StubRetriever:
+    def __init__(self) -> None:
+        self.calls = []
+
+    def retrieve(self, text: str) -> RetrievalAudit:
+        self.calls.append(text)
+        return _audit()
+
+
+def _standard(**overrides):
+    doc = {
+        "doctype": "Standard",
+        "name": "ASVS",
+        "sectionID": "V2.9.2",
+        "section": (
+            "Verify that the challenge nonce is at least 64 bits in length, and "
+            "statistically unique or unique over the lifetime of the cryptographic device."
+        ),
+        "hyperlink": "https://github.com/OWASP/ASVS/blob/v4.0.2/4.0/en/0x11-V2-Authentication.md",
+        "links": [],
+    }
+    doc.update(overrides)
+    return doc
+
+
+class StandardQueryTextTest(unittest.TestCase):
+    def test_joins_the_text_fields_and_ignores_provenance(self) -> None:
+        text = standard_query_text(
+            _standard(subsection="Nonce size", description="Use 64-bit nonces")
+        )
+        self.assertTrue(text.startswith("Verify that the challenge nonce"))
+        self.assertIn("Nonce size. Use 64-bit nonces", text)
+        self.assertNotIn("ASVS", text)
+        self.assertNotIn("V2.9.2", text)
+
+    def test_rejects_non_standard_doctype(self) -> None:
+        with self.assertRaises(InvalidStandardError):
+            standard_query_text(_standard(doctype="CRE"))
+
+    def test_rejects_missing_name(self) -> None:
+        with self.assertRaises(InvalidStandardError):
+            standard_query_text(_standard(name=""))
+
+    def test_rejects_a_document_with_no_text_to_embed(self) -> None:
+        with self.assertRaises(InvalidStandardError):
+            standard_query_text(_standard(section="", subsection="", description=""))
+
+    def test_rejects_non_string_text_fields(self) -> None:
+        with self.assertRaises(InvalidStandardError):
+            standard_query_text(_standard(section=["not", "a", "string"]))
+
+    def test_rejects_non_object_bodies(self) -> None:
+        with self.assertRaises(InvalidStandardError):
+            standard_query_text(["a", "list"])  # type: ignore[arg-type]
+
+
+class ClampTopKTest(unittest.TestCase):
+    def test_absent_means_default(self) -> None:
+        self.assertEqual(clamp_top_k(None), DEFAULT_TOP_K)
+        self.assertEqual(clamp_top_k(""), DEFAULT_TOP_K)
+
+    def test_parses_and_caps(self) -> None:
+        self.assertEqual(clamp_top_k("7"), 7)
+        self.assertEqual(clamp_top_k(99), MAX_TOP_K)
+
+    def test_rejects_non_positive_and_non_integer(self) -> None:
+        for bad in ("0", "-3", "x", 2.5):
+            with self.assertRaises(InvalidStandardError):
+                clamp_top_k(bad)
+
+
+class SuggestLinksTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.app = create_app(mode="test")
+        self.ctx = self.app.app_context()
+        self.ctx.push()
+        sqla.create_all()
+        self.collection = db.Node_collection()
+        self.added = self.collection.add_cre(
+            defs.CRE(
+                id="287-251",
+                name="Use a unique challenge nonce of sufficient size",
+                tags=["Cryptography"],
+            )
+        )
+
+    def tearDown(self) -> None:
+        sqla.session.remove()
+        sqla.drop_all()
+        self.ctx.pop()
+
+    def test_internal_uuid_candidates_resolve_to_the_public_cre_id(self) -> None:
+        """The embeddings table keys vectors by the CRE row's internal UUID, so
+        that is what the retriever returns. A caller must only ever see
+        "287-251"; this is the bug a stub keyed by external ids would hide."""
+        internal_id = self.added.id
+
+        class _UuidRetriever:
+            def retrieve(self, text):
+                return RetrievalAudit(
+                    retriever="stub/1.0.0",
+                    candidates=[CreCandidate(cre_id=internal_id, score_vector=0.9)],
+                    reranked=[],
+                    threshold=0.8,
+                )
+
+        (result,) = suggest_links(
+            self.collection, _standard(), retriever=_UuidRetriever()
+        )
+
+        self.assertEqual(result["id"], "287-251")
+        self.assertEqual(
+            result["name"], "Use a unique challenge nonce of sufficient size"
+        )
+        self.assertEqual(result["tags"], ["Cryptography"])
+
+    def test_returns_the_issue_275_shape_best_first(self) -> None:
+        results = suggest_links(
+            self.collection, _standard(), retriever=_StubRetriever()
+        )
+
+        self.assertEqual(
+            results[0],
+            {
+                "doctype": "CRE",
+                "id": "287-251",
+                "name": "Use a unique challenge nonce of sufficient size",
+                "tags": ["Cryptography"],
+                "score": 0.9123,
+            },
+        )
+
+    def test_a_candidate_missing_from_the_graph_is_returned_not_dropped(self) -> None:
+        """A stale embedding for a CRE the graph no longer has is still a
+        result the caller should see; silently dropping it would hide the
+        drift that caused it."""
+        results = suggest_links(
+            self.collection, _standard(), retriever=_StubRetriever()
+        )
+
+        self.assertEqual(results[1]["id"], "000-001")
+        self.assertEqual(results[1]["name"], "")
+        self.assertEqual(results[1]["tags"], [])
+
+    def test_graph_name_wins_over_the_retrievers_label(self) -> None:
+        results = suggest_links(
+            self.collection, _standard(), retriever=_StubRetriever()
+        )
+        self.assertNotEqual(results[0]["name"], "from retriever")
+
+    def test_top_k_slices_the_shortlist(self) -> None:
+        results = suggest_links(
+            self.collection, _standard(), top_k=1, retriever=_StubRetriever()
+        )
+        self.assertEqual(len(results), 1)
+
+    def test_top_k_outside_bounds_is_refused(self) -> None:
+        for bad in (0, MAX_TOP_K + 1):
+            with self.assertRaises(InvalidStandardError):
+                suggest_links(
+                    self.collection, _standard(), top_k=bad, retriever=_StubRetriever()
+                )
+
+    def test_embeds_the_section_text_not_the_whole_document(self) -> None:
+        stub = _StubRetriever()
+        suggest_links(self.collection, _standard(), retriever=stub)
+        self.assertEqual(len(stub.calls), 1)
+        self.assertTrue(stub.calls[0].startswith("Verify that the challenge nonce"))
+
+    def test_a_supplied_retriever_means_no_builder_is_invoked(self) -> None:
+        def _must_not_run(database):
+            raise AssertionError("builder must not be called")
+
+        suggest_links(
+            self.collection,
+            _standard(),
+            retriever=_StubRetriever(),
+            retriever_factory=_must_not_run,
+        )
+
+    def test_invalid_document_fails_before_any_retrieval(self) -> None:
+        stub = _StubRetriever()
+        with self.assertRaises(InvalidStandardError):
+            suggest_links(self.collection, _standard(section=""), retriever=stub)
+        self.assertEqual(stub.calls, [])
+
+
+class LinkSuggestRouteTest(unittest.TestCase):
+    """The real Flask route, with only the retriever builder seam patched."""
+
+    def setUp(self) -> None:
+        self.app = create_app(mode="test")
+        self.ctx = self.app.app_context()
+        self.ctx.push()
+        sqla.create_all()
+        os.environ["INSECURE_REQUESTS"] = "True"
+        os.environ["NO_LOGIN"] = "1"
+        self.collection = db.Node_collection()
+        self.collection.add_cre(
+            defs.CRE(
+                id="287-251", name="Use a unique challenge nonce", tags=["Cryptography"]
+            )
+        )
+
+    def tearDown(self) -> None:
+        os.environ.pop("NO_LOGIN", None)
+        sqla.session.remove()
+        sqla.drop_all()
+        self.ctx.pop()
+
+    def _post(self, body, query=""):
+        with mock.patch.object(
+            link_suggest, "_default_retriever", return_value=_StubRetriever()
+        ):
+            with self.app.test_client() as c:
+                return c.post(f"/rest/v1/link/suggest{query}", json=body)
+
+    def test_returns_ranked_cres(self) -> None:
+        resp = self._post(_standard())
+        self.assertEqual(resp.status_code, 200, resp.get_data(as_text=True))
+        body = resp.get_json()
+        self.assertEqual([r["id"] for r in body], ["287-251", "000-001"])
+        self.assertEqual(body[0]["doctype"], "CRE")
+        self.assertEqual(body[0]["tags"], ["Cryptography"])
+
+    def test_top_k_from_the_query_string(self) -> None:
+        resp = self._post(_standard(), query="?top_k=1")
+        self.assertEqual(len(resp.get_json()), 1)
+
+    def test_non_object_body_is_400(self) -> None:
+        resp = self._post(["not", "an", "object"])
+        self.assertEqual(resp.status_code, 400)
+
+    def test_document_without_text_is_400_with_the_reason(self) -> None:
+        resp = self._post(_standard(section=""))
+        self.assertEqual(resp.status_code, 400)
+        self.assertIn("section", resp.get_data(as_text=True))
+
+    def test_bad_top_k_is_400(self) -> None:
+        resp = self._post(_standard(), query="?top_k=zero")
+        self.assertEqual(resp.status_code, 400)
+
+    def test_missing_ml_dependencies_is_503(self) -> None:
+        with mock.patch.object(
+            link_suggest, "_default_retriever", side_effect=ImportError("no torch")
+        ):
+            with self.app.test_client() as c:
+                resp = c.post("/rest/v1/link/suggest", json=_standard())
+        self.assertEqual(resp.status_code, 503)
+
+    def test_missing_litellm_is_503(self) -> None:
+        with mock.patch.object(
+            link_suggest,
+            "_default_retriever",
+            side_effect=RuntimeError("litellm is not installed"),
+        ):
+            with self.app.test_client() as c:
+                resp = c.post("/rest/v1/link/suggest", json=_standard())
+        self.assertEqual(resp.status_code, 503)
+
+    def test_provider_failure_is_reported_not_swallowed(self) -> None:
+        class _Boom(Exception):
+            status_code = 502
+
+        with mock.patch.object(
+            link_suggest, "_default_retriever", side_effect=_Boom("upstream down")
+        ):
+            with self.app.test_client() as c:
+                resp = c.post("/rest/v1/link/suggest", json=_standard())
+        self.assertEqual(resp.status_code, 502)
+        self.assertIn("error", resp.get_json())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/application/utils/librarian/factory.py b/application/utils/librarian/factory.py
index d9f4ba118..2de90e53a 100644
--- a/application/utils/librarian/factory.py
+++ b/application/utils/librarian/factory.py
@@ -62,26 +62,22 @@ def build_scaler(config: Optional[LibrarianConfig] = None) -> Scaler:
     return TemperatureScaler(config.temperature)
 
 
-def build_components(
+def build_retriever_for(
     database: Any,
     *,
     config: Optional[LibrarianConfig] = None,
     embed_fn: Optional[Callable[[str], Sequence[float]]] = None,
-) -> LibrarianComponents:
-    """Construct C.1 + C.2 + C.3 against a connected OpenCRE database.
+) -> Any:
+    """Construct C.1 alone against a connected OpenCRE database.
 
-    Args:
-        database: a connected ``db.Node_collection`` (the caller owns it, the
-            same way Module B's ``run_noise_filter`` takes a session it did not
-            open).
-        config: Module C settings; defaults to ``load_config()``.
-        embed_fn: text -> embedding. Defaults to the prompt handler's embedder,
-            which calls the paid embedding API; injectable so a caller can
-            supply a local or fake embedder.
+    Returns ``(retriever, cre_embeddings)``: the retriever behind the shared
+    ``retrieve()`` seam, and the hub embeddings it was built from (the in_memory
+    pool source, and the id set ``build_components`` records as known CREs).
 
-    Raises whatever the pgvector guard raises when that backend is configured
-    but unavailable — deliberately, rather than falling back to a different
-    retrieval path and producing silently different rankings.
+    Split out of ``build_components`` so a caller that only needs the shortlist
+    (the ``/rest/v1/link/suggest`` endpoint, which runs in the web process and
+    must not load a torch cross-encoder) can share C.1 exactly, instead of
+    re-implementing retrieval and drifting from the pipeline.
     """
     from application.defs import cre_defs as defs
     from application.utils.librarian.candidate_retriever import (
@@ -89,10 +85,6 @@ def build_components(
         RetrieverBackend,
         build_retriever,
     )
-    from application.utils.librarian.cross_encoder import (
-        CrossEncoderReranker,
-        build_cross_encoder_score_fn,
-    )
 
     config = config or load_config()
     backend = RetrieverBackend(config.retriever_backend)
@@ -130,6 +122,40 @@ def build_components(
             else None
         ),
     )
+    return retriever, cre_embeddings
+
+
+def build_components(
+    database: Any,
+    *,
+    config: Optional[LibrarianConfig] = None,
+    embed_fn: Optional[Callable[[str], Sequence[float]]] = None,
+) -> LibrarianComponents:
+    """Construct C.1 + C.2 + C.3 against a connected OpenCRE database.
+
+    Args:
+        database: a connected ``db.Node_collection`` (the caller owns it, the
+            same way Module B's ``run_noise_filter`` takes a session it did not
+            open).
+        config: Module C settings; defaults to ``load_config()``.
+        embed_fn: text -> embedding. Defaults to the prompt handler's embedder,
+            which calls the paid embedding API; injectable so a caller can
+            supply a local or fake embedder.
+
+    Raises whatever the pgvector guard raises when that backend is configured
+    but unavailable — deliberately, rather than falling back to a different
+    retrieval path and producing silently different rankings.
+    """
+    from application.defs import cre_defs as defs
+    from application.utils.librarian.cross_encoder import (
+        CrossEncoderReranker,
+        build_cross_encoder_score_fn,
+    )
+
+    config = config or load_config()
+    retriever, cre_embeddings = build_retriever_for(
+        database, config=config, embed_fn=embed_fn
+    )
 
     # C.2 scores each (section, candidate) pair against the CRE's
     # embeddings_content — the same text the hub vectors were built from.
diff --git a/application/utils/librarian/link_suggest.py b/application/utils/librarian/link_suggest.py
new file mode 100644
index 000000000..3bffbcd5b
--- /dev/null
+++ b/application/utils/librarian/link_suggest.py
@@ -0,0 +1,190 @@
+"""``POST /rest/v1/link/suggest``: the Librarian's shortlist as a public API.
+
+Given a CRE ``Standard`` document (the same JSON shape the importers and the
+REST API already speak), return the CREs most likely to be its link targets,
+ranked. This is C.1 of the Librarian exposed directly, so a standard author or
+maintainer can ask "where does this section belong?" without running the OIE
+pipeline, and without a second similarity stack growing beside the one that is
+evaluated and gated in CI (#275, and the maintainer decision on #965: do not
+build a parallel mapping path).
+
+What this deliberately is not: the Librarian's *decision*. The pipeline's C.2
+rerank and C.3 calibration run in a worker with a torch cross-encoder loaded;
+the web process must not pay that (the same constraint the chat retrieval in
+#1161 respects). The score returned here is the retriever's cosine similarity,
+named as such, ranked, and suitable for a human to choose from. Automated
+mapping with a confidence threshold is ``mapping_job`` territory (#965), which
+calls the same retriever and then the rest of the chain.
+"""
+
+from cre_logging import get_logger
+
+logger = get_logger(__name__)
+
+from typing import Any, Callable, Dict, List, Mapping, Optional
+
+from application.defs import cre_defs as defs
+
+#: Hard ceiling on one request's shortlist. The retriever's own top_k is the
+#: pipeline setting (20 by default); a caller may ask for fewer, never more.
+MAX_TOP_K = 20
+DEFAULT_TOP_K = 5
+
+#: Fields of a Standard that carry text worth embedding, in the order they are
+#: joined. ``name`` is excluded on purpose: "ASVS" adds nothing to a section's
+#: meaning and would pull every ASVS section toward the same neighbourhood.
+_TEXT_FIELDS = ("section", "subsection", "description")
+
+
+class InvalidStandardError(ValueError):
+    """The request body is not a Standard document this endpoint can embed."""
+
+
+def standard_query_text(doc: Mapping[str, Any]) -> str:
+    """Build the text C.1 embeds from a Standard document, or raise.
+
+    Mirrors how the pipeline treats a section: the section heading, the
+    subsection and the description are what carry the requirement; ids and
+    hyperlinks are provenance. At least one text field must be non-empty.
+    """
+    if not isinstance(doc, Mapping):
+        raise InvalidStandardError("body must be a JSON object")
+    doctype = doc.get("doctype")
+    if doctype != defs.Credoctypes.Standard.value:
+        raise InvalidStandardError(
+            f"doctype must be {defs.Credoctypes.Standard.value!r}, got {doctype!r}"
+        )
+    name = doc.get("name")
+    if not isinstance(name, str) or not name.strip():
+        raise InvalidStandardError("name is required")
+    parts: List[str] = []
+    for field in _TEXT_FIELDS:
+        value = doc.get(field)
+        if value is None:
+            continue
+        if not isinstance(value, str):
+            raise InvalidStandardError(f"{field} must be a string")
+        if value.strip():
+            parts.append(value.strip())
+    if not parts:
+        raise InvalidStandardError(
+            "at least one of section, subsection or description must be non-empty"
+        )
+    return ". ".join(parts)
+
+
+def clamp_top_k(raw: Any) -> int:
+    """Parse a caller's top_k; absent means the default, bad input raises."""
+    if raw is None or raw == "":
+        return DEFAULT_TOP_K
+    # Whole numbers only. int() would quietly truncate 2.5 to 2 and read True
+    # as 1; a limit the caller mistyped should be refused, not reinterpreted.
+    if isinstance(raw, bool):
+        raise InvalidStandardError("top_k must be an integer")
+    if isinstance(raw, int):
+        value = raw
+    elif isinstance(raw, float) and raw.is_integer():
+        value = int(raw)
+    elif isinstance(raw, str) and raw.strip().lstrip("-").isdigit():
+        value = int(raw.strip())
+    else:
+        raise InvalidStandardError("top_k must be an integer")
+    if value < 1:
+        raise InvalidStandardError("top_k must be at least 1")
+    return min(value, MAX_TOP_K)
+
+
+def _default_retriever(database: Any) -> Any:
+    """C.1 exactly as the pipeline builds it, minus C.2 and C.3.
+
+    A seam: tests and callers with their own retriever pass one in and this is
+    never called. Lazy import keeps the web process free of ML imports until a
+    request actually needs them.
+    """
+    from application.utils.librarian.factory import build_retriever_for
+
+    retriever, _ = build_retriever_for(database)
+    return retriever
+
+
+def _resolve_cre(database: Any, candidate_id: str) -> Optional[defs.CRE]:
+    """Find the graph's CRE for a retriever candidate id, whichever space it is in.
+
+    The embeddings table keys CRE vectors by the CRE row's *internal* id, so
+    both retriever backends hand back UUIDs, while everything a human or an
+    importer sees is the external id ("287-251"). The eval harness translates
+    the pool for exactly this reason; here the translation happens per result,
+    so the endpoint is correct against a pool keyed either way.
+    """
+    for kwargs in ({"internal_id": candidate_id}, {"external_id": candidate_id}):
+        try:
+            matches = database.get_CREs(**kwargs) or []
+        except Exception:  # a malformed id must not take the whole shortlist down
+            matches = []
+        if matches:
+            found: defs.CRE = matches[0]
+            return found
+    return None
+
+
+def suggest_links(
+    database: Any,
+    doc: Mapping[str, Any],
+    *,
+    top_k: int = DEFAULT_TOP_K,
+    retriever: Optional[Any] = None,
+    retriever_factory: Optional[Callable[[Any], Any]] = None,
+) -> List[Dict[str, Any]]:
+    """Rank the CREs most likely to link to ``doc``.
+
+    Returns the response shape #275 asked for, one entry per candidate, best
+    first: ``{"doctype": "CRE", "id", "name", "tags", "score"}``. ``score`` is
+    the retriever's cosine similarity in [0, 1]. ``name`` and ``tags`` come from
+    the graph when the CRE is present; a candidate the retriever knows but the
+    graph does not (a stale embedding) is still returned, with the retriever's
+    name and no tags, rather than silently dropped.
+    """
+    text = standard_query_text(doc)
+    if top_k < 1 or top_k > MAX_TOP_K:
+        raise InvalidStandardError(f"top_k must be between 1 and {MAX_TOP_K}")
+    if retriever is None:
+        # Resolved at call time, not bound as a default, so the module-level
+        # builder stays the single seam tests and callers can replace.
+        factory = retriever_factory or _default_retriever
+        retriever = factory(database)
+    audit = retriever.retrieve(text)
+
+    results: List[Dict[str, Any]] = []
+    for cand in audit.candidates[:top_k]:
+        cre = _resolve_cre(database, cand.cre_id)
+        name = (
+            (cre.name if cre is not None and cre.name else None) or cand.cre_name or ""
+        )
+        tags: List[str] = (
+            sorted(t for t in (cre.tags or []) if t) if cre is not None else []
+        )
+        results.append(
+            {
+                "doctype": defs.Credoctypes.CRE.value,
+                # Always the public id ("287-251"), whatever the pool was keyed by.
+                "id": cre.id if cre is not None and cre.id else cand.cre_id,
+                "name": name,
+                "tags": tags,
+                "score": (
+                    round(cand.score_vector, 4)
+                    if cand.score_vector is not None
+                    else None
+                ),
+            }
+        )
+    return results
+
+
+__all__ = [
+    "DEFAULT_TOP_K",
+    "MAX_TOP_K",
+    "InvalidStandardError",
+    "clamp_top_k",
+    "standard_query_text",
+    "suggest_links",
+]
diff --git a/application/web/web_main.py b/application/web/web_main.py
index 094f818e8..e94494823 100644
--- a/application/web/web_main.py
+++ b/application/web/web_main.py
@@ -1285,6 +1285,73 @@ def chat_cre() -> Any:
     return jsonify(response)
 
 
+@app.route("/rest/v1/link/suggest", methods=["POST"])
+@login_required
+def link_suggest() -> Any:
+    """Rank the CREs a Standard section most likely links to (#275).
+
+    Body: a CRE ``Standard`` JSON document. Optional ``top_k`` (query string or
+    body, 1 to 20, default 5). Response: a JSON array of CREs, best first, each
+    with ``doctype``, ``id``, ``name``, ``tags`` and the retriever's cosine
+    ``score``. Same auth and same unavailable/rate-limit handling as
+    ``/rest/v1/completion``: both call the paid embedding API.
+    """
+    payload = request.get_json(force=True, silent=True)
+    if not isinstance(payload, dict):
+        abort(400, description="body must be a JSON object (a CRE Standard document)")
+    if posthog:
+        posthog.capture("link_suggest", "")
+
+    database = db.Node_collection()
+    # Lazy import to avoid loading heavy prompt/ML dependencies at web boot.
+    try:
+        from application.prompt_client import llm_error_utils
+        from application.utils.librarian import link_suggest as suggest
+    except ImportError:
+        abort(
+            503,
+            "Link suggestion is unavailable on this deployment "
+            "(ML/LLM dependencies not installed).",
+        )
+
+    try:
+        top_k = suggest.clamp_top_k(request.args.get("top_k", payload.get("top_k")))
+        results = suggest.suggest_links(database, payload, top_k=top_k)
+    except suggest.InvalidStandardError as exc:
+        abort(400, description=str(exc))
+    except ImportError:
+        abort(
+            503,
+            "Link suggestion is unavailable on this deployment "
+            "(ML/LLM dependencies not installed).",
+        )
+    except RuntimeError as exc:
+        # PromptHandler raises RuntimeError when litellm is missing.
+        if "litellm" not in str(exc).lower():
+            raise
+        abort(
+            503,
+            "Link suggestion is unavailable on this deployment "
+            "(ML/LLM dependencies not installed).",
+        )
+    except Exception as e:
+        if llm_error_utils.is_rate_limit_error(e):
+            return (
+                jsonify(
+                    {
+                        "error": (
+                            "The AI service is temporarily rate-limited or out of quota. "
+                            "Please try again in a minute."
+                        )
+                    }
+                ),
+                503,
+            )
+        status_code = _llm_error_status_code(e) or 500
+        return jsonify({"error": f"AI Service Error: {str(e)}"}), status_code
+    return jsonify(results)
+
+
 class CREFlow:
     """ "This class handles authentication with google's oauth"""
 

From 569a91287f5d469404d80a7a564144e0d101b7ea Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sat, 24 Oct 2026 01:58:17 +0530
Subject: [PATCH 2/8] link/suggest: generic error body, strict top_k parsing,
 one lookup per candidate

Addresses the review on #1171.

The catch-all returned str(e) to the client. suggest_links runs database
lookups, config loading and pgvector guards, whose exceptions can carry SQL
text, paths and config values; CodeQL flagged it as information exposure.
The full exception now goes to the server log and the client gets the status
code with a generic message. The completion route has the same pattern; it is
not copied here.

top_k string parsing used an isdigit-style pre-check that accepted inputs
int() rejects ("--3", a superscript digit), so those surfaced as a 500 from
the catch-all instead of a 400. int() is now the parser and its failure is
converted; two tests cover the two inputs at both the service and the route.

_resolve_cre did an internal-id lookup and then an external-id lookup per
candidate, and get_CREs warns on each miss, so one stale embedding warned
twice on every request. The lookup is now chosen once by the id's shape
(public ids are "ddd-ddd", the pattern cre_defs.CRE enforces).
---
 .../tests/librarian/link_suggest_test.py      |  9 ++++-
 application/utils/librarian/link_suggest.py   | 38 +++++++++++++------
 application/web/web_main.py                   |  9 ++++-
 3 files changed, 43 insertions(+), 13 deletions(-)

diff --git a/application/tests/librarian/link_suggest_test.py b/application/tests/librarian/link_suggest_test.py
index e153d368e..834ae40db 100644
--- a/application/tests/librarian/link_suggest_test.py
+++ b/application/tests/librarian/link_suggest_test.py
@@ -109,7 +109,7 @@ def test_parses_and_caps(self) -> None:
         self.assertEqual(clamp_top_k(99), MAX_TOP_K)
 
     def test_rejects_non_positive_and_non_integer(self) -> None:
-        for bad in ("0", "-3", "x", 2.5):
+        for bad in ("0", "-3", "x", 2.5, "--3", "\u00b2", True):
             with self.assertRaises(InvalidStandardError):
                 clamp_top_k(bad)
 
@@ -303,6 +303,11 @@ def test_missing_litellm_is_503(self) -> None:
                 resp = c.post("/rest/v1/link/suggest", json=_standard())
         self.assertEqual(resp.status_code, 503)
 
+    def test_malformed_top_k_strings_are_400_not_500(self) -> None:
+        for bad in ("--3", "\u00b2"):
+            resp = self._post(_standard(), query=f"?top_k={bad}")
+            self.assertEqual(resp.status_code, 400, bad)
+
     def test_provider_failure_is_reported_not_swallowed(self) -> None:
         class _Boom(Exception):
             status_code = 502
@@ -314,6 +319,8 @@ class _Boom(Exception):
                 resp = c.post("/rest/v1/link/suggest", json=_standard())
         self.assertEqual(resp.status_code, 502)
         self.assertIn("error", resp.get_json())
+        # the client gets a status and a generic message, never the raw text
+        self.assertNotIn("upstream down", resp.get_data(as_text=True))
 
 
 if __name__ == "__main__":
diff --git a/application/utils/librarian/link_suggest.py b/application/utils/librarian/link_suggest.py
index 3bffbcd5b..0ea834b0f 100644
--- a/application/utils/librarian/link_suggest.py
+++ b/application/utils/librarian/link_suggest.py
@@ -21,10 +21,14 @@
 
 logger = get_logger(__name__)
 
+import re
 from typing import Any, Callable, Dict, List, Mapping, Optional
 
 from application.defs import cre_defs as defs
 
+#: The public CRE id shape ("287-251"); everything else is treated as a row id.
+_PUBLIC_ID = re.compile(r"\d{3}-\d{3}")
+
 #: Hard ceiling on one request's shortlist. The retriever's own top_k is the
 #: pipeline setting (20 by default); a caller may ask for fewer, never more.
 MAX_TOP_K = 20
@@ -85,8 +89,13 @@ def clamp_top_k(raw: Any) -> int:
         value = raw
     elif isinstance(raw, float) and raw.is_integer():
         value = int(raw)
-    elif isinstance(raw, str) and raw.strip().lstrip("-").isdigit():
-        value = int(raw.strip())
+    elif isinstance(raw, str):
+        # int() is the parser; "--3" and "\u00b2" pass isdigit-style checks and
+        # then blow up inside int(), so its failure is the signal, not a pre-check.
+        try:
+            value = int(raw.strip())
+        except ValueError:
+            raise InvalidStandardError("top_k must be an integer") from None
     else:
         raise InvalidStandardError("top_k must be an integer")
     if value < 1:
@@ -116,15 +125,22 @@ def _resolve_cre(database: Any, candidate_id: str) -> Optional[defs.CRE]:
     the pool for exactly this reason; here the translation happens per result,
     so the endpoint is correct against a pool keyed either way.
     """
-    for kwargs in ({"internal_id": candidate_id}, {"external_id": candidate_id}):
-        try:
-            matches = database.get_CREs(**kwargs) or []
-        except Exception:  # a malformed id must not take the whole shortlist down
-            matches = []
-        if matches:
-            found: defs.CRE = matches[0]
-            return found
-    return None
+    # One lookup, chosen by the id's shape: public ids are "ddd-ddd" (the
+    # pattern cre_defs.CRE enforces), anything else is a row id. A single miss
+    # then logs once, instead of twice per stale candidate on every request.
+    kwargs = (
+        {"external_id": candidate_id}
+        if _PUBLIC_ID.fullmatch(candidate_id or "")
+        else {"internal_id": candidate_id}
+    )
+    try:
+        matches = database.get_CREs(**kwargs) or []
+    except Exception:  # a malformed id must not take the whole shortlist down
+        return None
+    if not matches:
+        return None
+    found: defs.CRE = matches[0]
+    return found
 
 
 def suggest_links(
diff --git a/application/web/web_main.py b/application/web/web_main.py
index e94494823..97d5b8625 100644
--- a/application/web/web_main.py
+++ b/application/web/web_main.py
@@ -1347,8 +1347,15 @@ def link_suggest() -> Any:
                 ),
                 503,
             )
+        # Never echo the exception: SQLAlchemy and provider errors can carry SQL
+        # text, paths and config values. Full detail goes to the server log; the
+        # client gets the status code and a generic message.
         status_code = _llm_error_status_code(e) or 500
-        return jsonify({"error": f"AI Service Error: {str(e)}"}), status_code
+        logger.error("link_suggest failed with status %s", status_code, exc_info=True)
+        return (
+            jsonify({"error": "Link suggestion failed upstream; see server logs."}),
+            status_code,
+        )
     return jsonify(results)
 
 

From 431091b6e57b39ad0d0e115fa2035048fcdfa91a Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sat, 24 Oct 2026 01:59:04 +0530
Subject: [PATCH 3/8] link/suggest: top_k strings are an optional sign and
 ASCII digits only

int() also accepts "2_0"; a top_k is not a place for underscore literals.
The string form is now matched before parsing, so "--3", a superscript
digit and "2_0" are all a 400.
---
 application/tests/librarian/link_suggest_test.py |  4 ++--
 application/utils/librarian/link_suggest.py      | 16 +++++++++-------
 2 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/application/tests/librarian/link_suggest_test.py b/application/tests/librarian/link_suggest_test.py
index 834ae40db..8d891175c 100644
--- a/application/tests/librarian/link_suggest_test.py
+++ b/application/tests/librarian/link_suggest_test.py
@@ -109,7 +109,7 @@ def test_parses_and_caps(self) -> None:
         self.assertEqual(clamp_top_k(99), MAX_TOP_K)
 
     def test_rejects_non_positive_and_non_integer(self) -> None:
-        for bad in ("0", "-3", "x", 2.5, "--3", "\u00b2", True):
+        for bad in ("0", "-3", "x", 2.5, "--3", "\u00b2", "2_0", " 7 x", True):
             with self.assertRaises(InvalidStandardError):
                 clamp_top_k(bad)
 
@@ -304,7 +304,7 @@ def test_missing_litellm_is_503(self) -> None:
         self.assertEqual(resp.status_code, 503)
 
     def test_malformed_top_k_strings_are_400_not_500(self) -> None:
-        for bad in ("--3", "\u00b2"):
+        for bad in ("--3", "\u00b2", "2_0"):
             resp = self._post(_standard(), query=f"?top_k={bad}")
             self.assertEqual(resp.status_code, 400, bad)
 
diff --git a/application/utils/librarian/link_suggest.py b/application/utils/librarian/link_suggest.py
index 0ea834b0f..7cfb2a7da 100644
--- a/application/utils/librarian/link_suggest.py
+++ b/application/utils/librarian/link_suggest.py
@@ -27,7 +27,9 @@
 from application.defs import cre_defs as defs
 
 #: The public CRE id shape ("287-251"); everything else is treated as a row id.
-_PUBLIC_ID = re.compile(r"\d{3}-\d{3}")
+_PUBLIC_ID = re.compile(r"[0-9]{3}-[0-9]{3}")
+#: What a top_k string may be: an optional sign and ASCII digits, nothing else.
+_ASCII_INT = re.compile(r"[+-]?[0-9]+")
 
 #: Hard ceiling on one request's shortlist. The retriever's own top_k is the
 #: pipeline setting (20 by default); a caller may ask for fewer, never more.
@@ -90,12 +92,12 @@ def clamp_top_k(raw: Any) -> int:
     elif isinstance(raw, float) and raw.is_integer():
         value = int(raw)
     elif isinstance(raw, str):
-        # int() is the parser; "--3" and "\u00b2" pass isdigit-style checks and
-        # then blow up inside int(), so its failure is the signal, not a pre-check.
-        try:
-            value = int(raw.strip())
-        except ValueError:
-            raise InvalidStandardError("top_k must be an integer") from None
+        # Strict ASCII integer: isdigit() admits "\u00b2" and lstrip("-") admits
+        # "--3", while int() itself admits "2_0". Only an optional sign and
+        # ASCII digits are a top_k; everything else is a 400.
+        if not _ASCII_INT.fullmatch(raw.strip()):
+            raise InvalidStandardError("top_k must be an integer")
+        value = int(raw.strip())
     else:
         raise InvalidStandardError("top_k must be an integer")
     if value < 1:

From c5296b85d24e5ea4c94f766881305b569636e40e Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sun, 25 Oct 2026 20:03:26 +0530
Subject: [PATCH 4/8] link/suggest: one failure path for every exception, log
 failed CRE lookups

Review fixes on the suggest endpoint.

- The route no longer has a dedicated RuntimeError clause. It re-raised any
  RuntimeError whose message did not mention litellm, and a sibling except
  cannot catch a re-raise from the same try, so a retriever or database
  RuntimeError skipped the rate-limit check, the status mapping, the generic
  client body and the server-side log, and the client got Flask's default
  500 page. The litellm check now sits at the top of the generic handler;
  everything else follows the same path as any other failure.
- _resolve_cre logs a CRE lookup that raises, with the candidate id and the
  traceback, before returning the candidate unresolved. A lookup that raises
  is not a stale embedding, and the log now says which.
- Two tests: a non-litellm RuntimeError yields the generic 500 body with no
  exception text; a raising lookup is logged with the candidate id.
---
 .../tests/librarian/link_suggest_test.py      | 29 +++++++++++++++++++
 application/utils/librarian/link_suggest.py   | 11 ++++++-
 application/web/web_main.py                   | 18 ++++++------
 3 files changed, 48 insertions(+), 10 deletions(-)

diff --git a/application/tests/librarian/link_suggest_test.py b/application/tests/librarian/link_suggest_test.py
index 8d891175c..681d7eb41 100644
--- a/application/tests/librarian/link_suggest_test.py
+++ b/application/tests/librarian/link_suggest_test.py
@@ -134,6 +134,16 @@ def tearDown(self) -> None:
         sqla.drop_all()
         self.ctx.pop()
 
+    def test_a_failing_cre_lookup_is_logged_with_the_candidate(self) -> None:
+        """A lookup that raises is not a miss: it is logged, named, and the
+        candidate still comes back unresolved so the shortlist survives."""
+        broken = mock.Mock()
+        broken.get_CREs.side_effect = RuntimeError("database gone")
+        with self.assertLogs(link_suggest.logger, level="WARNING") as logs:
+            self.assertIsNone(link_suggest._resolve_cre(broken, "287-251"))
+        self.assertIn("287-251", logs.output[0])
+        self.assertIn("database gone", logs.output[0])
+
     def test_internal_uuid_candidates_resolve_to_the_public_cre_id(self) -> None:
         """The embeddings table keys vectors by the CRE row's internal UUID, so
         that is what the retriever returns. A caller must only ever see
@@ -303,6 +313,25 @@ def test_missing_litellm_is_503(self) -> None:
                 resp = c.post("/rest/v1/link/suggest", json=_standard())
         self.assertEqual(resp.status_code, 503)
 
+    def test_other_runtime_errors_get_the_generic_body_not_flasks_500_page(
+        self,
+    ) -> None:
+        """Only a missing litellm is a 503; any other RuntimeError must go
+        through the same generic handling as every other failure."""
+        with mock.patch.object(
+            link_suggest,
+            "_default_retriever",
+            side_effect=RuntimeError("index file /srv/cre/secret.idx is corrupt"),
+        ):
+            with self.app.test_client() as c:
+                resp = c.post("/rest/v1/link/suggest", json=_standard())
+        self.assertEqual(resp.status_code, 500)
+        self.assertEqual(
+            resp.get_json(),
+            {"error": "Link suggestion failed upstream; see server logs."},
+        )
+        self.assertNotIn("secret.idx", resp.get_data(as_text=True))
+
     def test_malformed_top_k_strings_are_400_not_500(self) -> None:
         for bad in ("--3", "\u00b2", "2_0"):
             resp = self._post(_standard(), query=f"?top_k={bad}")
diff --git a/application/utils/librarian/link_suggest.py b/application/utils/librarian/link_suggest.py
index 7cfb2a7da..d72545dba 100644
--- a/application/utils/librarian/link_suggest.py
+++ b/application/utils/librarian/link_suggest.py
@@ -137,7 +137,16 @@ def _resolve_cre(database: Any, candidate_id: str) -> Optional[defs.CRE]:
     )
     try:
         matches = database.get_CREs(**kwargs) or []
-    except Exception:  # a malformed id must not take the whole shortlist down
+    except Exception:
+        # A malformed id must not take the whole shortlist down, but a lookup
+        # that raises is not the same as a lookup that misses: say which
+        # candidate failed so a broken database path does not read as "stale".
+        logger.warning(
+            "link_suggest: CRE lookup failed for candidate %r; "
+            "returning it unresolved",
+            candidate_id,
+            exc_info=True,
+        )
         return None
     if not matches:
         return None
diff --git a/application/web/web_main.py b/application/web/web_main.py
index 97d5b8625..4d0da4f97 100644
--- a/application/web/web_main.py
+++ b/application/web/web_main.py
@@ -1325,16 +1325,16 @@ def link_suggest() -> Any:
             "Link suggestion is unavailable on this deployment "
             "(ML/LLM dependencies not installed).",
         )
-    except RuntimeError as exc:
-        # PromptHandler raises RuntimeError when litellm is missing.
-        if "litellm" not in str(exc).lower():
-            raise
-        abort(
-            503,
-            "Link suggestion is unavailable on this deployment "
-            "(ML/LLM dependencies not installed).",
-        )
     except Exception as e:
+        # PromptHandler raises RuntimeError when litellm is missing. Any other
+        # RuntimeError (retriever, database) falls through to the generic
+        # handling below; a dedicated except clause would re-raise past it.
+        if isinstance(e, RuntimeError) and "litellm" in str(e).lower():
+            abort(
+                503,
+                "Link suggestion is unavailable on this deployment "
+                "(ML/LLM dependencies not installed).",
+            )
         if llm_error_utils.is_rate_limit_error(e):
             return (
                 jsonify(

From 4cad5cf756d092f4a5b2aa065022e01202ad52ca Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Mon, 26 Oct 2026 14:46:21 +0530
Subject: [PATCH 5/8] link/suggest: a failed CRE lookup is an error, not an
 unresolved candidate

_resolve_cre no longer catches exceptions from get_CREs. The lookup is
chosen by id shape and is a plain equality filter either way, so a
malformed id simply misses and returns None; it never raises. An exception
therefore means the database failed, and swallowing it returned a 200 whose
candidates carried raw internal UUIDs. It now reaches the route's single
error path: logged server-side with the traceback, generic body to the
client.

Tests: a raising lookup propagates; malformed and empty ids are misses; a
database failure during resolution is a 500 with the generic body and no
exception text.
---
 .../tests/librarian/link_suggest_test.py      | 32 +++++++++++++++----
 application/utils/librarian/link_suggest.py   | 18 +++--------
 2 files changed, 30 insertions(+), 20 deletions(-)

diff --git a/application/tests/librarian/link_suggest_test.py b/application/tests/librarian/link_suggest_test.py
index 681d7eb41..6a2c1fc15 100644
--- a/application/tests/librarian/link_suggest_test.py
+++ b/application/tests/librarian/link_suggest_test.py
@@ -134,15 +134,19 @@ def tearDown(self) -> None:
         sqla.drop_all()
         self.ctx.pop()
 
-    def test_a_failing_cre_lookup_is_logged_with_the_candidate(self) -> None:
-        """A lookup that raises is not a miss: it is logged, named, and the
-        candidate still comes back unresolved so the shortlist survives."""
+    def test_a_failing_cre_lookup_propagates_instead_of_reading_as_a_miss(
+        self,
+    ) -> None:
+        """A lookup that raises is not a stale embedding: it must not come
+        back as an unresolved candidate carrying a raw UUID."""
         broken = mock.Mock()
         broken.get_CREs.side_effect = RuntimeError("database gone")
-        with self.assertLogs(link_suggest.logger, level="WARNING") as logs:
-            self.assertIsNone(link_suggest._resolve_cre(broken, "287-251"))
-        self.assertIn("287-251", logs.output[0])
-        self.assertIn("database gone", logs.output[0])
+        with self.assertRaises(RuntimeError):
+            link_suggest._resolve_cre(broken, "287-251")
+
+    def test_a_malformed_candidate_id_is_a_miss_not_an_error(self) -> None:
+        self.assertIsNone(link_suggest._resolve_cre(self.collection, "not-an-id"))
+        self.assertIsNone(link_suggest._resolve_cre(self.collection, ""))
 
     def test_internal_uuid_candidates_resolve_to_the_public_cre_id(self) -> None:
         """The embeddings table keys vectors by the CRE row's internal UUID, so
@@ -332,6 +336,20 @@ def test_other_runtime_errors_get_the_generic_body_not_flasks_500_page(
         )
         self.assertNotIn("secret.idx", resp.get_data(as_text=True))
 
+    def test_a_database_failure_during_resolution_is_a_500_not_a_200(self) -> None:
+        with mock.patch.object(
+            db.Node_collection,
+            "get_CREs",
+            side_effect=RuntimeError("connection reset by /var/run/pg.sock"),
+        ):
+            resp = self._post(_standard())
+        self.assertEqual(resp.status_code, 500)
+        self.assertEqual(
+            resp.get_json(),
+            {"error": "Link suggestion failed upstream; see server logs."},
+        )
+        self.assertNotIn("pg.sock", resp.get_data(as_text=True))
+
     def test_malformed_top_k_strings_are_400_not_500(self) -> None:
         for bad in ("--3", "\u00b2", "2_0"):
             resp = self._post(_standard(), query=f"?top_k={bad}")
diff --git a/application/utils/librarian/link_suggest.py b/application/utils/librarian/link_suggest.py
index d72545dba..ab9096cc3 100644
--- a/application/utils/librarian/link_suggest.py
+++ b/application/utils/librarian/link_suggest.py
@@ -135,19 +135,11 @@ def _resolve_cre(database: Any, candidate_id: str) -> Optional[defs.CRE]:
         if _PUBLIC_ID.fullmatch(candidate_id or "")
         else {"internal_id": candidate_id}
     )
-    try:
-        matches = database.get_CREs(**kwargs) or []
-    except Exception:
-        # A malformed id must not take the whole shortlist down, but a lookup
-        # that raises is not the same as a lookup that misses: say which
-        # candidate failed so a broken database path does not read as "stale".
-        logger.warning(
-            "link_suggest: CRE lookup failed for candidate %r; "
-            "returning it unresolved",
-            candidate_id,
-            exc_info=True,
-        )
-        return None
+    # No try/except: an id of any shape is a plain equality filter, so a
+    # malformed id misses (None below) and never raises. A lookup that raises
+    # means the database failed, and that must reach the route's error
+    # handling rather than come back as a 200 full of unresolved UUIDs.
+    matches = database.get_CREs(**kwargs) or []
     if not matches:
         return None
     found: defs.CRE = matches[0]

From afbd618095a5d4a36af713b92ef487def0661ca6 Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sat, 24 Oct 2026 00:54:38 +0530
Subject: [PATCH 6/8] MyOpenCRE mapping job: reuse the Librarian for user
 standards

Fixes #965.

RFC user-auth-myopencre TODO 3, built on the maintainer decision in #965:
there is no second mapping stack. run_mapping takes a user standard's PENDING
sections and runs them through the same C.1 retrieve, C.2 rerank, C.3
calibrate and C.4 decide the OIE pipeline runs, then records one outcome per
section: AUTO_MAPPED at or above the Librarian's link threshold, NEEDS_REVIEW
below it with the candidate kept as the reviewer's hint, UNMAPPABLE when
nothing was retrieved. The status rule is decision_engine.decide itself, so
MyOpenCRE and the pipeline cannot disagree about what a confidence means.

Nothing of TODO 2 existed in code, only in the RFC, so this also lays the
schema the job reads and writes: user_standards and user_standard_sections,
migration c4a1e7b2d9f0 off f3a91c07d24b, plus the Node_collection methods the
RFC names. cre_db_id is a soft reference to the CRE row's internal id,
validated at write time, never a foreign key. USER_CONFIRMED and USER_REJECTED
rows are ground truth: the job never reads them, and set_section_mapping
refuses to move them even if asked, so no caller can break that by accident.
A re-uploaded section whose text changed returns to PENDING unless the user
has decided it.

Retrievers hand back the embeddings pool's keys, which are internal CRE ids;
candidates are resolved through the graph (shared with /rest/v1/link/suggest)
and translated with the new get_cre_internal_id, so a candidate the graph
cannot resolve goes to review with no dangling reference. Per-section failures
are contained, counted, and left PENDING for the next run; a re-run processes
only what the previous run did not finish. run_mapping_job is the rq entry
point by dotted path, as run_gap_pair_job is; the cross-encoder loads in the
worker, never in the web process.

12 fixture-driven tests, as the RFC asks: deterministic status counts, the
stored id is the validated internal id, user decisions survive a re-index
re-run, idempotent re-run, errored sections stay PENDING, threshold from the
Librarian config, ownership isolation, duplicate-name constraint. Migration
applied, downgraded and re-applied on a scratch SQLite through Alembic.
db.py strict mypy count unchanged (129 on main, 129 here).
---
 application/database/db.py                    | 256 ++++++++++++-
 application/tests/mapping_job_test.py         | 337 ++++++++++++++++++
 application/utils/mapping_job.py              | 192 ++++++++++
 .../c4a1e7b2d9f0_add_user_standards.py        |  75 ++++
 4 files changed, 859 insertions(+), 1 deletion(-)
 create mode 100644 application/tests/mapping_job_test.py
 create mode 100644 application/utils/mapping_job.py
 create mode 100644 migrations/versions/c4a1e7b2d9f0_add_user_standards.py

diff --git a/application/database/db.py b/application/database/db.py
index 22e7358be..8beec0cc9 100644
--- a/application/database/db.py
+++ b/application/database/db.py
@@ -15,7 +15,7 @@
 
 from collections import Counter, defaultdict
 from itertools import permutations
-from typing import Any, Dict, List, Optional, Tuple, cast
+from typing import Any, Dict, List, Optional, Sequence, Tuple, cast
 from neomodel.exceptions import (
     DoesNotExist,
     FeatureNotSupported,
@@ -57,6 +57,8 @@
 from .. import sqla  # type: ignore
 
 
+from enum import Enum
+
 BaseModel: DefaultMeta = sqla.Model
 
 
@@ -462,6 +464,80 @@ class HarvestInput(BaseModel):  # type: ignore
     )
 
 
+class SectionStatus(str, Enum):
+    """Lifecycle of one user-standard section (RFC user-auth-myopencre, TODO 2).
+
+    ``USER_CONFIRMED`` and ``USER_REJECTED`` are ground truth: the mapping job
+    never rewrites them (RFC TODO 3), only the user can.
+    """
+
+    PENDING = "PENDING"
+    AUTO_MAPPED = "AUTO_MAPPED"
+    NEEDS_REVIEW = "NEEDS_REVIEW"
+    UNMAPPABLE = "UNMAPPABLE"
+    USER_CONFIRMED = "USER_CONFIRMED"
+    USER_REJECTED = "USER_REJECTED"
+
+
+USER_DECIDED_STATUSES = frozenset(
+    {SectionStatus.USER_CONFIRMED.value, SectionStatus.USER_REJECTED.value}
+)
+
+
+class UserStandard(BaseModel):  # type: ignore
+    """A standard a MyOpenCRE user uploaded for mapping (RFC TODO 2).
+
+    Owned by exactly one user; two users uploading the same-named standard get
+    two rows with zero shared state. Deleting the user cascades.
+    """
+
+    __tablename__ = "user_standards"
+    id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
+    user_id = sqla.Column(
+        sqla.String,
+        sqla.ForeignKey("users.id", onupdate="CASCADE", ondelete="CASCADE"),
+        nullable=False,
+    )
+    name = sqla.Column(sqla.String, nullable=False)
+    created_at = sqla.Column(sqla.DateTime, nullable=False)
+    updated_at = sqla.Column(sqla.DateTime, nullable=False)
+    __table_args__ = (
+        sqla.UniqueConstraint("user_id", "name", name="uq_user_standards_user_name"),
+    )
+
+
+class UserStandardSection(BaseModel):  # type: ignore
+    """One section of a user standard and what the mapping job made of it.
+
+    ``cre_db_id`` is a soft reference to the CRE row's internal id, validated at
+    write time, never a foreign key: CRE nodes live in the graph layer and a
+    hard FK would couple the two (RFC TODO 2). ``confidence`` is the Librarian's
+    calibrated C.3 confidence for the stored candidate.
+    """
+
+    __tablename__ = "user_standard_sections"
+    id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
+    standard_id = sqla.Column(
+        sqla.String,
+        sqla.ForeignKey("user_standards.id", onupdate="CASCADE", ondelete="CASCADE"),
+        nullable=False,
+    )
+    section_id = sqla.Column(sqla.String, nullable=False)
+    section_text = sqla.Column(sqla.Text, nullable=False)
+    cre_db_id = sqla.Column(sqla.String, nullable=True)
+    confidence = sqla.Column(sqla.Float, nullable=True)
+    status = sqla.Column(sqla.String, nullable=False)
+    created_at = sqla.Column(sqla.DateTime, nullable=False)
+    updated_at = sqla.Column(sqla.DateTime, nullable=False)
+    __table_args__ = (
+        sqla.UniqueConstraint(
+            "standard_id", "section_id", name="uq_user_standard_sections_section"
+        ),
+        sqla.Index("ix_user_standard_sections_status", "standard_id", "status"),
+        sqla.Index("ix_user_standard_sections_cre", "standard_id", "cre_db_id"),
+    )
+
+
 class KnowledgeQueueItem(BaseModel):  # type: ignore
     """Module B's output queue: security-knowledge chunks for Module C.
 
@@ -2381,6 +2457,184 @@ def delete_gapanalysis_results_for(self, node_name):
         self.session.flush()
         return res
 
+    # ---- MyOpenCRE user standards (RFC user-auth-myopencre, TODO 2 + 3) ----
+
+    @staticmethod
+    def _utcnow_naive() -> Any:
+        from datetime import datetime, timezone
+
+        return datetime.now(timezone.utc).replace(tzinfo=None)
+
+    def create_user_standard(self, user_id: str, name: str) -> UserStandard:
+        """Create one user-owned standard; (user_id, name) is unique."""
+        now = self._utcnow_naive()
+        standard = UserStandard(
+            user_id=user_id, name=name, created_at=now, updated_at=now
+        )
+        self.session.add(standard)
+        self.session.commit()
+        return standard
+
+    def get_user_standards(self, user_id: str) -> List[UserStandard]:
+        """Only the requesting user's standards, never another user's."""
+        rows: List[UserStandard] = (
+            UserStandard.query.filter(UserStandard.user_id == user_id)
+            .order_by(UserStandard.created_at, UserStandard.id)
+            .all()
+        )
+        return rows
+
+    def upsert_user_standard_sections(
+        self, standard_id: str, sections: Sequence[Tuple[str, str]]
+    ) -> int:
+        """Insert or refresh sections as ``(section_id, section_text)`` pairs.
+
+        New rows start ``PENDING``. A re-uploaded row whose text changed goes
+        back to ``PENDING`` so the job re-maps it, unless the user has already
+        decided it: ``USER_CONFIRMED`` / ``USER_REJECTED`` keep their status and
+        mapping, because user decisions are ground truth. Returns rows touched.
+        """
+        now = self._utcnow_naive()
+        existing = {
+            row.section_id: row
+            for row in UserStandardSection.query.filter(
+                UserStandardSection.standard_id == standard_id
+            ).all()
+        }
+        touched = 0
+        for section_id, section_text in sections:
+            row = existing.get(section_id)
+            if row is None:
+                self.session.add(
+                    UserStandardSection(
+                        standard_id=standard_id,
+                        section_id=section_id,
+                        section_text=section_text,
+                        status=SectionStatus.PENDING.value,
+                        created_at=now,
+                        updated_at=now,
+                    )
+                )
+                touched += 1
+                continue
+            if row.section_text == section_text:
+                continue
+            row.section_text = section_text
+            row.updated_at = now
+            if row.status not in USER_DECIDED_STATUSES:
+                row.status = SectionStatus.PENDING.value
+                row.cre_db_id = None
+                row.confidence = None
+            touched += 1
+        self.session.commit()
+        return touched
+
+    def _owned_standard(self, standard_id: str, user_id: str) -> Optional[UserStandard]:
+        row: Optional[UserStandard] = UserStandard.query.filter(
+            UserStandard.id == standard_id, UserStandard.user_id == user_id
+        ).first()
+        return row
+
+    def get_user_standard_sections(
+        self, standard_id: str, user_id: str
+    ) -> List[UserStandardSection]:
+        """Sections of a standard the user owns; empty for anyone else."""
+        if self._owned_standard(standard_id, user_id) is None:
+            return []
+        rows: List[UserStandardSection] = (
+            UserStandardSection.query.filter(
+                UserStandardSection.standard_id == standard_id
+            )
+            .order_by(UserStandardSection.section_id)
+            .all()
+        )
+        return rows
+
+    def get_user_standard_sections_by_status(
+        self, standard_id: str, statuses: Sequence[str]
+    ) -> List[UserStandardSection]:
+        """Job-side read, by status, in a stable order. No user scoping: the
+        mapping job runs as the system on a standard id it was handed."""
+        rows: List[UserStandardSection] = (
+            UserStandardSection.query.filter(
+                UserStandardSection.standard_id == standard_id,
+                UserStandardSection.status.in_(list(statuses)),
+            )
+            .order_by(UserStandardSection.section_id, UserStandardSection.id)
+            .all()
+        )
+        return rows
+
+    def set_section_mapping(
+        self,
+        section_id: str,
+        *,
+        cre_db_id: Optional[str],
+        confidence: Optional[float],
+        status: str,
+    ) -> None:
+        """Job-side write of one section's outcome. Never moves a user-decided
+        row, whatever it is asked to write: that invariant lives here so no
+        caller can break it by accident."""
+        row = UserStandardSection.query.filter(
+            UserStandardSection.id == section_id
+        ).first()
+        if row is None or row.status in USER_DECIDED_STATUSES:
+            return
+        row.cre_db_id = cre_db_id
+        row.confidence = confidence
+        row.status = status
+        row.updated_at = self._utcnow_naive()
+        self.session.commit()
+
+    def _owned_section(
+        self, section_id: str, user_id: str
+    ) -> Optional[UserStandardSection]:
+        row: Optional[UserStandardSection] = (
+            UserStandardSection.query.join(
+                UserStandard, UserStandard.id == UserStandardSection.standard_id
+            )
+            .filter(
+                UserStandardSection.id == section_id, UserStandard.user_id == user_id
+            )
+            .first()
+        )
+        return row
+
+    def confirm_section_mapping(
+        self, section_id: str, cre_db_id: str, user_id: str
+    ) -> bool:
+        """The user accepts a CRE for a section; returns False if not theirs
+        or if ``cre_db_id`` is not a CRE in the graph (validated at write)."""
+        row = self._owned_section(section_id, user_id)
+        if row is None or not self.get_CREs(internal_id=cre_db_id):
+            return False
+        row.cre_db_id = cre_db_id
+        row.status = SectionStatus.USER_CONFIRMED.value
+        row.updated_at = self._utcnow_naive()
+        self.session.commit()
+        return True
+
+    def reject_section_mapping(self, section_id: str, user_id: str) -> bool:
+        row = self._owned_section(section_id, user_id)
+        if row is None:
+            return False
+        row.status = SectionStatus.USER_REJECTED.value
+        row.updated_at = self._utcnow_naive()
+        self.session.commit()
+        return True
+
+    def get_cre_internal_id(self, external_id: str) -> Optional[str]:
+        """The CRE row's internal id for a public id like "287-251", or None.
+
+        The reverse of ``get_CREs(internal_id=...)``. Soft references such as
+        ``user_standard_sections.cre_db_id`` store the internal id; everything
+        a human sees is the external id, so writers translate through here.
+        """
+        row = CRE.query.filter(CRE.external_id == external_id).first()
+        internal_id: Optional[str] = row.id if row is not None else None
+        return internal_id
+
     def add_cre(self, cre: cre_defs.CRE) -> CRE:
         entry: CRE = None
         query = self.session.query(CRE).filter(func.lower(CRE.name) == cre.name.lower())
diff --git a/application/tests/mapping_job_test.py b/application/tests/mapping_job_test.py
new file mode 100644
index 000000000..a30566f36
--- /dev/null
+++ b/application/tests/mapping_job_test.py
@@ -0,0 +1,337 @@
+"""Tests for the MyOpenCRE mapping job (#965, RFC TODO 2 + 3).
+
+Hermetic and fixture-driven, as the RFC asks: an in-memory graph with a few
+CREs, one user with one standard and its sections, and stub Librarian
+components with fixed confidences. What is proved is the contract, not the
+model: deterministic status counts, the stored ``cre_db_id`` is a validated
+internal id, user decisions are never rewritten, and a re-run is a no-op.
+"""
+
+from cre_logging import get_logger
+
+logger = get_logger(__name__)
+
+import unittest
+from datetime import datetime, timezone
+
+from application import create_app, sqla
+from application.database import db
+from application.database.db import SectionStatus, User
+from application.defs import cre_defs as defs
+from application.utils.librarian.config_loader import LibrarianConfig
+from application.utils.librarian.factory import LibrarianComponents
+from application.utils.librarian.schemas import CreCandidate, RetrievalAudit
+from application.utils.mapping_job import (
+    REMAP_AFTER_REINDEX,
+    run_mapping,
+)
+
+AT = datetime(2026, 10, 24, tzinfo=timezone.utc).replace(tzinfo=None)
+
+
+def _config(threshold: float = 0.80) -> LibrarianConfig:
+    return LibrarianConfig(
+        crossencoder_model="stub",
+        retriever_backend="in_memory",
+        top_k_retrieval=20,
+        top_k_rerank=5,
+        link_threshold=threshold,
+        temperature=1.0,
+        batch_size=32,
+        ece_target=0.10,
+        conformal_alpha=0.10,
+    )
+
+
+class _Retriever:
+    """text -> candidate ids; a text with no entry retrieves nothing."""
+
+    def __init__(self, by_text):
+        self.by_text = by_text
+
+    def retrieve(self, text):
+        ids = self.by_text.get(text, [])
+        if ids == "boom":
+            raise RuntimeError("embedding timeout")
+        return RetrievalAudit(
+            retriever="stub/1.0.0",
+            candidates=[CreCandidate(cre_id=i, score_vector=0.9) for i in ids],
+            reranked=[],
+            threshold=0.0,
+        )
+
+
+class _Reranker:
+    def rerank(self, text, audit):
+        return audit.model_copy(
+            update={
+                "reranked": [
+                    CreCandidate(cre_id=c.cre_id, score_rerank=10.0)
+                    for c in audit.candidates
+                ]
+            }
+        )
+
+
+class _Scaler:
+    """Confidence keyed by the first logit's position: tests set one value
+    per section through the retriever's candidate id instead."""
+
+    def __init__(self, by_cre):
+        self.by_cre = by_cre
+        self.current = None
+
+    def confidence(self, logits):
+        return self.by_cre.get(self.current, 0.0)
+
+
+class _Components(LibrarianComponents):
+    pass
+
+
+def _components(by_text, conf_by_cre):
+    retriever = _Retriever(by_text)
+    scaler = _Scaler(conf_by_cre)
+
+    class _Tracking(_Retriever):
+        def retrieve(self, text):
+            audit = super().retrieve(text)
+            scaler.current = audit.candidates[0].cre_id if audit.candidates else None
+            return audit
+
+    return LibrarianComponents(
+        retriever=_Tracking(by_text),
+        reranker=_Reranker(),
+        scaler=scaler,
+        known_cre_ids=frozenset(conf_by_cre),
+    )
+
+
+class MappingJobTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.app = create_app(mode="test")
+        self.ctx = self.app.app_context()
+        self.ctx.push()
+        sqla.create_all()
+        self.collection = db.Node_collection()
+        self.user = User(
+            google_sub="sub-1", email="u@example.org", created_at=AT, last_seen_at=AT
+        )
+        self.other = User(
+            google_sub="sub-2", email="o@example.org", created_at=AT, last_seen_at=AT
+        )
+        sqla.session.add_all([self.user, self.other])
+        sqla.session.commit()
+        self.nonce = self.collection.add_cre(
+            defs.CRE(
+                id="287-251", name="Use a unique challenge nonce", tags=["Cryptography"]
+            )
+        )
+        self.salt = self.collection.add_cre(
+            defs.CRE(id="082-530", name="Unique random salt")
+        )
+        self.standard = self.collection.create_user_standard(self.user.id, "My Policy")
+        self.collection.upsert_user_standard_sections(
+            self.standard.id,
+            [
+                ("1.1", "challenge nonce must be 64 bits"),
+                ("1.2", "salt every credential"),
+                ("1.3", "sessions expire after logout"),
+                ("1.4", "something no CRE covers"),
+            ],
+        )
+        # retrieval is keyed by the graph's INTERNAL ids, as in production
+        self.by_text = {
+            "challenge nonce must be 64 bits": [self.nonce.id],
+            "salt every credential": [self.salt.id],
+            "sessions expire after logout": [self.nonce.id],
+            "something no CRE covers": [],
+        }
+        self.conf = {self.nonce.id: 0.95, self.salt.id: 0.42}
+
+    def tearDown(self) -> None:
+        sqla.session.remove()
+        sqla.drop_all()
+        self.ctx.pop()
+
+    def _rows(self):
+        return {
+            r.section_id: r
+            for r in self.collection.get_user_standard_sections(
+                self.standard.id, self.user.id
+            )
+        }
+
+    def test_upload_starts_every_section_pending(self) -> None:
+        self.assertEqual({r.status for r in self._rows().values()}, {"PENDING"})
+
+    def test_deterministic_status_counts(self) -> None:
+        summary = run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(self.by_text, self.conf),
+            config=_config(),
+        )
+
+        self.assertEqual(
+            (
+                summary.considered,
+                summary.auto_mapped,
+                summary.needs_review,
+                summary.unmappable,
+                summary.errored,
+            ),
+            (4, 2, 1, 1, 0),
+        )
+        rows = self._rows()
+        self.assertEqual(rows["1.1"].status, SectionStatus.AUTO_MAPPED.value)
+        self.assertEqual(rows["1.3"].status, SectionStatus.AUTO_MAPPED.value)
+        self.assertEqual(rows["1.2"].status, SectionStatus.NEEDS_REVIEW.value)
+        self.assertEqual(rows["1.4"].status, SectionStatus.UNMAPPABLE.value)
+
+    def test_stored_cre_db_id_is_the_validated_internal_id(self) -> None:
+        run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(self.by_text, self.conf),
+            config=_config(),
+        )
+        rows = self._rows()
+        self.assertEqual(rows["1.1"].cre_db_id, self.nonce.id)
+        self.assertEqual(rows["1.1"].confidence, 0.95)
+        # a review row keeps the candidate as the reviewer's hint
+        self.assertEqual(rows["1.2"].cre_db_id, self.salt.id)
+        self.assertEqual(rows["1.2"].confidence, 0.42)
+        self.assertIsNone(rows["1.4"].cre_db_id)
+        self.assertIsNone(rows["1.4"].confidence)
+
+    def test_candidate_the_graph_cannot_resolve_goes_to_review_without_a_reference(
+        self,
+    ) -> None:
+        by_text = dict(
+            self.by_text, **{"challenge nonce must be 64 bits": ["ghost-id"]}
+        )
+        conf = dict(self.conf, **{"ghost-id": 0.99})
+        run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(by_text, conf),
+            config=_config(),
+        )
+        row = self._rows()["1.1"]
+        self.assertEqual(row.status, SectionStatus.NEEDS_REVIEW.value)
+        self.assertIsNone(row.cre_db_id)
+
+    def test_rerun_is_a_no_op(self) -> None:
+        comps = _components(self.by_text, self.conf)
+        run_mapping(
+            self.standard.id, self.collection, components=comps, config=_config()
+        )
+        again = run_mapping(
+            self.standard.id, self.collection, components=comps, config=_config()
+        )
+        self.assertEqual(again.considered, 0)
+
+    def test_user_confirmed_rows_are_never_remapped(self) -> None:
+        run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(self.by_text, self.conf),
+            config=_config(),
+        )
+        row = self._rows()["1.2"]
+        self.assertTrue(
+            self.collection.confirm_section_mapping(row.id, self.nonce.id, self.user.id)
+        )
+        # a re-index re-run over every mappable status must leave it alone
+        run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(
+                self.by_text, {self.nonce.id: 0.1, self.salt.id: 0.1}
+            ),
+            config=_config(),
+            statuses=REMAP_AFTER_REINDEX,
+        )
+        row = self._rows()["1.2"]
+        self.assertEqual(row.status, SectionStatus.USER_CONFIRMED.value)
+        self.assertEqual(row.cre_db_id, self.nonce.id)
+        # ...while the undecided rows were re-mapped under the new confidences
+        self.assertEqual(self._rows()["1.1"].status, SectionStatus.NEEDS_REVIEW.value)
+
+    def test_errored_section_stays_pending_and_is_counted(self) -> None:
+        by_text = dict(self.by_text, **{"salt every credential": "boom"})
+        summary = run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(by_text, self.conf),
+            config=_config(),
+        )
+        self.assertEqual(summary.errored, 1)
+        self.assertEqual(self._rows()["1.2"].status, SectionStatus.PENDING.value)
+
+    def test_threshold_comes_from_the_librarian_config(self) -> None:
+        summary = run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(self.by_text, self.conf),
+            config=_config(threshold=0.40),
+        )
+        self.assertEqual(summary.auto_mapped, 3)
+        self.assertEqual(summary.needs_review, 0)
+
+    def test_confirm_rejects_a_cre_not_in_the_graph(self) -> None:
+        row = self._rows()["1.1"]
+        self.assertFalse(
+            self.collection.confirm_section_mapping(row.id, "nope", self.user.id)
+        )
+
+    def test_another_user_sees_nothing_and_cannot_decide(self) -> None:
+        self.assertEqual(
+            self.collection.get_user_standard_sections(self.standard.id, self.other.id),
+            [],
+        )
+        row = self._rows()["1.1"]
+        self.assertFalse(self.collection.reject_section_mapping(row.id, self.other.id))
+
+    def test_reupload_resets_changed_text_but_keeps_user_decisions(self) -> None:
+        run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(self.by_text, self.conf),
+            config=_config(),
+        )
+        rows = self._rows()
+        self.collection.confirm_section_mapping(
+            rows["1.1"].id, self.nonce.id, self.user.id
+        )
+        touched = self.collection.upsert_user_standard_sections(
+            self.standard.id,
+            [
+                ("1.1", "changed text"),
+                ("1.2", "changed text too"),
+                ("1.3", "sessions expire after logout"),
+            ],
+        )
+        self.assertEqual(touched, 2)
+        rows = self._rows()
+        self.assertEqual(rows["1.1"].status, SectionStatus.USER_CONFIRMED.value)
+        self.assertEqual(rows["1.2"].status, SectionStatus.PENDING.value)
+        self.assertIsNone(rows["1.2"].cre_db_id)
+        self.assertEqual(rows["1.3"].status, SectionStatus.AUTO_MAPPED.value)
+
+    def test_same_name_for_two_users_is_two_rows_and_duplicate_for_one_user_fails(
+        self,
+    ) -> None:
+        other_std = self.collection.create_user_standard(self.other.id, "My Policy")
+        self.assertNotEqual(other_std.id, self.standard.id)
+        self.assertEqual(len(self.collection.get_user_standards(self.user.id)), 1)
+        from sqlalchemy.exc import IntegrityError
+
+        with self.assertRaises(IntegrityError):
+            self.collection.create_user_standard(self.user.id, "My Policy")
+        sqla.session.rollback()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/application/utils/mapping_job.py b/application/utils/mapping_job.py
new file mode 100644
index 000000000..948be581a
--- /dev/null
+++ b/application/utils/mapping_job.py
@@ -0,0 +1,192 @@
+"""MyOpenCRE automated mapping: the Librarian applied to a user's standard.
+
+RFC user-auth-myopencre TODO 3, built on the maintainer decision in #965: there
+is no second mapping stack. ``run_mapping`` takes a user standard's PENDING
+sections and runs them through the same C.1 retrieve, C.2 rerank, C.3
+calibrate and C.4 decide the OIE pipeline runs, then records one outcome per
+section:
+
+- ``AUTO_MAPPED``   confidence at or above the Librarian's link threshold
+- ``NEEDS_REVIEW``  below it (the candidate is kept as the reviewer's hint)
+- ``UNMAPPABLE``    no candidate at all
+
+The status rule is ``decision_engine.decide`` itself, not a re-derivation of
+it, so MyOpenCRE and the pipeline cannot disagree about what a confidence
+means. ``USER_CONFIRMED`` and ``USER_REJECTED`` rows are never read or written
+here: user decisions are ground truth, and the write path in the database
+refuses to move them even if asked.
+
+This runs in a worker (the rq entry point is ``run_mapping_job``), which is
+why it may load the cross-encoder; the web process gets shortlists from
+``/rest/v1/link/suggest`` instead and never pays for the model.
+"""
+
+from cre_logging import get_logger
+
+logger = get_logger(__name__)
+
+import os
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Optional, Sequence
+
+from application.database.db import SectionStatus
+from application.utils.librarian.config_loader import LibrarianConfig, load_config
+from application.utils.librarian.decision_engine import Decision, ReasonCode, decide
+from application.utils.librarian.factory import LibrarianComponents
+from application.utils.librarian.link_suggest import _resolve_cre
+
+#: Which rows a run considers. Re-indexing the graph re-runs AUTO_MAPPED and
+#: NEEDS_REVIEW too (RFC TODO 3); a normal run only drains PENDING.
+PENDING_ONLY: Sequence[str] = (SectionStatus.PENDING.value,)
+REMAP_AFTER_REINDEX: Sequence[str] = (
+    SectionStatus.PENDING.value,
+    SectionStatus.AUTO_MAPPED.value,
+    SectionStatus.NEEDS_REVIEW.value,
+)
+
+
+@dataclass
+class MappingSummary:
+    """Outcome of one ``run_mapping`` call; ``to_dict`` for job results."""
+
+    standard_id: str
+    threshold: float
+    considered: int = 0
+    auto_mapped: int = 0
+    needs_review: int = 0
+    unmappable: int = 0
+    #: sections whose stage raised; left as they were so the next run retries
+    errored: int = 0
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+
+def _store_id_for(database: Any, candidate_id: str) -> Optional[str]:
+    """The CRE row id to persist for a retriever candidate, or None.
+
+    Retrievers hand back whatever the embeddings pool was keyed by (the CRE's
+    internal id today). ``cre_db_id`` is defined as the internal id and is
+    validated at write time, so a candidate the graph cannot resolve is not
+    stored: the section goes to review with no suggestion rather than carrying
+    a dangling reference.
+    """
+    cre = _resolve_cre(database, candidate_id)
+    if cre is None or not cre.id:
+        return None
+    internal_id: Optional[str] = database.get_cre_internal_id(cre.id)
+    return internal_id
+
+
+def run_mapping(
+    standard_id: str,
+    database: Any,
+    *,
+    components: Optional[LibrarianComponents] = None,
+    config: Optional[LibrarianConfig] = None,
+    statuses: Sequence[str] = PENDING_ONLY,
+) -> MappingSummary:
+    """Map every section of ``standard_id`` in ``statuses`` and persist outcomes.
+
+    Idempotent: a section leaves PENDING on success, so re-running the job
+    processes only what the previous run did not finish. Per-section failures
+    are contained and counted, never fatal to the batch.
+    """
+    config = config or load_config()
+    if components is None:
+        from application.utils.librarian.factory import build_components
+
+        components = build_components(database, config=config)
+    summary = MappingSummary(standard_id=standard_id, threshold=config.link_threshold)
+
+    for row in database.get_user_standard_sections_by_status(standard_id, statuses):
+        summary.considered += 1
+        try:
+            audit = components.retriever.retrieve(row.section_text)
+            audit = components.reranker.rerank(row.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] or [
+                c.cre_id for c in audit.candidates
+            ]
+            confidence = components.scaler.confidence(logits) if logits else 0.0
+            result = decide(confidence, cre_ids, threshold=config.link_threshold)
+        except Exception:
+            summary.errored += 1
+            logger.warning(
+                "mapping job: section %s of standard %s failed; left for retry",
+                row.section_id,
+                standard_id,
+                exc_info=True,
+            )
+            continue
+
+        if result.reason_code is ReasonCode.no_candidates:
+            database.set_section_mapping(
+                row.id,
+                cre_db_id=None,
+                confidence=None,
+                status=SectionStatus.UNMAPPABLE.value,
+            )
+            summary.unmappable += 1
+            continue
+
+        cre_db_id = _store_id_for(database, result.cre_ids[0])
+        if result.decision is Decision.linked and cre_db_id is not None:
+            status = SectionStatus.AUTO_MAPPED.value
+            summary.auto_mapped += 1
+        else:
+            # Below threshold, or a candidate the graph could not resolve: a
+            # human decides, with the Librarian's confidence beside it.
+            status = SectionStatus.NEEDS_REVIEW.value
+            summary.needs_review += 1
+        database.set_section_mapping(
+            row.id,
+            cre_db_id=cre_db_id,
+            confidence=round(result.confidence, 4),
+            status=status,
+        )
+
+    logger.info("mapping job: %s", summary.to_dict())
+    return summary
+
+
+def run_mapping_job(standard_id: str, db_connection_str: str) -> Dict[str, Any]:
+    """rq entry point: connect, map the PENDING sections, return the summary.
+
+    Registered by dotted path, like ``run_gap_pair_job``, so the worker needs
+    no import of the web app.
+    """
+    from application.cmd.cre_main import db_connect
+
+    database = db_connect(path=db_connection_str)
+    return run_mapping(standard_id, database).to_dict()
+
+
+def enqueue_mapping(standard_id: str, db_connection_str: str) -> str:
+    """Queue ``run_mapping_job`` on the mapping queue; returns the rq job id."""
+    from rq import Queue
+
+    from application.utils import redis
+
+    conn = redis.connect()  # type: ignore[no-untyped-call]
+    queue = Queue(
+        name=os.environ.get("CRE_MAPPING_QUEUE_NAME", "mapping"), connection=conn
+    )
+    job = queue.enqueue_call(
+        description=f"mapping:{standard_id}",
+        func="application.utils.mapping_job.run_mapping_job",
+        kwargs={"standard_id": standard_id, "db_connection_str": db_connection_str},
+        timeout=1800,
+    )
+    return str(job.id)
+
+
+__all__ = [
+    "MappingSummary",
+    "PENDING_ONLY",
+    "REMAP_AFTER_REINDEX",
+    "enqueue_mapping",
+    "run_mapping",
+    "run_mapping_job",
+]
diff --git a/migrations/versions/c4a1e7b2d9f0_add_user_standards.py b/migrations/versions/c4a1e7b2d9f0_add_user_standards.py
new file mode 100644
index 000000000..57958422e
--- /dev/null
+++ b/migrations/versions/c4a1e7b2d9f0_add_user_standards.py
@@ -0,0 +1,75 @@
+"""add user_standards and user_standard_sections for MyOpenCRE mapping (#965)
+
+RFC user-auth-myopencre TODO 2: a user-owned standard and its sections, with
+the mapping job's outcome per section. ``cre_db_id`` is a soft reference to
+the CRE row (validated at write time), never a foreign key, so the graph
+layer and the user layer stay decoupled.
+
+Revision ID: c4a1e7b2d9f0
+Revises: f3a91c07d24b
+Create Date: 2026-10-24
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "c4a1e7b2d9f0"
+down_revision = "f3a91c07d24b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "user_standards",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.String,
+            sa.ForeignKey("users.id", onupdate="CASCADE", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("name", sa.String, nullable=False),
+        sa.Column("created_at", sa.DateTime, nullable=False),
+        sa.Column("updated_at", sa.DateTime, nullable=False),
+        sa.UniqueConstraint("user_id", "name", name="uq_user_standards_user_name"),
+    )
+    op.create_table(
+        "user_standard_sections",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column(
+            "standard_id",
+            sa.String,
+            sa.ForeignKey("user_standards.id", onupdate="CASCADE", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("section_id", sa.String, nullable=False),
+        sa.Column("section_text", sa.Text, nullable=False),
+        sa.Column("cre_db_id", sa.String, nullable=True),
+        sa.Column("confidence", sa.Float, nullable=True),
+        sa.Column("status", sa.String, nullable=False),
+        sa.Column("created_at", sa.DateTime, nullable=False),
+        sa.Column("updated_at", sa.DateTime, nullable=False),
+        sa.UniqueConstraint(
+            "standard_id", "section_id", name="uq_user_standard_sections_section"
+        ),
+    )
+    op.create_index(
+        "ix_user_standard_sections_status",
+        "user_standard_sections",
+        ["standard_id", "status"],
+    )
+    op.create_index(
+        "ix_user_standard_sections_cre",
+        "user_standard_sections",
+        ["standard_id", "cre_db_id"],
+    )
+
+
+def downgrade():
+    op.drop_index("ix_user_standard_sections_cre", table_name="user_standard_sections")
+    op.drop_index(
+        "ix_user_standard_sections_status", table_name="user_standard_sections"
+    )
+    op.drop_table("user_standard_sections")
+    op.drop_table("user_standards")

From df2cfc1686de89237588c4d5d05ed660c8b5d454 Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sat, 24 Oct 2026 02:03:47 +0530
Subject: [PATCH 7/8] mapping job: atomic job writes, dedup uploads, fail
 loudly, consumed queue

Review fixes on the MyOpenCRE mapping job.

- set_section_mapping puts the user-decided guard in the UPDATE itself and
  returns whether a row moved, so a confirm that lands between the job's read
  and its write wins instead of being overwritten from a stale PENDING view.
- upsert_user_standard_sections treats a repeated section_id within one
  upload as an update of the row just added (last wins) instead of a second
  insert into the unique constraint.
- confirm_section_mapping clears confidence when the confirmed CRE differs
  from the job's candidate; the score was computed for a different CRE.
- run_mapping_job raises MappingPartialFailure after the loop when any
  section errored, so rq records the job as failed and the PENDING leftovers
  are visible and re-runnable; the committed updates stay.
- enqueue_mapping defaults to the "default" queue, which the stock worker
  listens to; "mapping" had no consumer. A test pins DEFAULT_QUEUE to
  application.worker.DEFAULT_LISTEN.
- get_user_standard_sections_by_status short-circuits an empty status list.
- Dead _Components class removed from the test module; seven tests added.
---
 application/database/db.py            | 61 ++++++++++++-------
 application/tests/mapping_job_test.py | 88 +++++++++++++++++++++++++--
 application/utils/mapping_job.py      | 40 +++++++++++-
 3 files changed, 160 insertions(+), 29 deletions(-)

diff --git a/application/database/db.py b/application/database/db.py
index 8beec0cc9..405f44312 100644
--- a/application/database/db.py
+++ b/application/database/db.py
@@ -2505,16 +2505,18 @@ def upsert_user_standard_sections(
         for section_id, section_text in sections:
             row = existing.get(section_id)
             if row is None:
-                self.session.add(
-                    UserStandardSection(
-                        standard_id=standard_id,
-                        section_id=section_id,
-                        section_text=section_text,
-                        status=SectionStatus.PENDING.value,
-                        created_at=now,
-                        updated_at=now,
-                    )
+                row = UserStandardSection(
+                    standard_id=standard_id,
+                    section_id=section_id,
+                    section_text=section_text,
+                    status=SectionStatus.PENDING.value,
+                    created_at=now,
+                    updated_at=now,
                 )
+                self.session.add(row)
+                # A repeated section_id later in the same upload must update
+                # this row, not insert a second one into the unique constraint.
+                existing[section_id] = row
                 touched += 1
                 continue
             if row.section_text == section_text:
@@ -2555,6 +2557,8 @@ def get_user_standard_sections_by_status(
     ) -> List[UserStandardSection]:
         """Job-side read, by status, in a stable order. No user scoping: the
         mapping job runs as the system on a standard id it was handed."""
+        if not statuses:
+            return []
         rows: List[UserStandardSection] = (
             UserStandardSection.query.filter(
                 UserStandardSection.standard_id == standard_id,
@@ -2572,20 +2576,29 @@ def set_section_mapping(
         cre_db_id: Optional[str],
         confidence: Optional[float],
         status: str,
-    ) -> None:
-        """Job-side write of one section's outcome. Never moves a user-decided
-        row, whatever it is asked to write: that invariant lives here so no
-        caller can break it by accident."""
-        row = UserStandardSection.query.filter(
-            UserStandardSection.id == section_id
-        ).first()
-        if row is None or row.status in USER_DECIDED_STATUSES:
-            return
-        row.cre_db_id = cre_db_id
-        row.confidence = confidence
-        row.status = status
-        row.updated_at = self._utcnow_naive()
+    ) -> bool:
+        """Job-side write of one section's outcome; True if a row moved.
+
+        Never moves a user-decided row, whatever it is asked to write, and the
+        check is part of the UPDATE itself: a user who confirms a section
+        between the job loading it and writing it wins, because the worker's
+        stale PENDING view is never consulted. That invariant lives here so no
+        caller can break it by accident.
+        """
+        updated = UserStandardSection.query.filter(
+            UserStandardSection.id == section_id,
+            UserStandardSection.status.notin_(list(USER_DECIDED_STATUSES)),
+        ).update(
+            {
+                UserStandardSection.cre_db_id: cre_db_id,
+                UserStandardSection.confidence: confidence,
+                UserStandardSection.status: status,
+                UserStandardSection.updated_at: self._utcnow_naive(),
+            },
+            synchronize_session=False,
+        )
         self.session.commit()
+        return bool(updated)
 
     def _owned_section(
         self, section_id: str, user_id: str
@@ -2609,6 +2622,10 @@ def confirm_section_mapping(
         row = self._owned_section(section_id, user_id)
         if row is None or not self.get_CREs(internal_id=cre_db_id):
             return False
+        if row.cre_db_id != cre_db_id:
+            # The stored confidence was computed for the job's candidate; it
+            # says nothing about the CRE the user chose instead.
+            row.confidence = None
         row.cre_db_id = cre_db_id
         row.status = SectionStatus.USER_CONFIRMED.value
         row.updated_at = self._utcnow_naive()
diff --git a/application/tests/mapping_job_test.py b/application/tests/mapping_job_test.py
index a30566f36..08b2ef38c 100644
--- a/application/tests/mapping_job_test.py
+++ b/application/tests/mapping_job_test.py
@@ -22,7 +22,11 @@
 from application.utils.librarian.factory import LibrarianComponents
 from application.utils.librarian.schemas import CreCandidate, RetrievalAudit
 from application.utils.mapping_job import (
+    DEFAULT_QUEUE,
     REMAP_AFTER_REINDEX,
+    MappingPartialFailure,
+    MappingSummary,
+    check_complete,
     run_mapping,
 )
 
@@ -85,10 +89,6 @@ def confidence(self, logits):
         return self.by_cre.get(self.current, 0.0)
 
 
-class _Components(LibrarianComponents):
-    pass
-
-
 def _components(by_text, conf_by_cre):
     retriever = _Retriever(by_text)
     scaler = _Scaler(conf_by_cre)
@@ -320,6 +320,86 @@ def test_reupload_resets_changed_text_but_keeps_user_decisions(self) -> None:
         self.assertIsNone(rows["1.2"].cre_db_id)
         self.assertEqual(rows["1.3"].status, SectionStatus.AUTO_MAPPED.value)
 
+    def test_duplicate_section_ids_in_one_upload_last_one_wins(self) -> None:
+        """A repeated section id must update the row added moments earlier in
+        the same call, not insert a second one into the unique constraint."""
+        std = self.collection.create_user_standard(self.user.id, "Dupes")
+        touched = self.collection.upsert_user_standard_sections(
+            std.id, [("9.1", "first text"), ("9.1", "second text"), ("9.2", "x")]
+        )
+        rows = {
+            r.section_id: r
+            for r in self.collection.get_user_standard_sections(std.id, self.user.id)
+        }
+        self.assertEqual(sorted(rows), ["9.1", "9.2"])
+        self.assertEqual(rows["9.1"].section_text, "second text")
+        self.assertEqual(touched, 3)
+
+    def test_confirming_a_different_cre_clears_the_jobs_confidence(self) -> None:
+        run_mapping(
+            self.standard.id,
+            self.collection,
+            components=_components(self.by_text, self.conf),
+            config=_config(),
+        )
+        # 1.2's candidate is salt at 0.42; the user picks nonce instead
+        row = self._rows()["1.2"]
+        self.collection.confirm_section_mapping(row.id, self.nonce.id, self.user.id)
+        self.assertIsNone(self._rows()["1.2"].confidence)
+        # confirming the job's own candidate keeps the score it was computed for
+        row = self._rows()["1.1"]
+        self.collection.confirm_section_mapping(row.id, self.nonce.id, self.user.id)
+        self.assertEqual(self._rows()["1.1"].confidence, 0.95)
+
+    def test_job_write_cannot_overwrite_a_decision_made_after_it_loaded(self) -> None:
+        """The status guard is in the UPDATE, so a user decision that lands
+        between the job's read and its write is the one that survives."""
+        row = self._rows()["1.1"]
+        self.collection.confirm_section_mapping(row.id, self.salt.id, self.user.id)
+        moved = self.collection.set_section_mapping(
+            row.id,
+            cre_db_id=self.nonce.id,
+            confidence=0.99,
+            status=SectionStatus.AUTO_MAPPED.value,
+        )
+        self.assertFalse(moved)
+        after = self._rows()["1.1"]
+        self.assertEqual(after.status, SectionStatus.USER_CONFIRMED.value)
+        self.assertEqual(after.cre_db_id, self.salt.id)
+
+    def test_job_write_moves_a_pending_row(self) -> None:
+        row = self._rows()["1.1"]
+        moved = self.collection.set_section_mapping(
+            row.id,
+            cre_db_id=self.nonce.id,
+            confidence=0.9,
+            status=SectionStatus.AUTO_MAPPED.value,
+        )
+        self.assertTrue(moved)
+        self.assertEqual(self._rows()["1.1"].status, SectionStatus.AUTO_MAPPED.value)
+
+    def test_partial_failure_is_raised_so_rq_records_it(self) -> None:
+        ok = MappingSummary(standard_id="s", threshold=0.8, considered=2, auto_mapped=2)
+        self.assertIs(check_complete(ok), ok)
+        bad = MappingSummary(
+            standard_id="s", threshold=0.8, considered=2, auto_mapped=1, errored=1
+        )
+        with self.assertRaises(MappingPartialFailure) as caught:
+            check_complete(bad)
+        self.assertIs(caught.exception.summary, bad)
+        self.assertIn("1 of 2", str(caught.exception))
+
+    def test_default_queue_is_one_the_stock_worker_listens_to(self) -> None:
+        from application import worker
+
+        self.assertIn(DEFAULT_QUEUE, worker.DEFAULT_LISTEN)
+
+    def test_empty_status_list_reads_nothing(self) -> None:
+        self.assertEqual(
+            self.collection.get_user_standard_sections_by_status(self.standard.id, []),
+            [],
+        )
+
     def test_same_name_for_two_users_is_two_rows_and_duplicate_for_one_user_fails(
         self,
     ) -> None:
diff --git a/application/utils/mapping_job.py b/application/utils/mapping_job.py
index 948be581a..6e85e92d8 100644
--- a/application/utils/mapping_job.py
+++ b/application/utils/mapping_job.py
@@ -35,6 +35,12 @@
 from application.utils.librarian.factory import LibrarianComponents
 from application.utils.librarian.link_suggest import _resolve_cre
 
+#: The queue ``enqueue_mapping`` uses unless ``CRE_MAPPING_QUEUE_NAME`` says
+#: otherwise. It must be one the stock worker listens to
+#: (``application.worker.DEFAULT_LISTEN``), or jobs sit queued forever; a test
+#: pins the two together.
+DEFAULT_QUEUE = "default"
+
 #: Which rows a run considers. Re-indexing the graph re-runs AUTO_MAPPED and
 #: NEEDS_REVIEW too (RFC TODO 3); a normal run only drains PENDING.
 PENDING_ONLY: Sequence[str] = (SectionStatus.PENDING.value,)
@@ -151,16 +157,41 @@ def run_mapping(
     return summary
 
 
+class MappingPartialFailure(RuntimeError):
+    """Some sections errored. Raised so rq records the job as failed; the
+    sections that did map are already committed and are not retried."""
+
+    def __init__(self, summary: MappingSummary) -> None:
+        self.summary = summary
+        super().__init__(
+            f"mapping of standard {summary.standard_id}: {summary.errored} of "
+            f"{summary.considered} sections errored and were left PENDING"
+        )
+
+
+def check_complete(summary: MappingSummary) -> MappingSummary:
+    """Return ``summary`` if every considered section was mapped, else raise.
+
+    A job that swallowed failures would report success to rq and nothing
+    would ever retry the PENDING leftovers; raising is what makes the partial
+    run visible on the queue and re-runnable.
+    """
+    if summary.errored:
+        raise MappingPartialFailure(summary)
+    return summary
+
+
 def run_mapping_job(standard_id: str, db_connection_str: str) -> Dict[str, Any]:
     """rq entry point: connect, map the PENDING sections, return the summary.
 
     Registered by dotted path, like ``run_gap_pair_job``, so the worker needs
-    no import of the web app.
+    no import of the web app. Raises ``MappingPartialFailure`` when any
+    section errored, after the successful ones are committed.
     """
     from application.cmd.cre_main import db_connect
 
     database = db_connect(path=db_connection_str)
-    return run_mapping(standard_id, database).to_dict()
+    return check_complete(run_mapping(standard_id, database)).to_dict()
 
 
 def enqueue_mapping(standard_id: str, db_connection_str: str) -> str:
@@ -171,7 +202,7 @@ def enqueue_mapping(standard_id: str, db_connection_str: str) -> str:
 
     conn = redis.connect()  # type: ignore[no-untyped-call]
     queue = Queue(
-        name=os.environ.get("CRE_MAPPING_QUEUE_NAME", "mapping"), connection=conn
+        name=os.environ.get("CRE_MAPPING_QUEUE_NAME", DEFAULT_QUEUE), connection=conn
     )
     job = queue.enqueue_call(
         description=f"mapping:{standard_id}",
@@ -183,7 +214,10 @@ def enqueue_mapping(standard_id: str, db_connection_str: str) -> str:
 
 
 __all__ = [
+    "DEFAULT_QUEUE",
+    "MappingPartialFailure",
     "MappingSummary",
+    "check_complete",
     "PENDING_ONLY",
     "REMAP_AFTER_REINDEX",
     "enqueue_mapping",

From 76906fa12340adc4da9a0c13949c611fa8718f28 Mon Sep 17 00:00:00 2001
From: PRAteek-singHWY <prateek23022004@gmail.com>
Date: Sun, 25 Oct 2026 13:56:03 +0530
Subject: [PATCH 8/8] mapping job test: re-upload also keeps the confirmed CRE
 reference

---
 application/tests/mapping_job_test.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/application/tests/mapping_job_test.py b/application/tests/mapping_job_test.py
index 08b2ef38c..3fb31e96f 100644
--- a/application/tests/mapping_job_test.py
+++ b/application/tests/mapping_job_test.py
@@ -316,6 +316,7 @@ def test_reupload_resets_changed_text_but_keeps_user_decisions(self) -> None:
         self.assertEqual(touched, 2)
         rows = self._rows()
         self.assertEqual(rows["1.1"].status, SectionStatus.USER_CONFIRMED.value)
+        self.assertEqual(rows["1.1"].cre_db_id, self.nonce.id)
         self.assertEqual(rows["1.2"].status, SectionStatus.PENDING.value)
         self.assertIsNone(rows["1.2"].cre_db_id)
         self.assertEqual(rows["1.3"].status, SectionStatus.AUTO_MAPPED.value)