diff --git a/application/database/db.py b/application/database/db.py index 22e7358be..405f44312 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,201 @@ 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: + 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: + 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.""" + if not statuses: + return [] + 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, + ) -> 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 + ) -> 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 + 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() + 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/librarian/link_suggest_test.py b/application/tests/librarian/link_suggest_test.py new file mode 100644 index 000000000..6a2c1fc15 --- /dev/null +++ b/application/tests/librarian/link_suggest_test.py @@ -0,0 +1,374 @@ +"""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, "--3", "\u00b2", "2_0", " 7 x", True): + 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_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.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 + 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_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_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}") + self.assertEqual(resp.status_code, 400, bad) + + 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()) + # 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__": + unittest.main() diff --git a/application/tests/mapping_job_test.py b/application/tests/mapping_job_test.py new file mode 100644 index 000000000..3fb31e96f --- /dev/null +++ b/application/tests/mapping_job_test.py @@ -0,0 +1,418 @@ +"""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 ( + DEFAULT_QUEUE, + REMAP_AFTER_REINDEX, + MappingPartialFailure, + MappingSummary, + check_complete, + 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) + + +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.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) + + 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: + 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/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..ab9096cc3 --- /dev/null +++ b/application/utils/librarian/link_suggest.py @@ -0,0 +1,209 @@ +"""``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__) + +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"[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. +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): + # 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: + 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. + """ + # 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} + ) + # 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] + return found + + +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/utils/mapping_job.py b/application/utils/mapping_job.py new file mode 100644 index 000000000..6e85e92d8 --- /dev/null +++ b/application/utils/mapping_job.py @@ -0,0 +1,226 @@ +"""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 + +#: 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,) +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 + + +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. 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 check_complete(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", DEFAULT_QUEUE), 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__ = [ + "DEFAULT_QUEUE", + "MappingPartialFailure", + "MappingSummary", + "check_complete", + "PENDING_ONLY", + "REMAP_AFTER_REINDEX", + "enqueue_mapping", + "run_mapping", + "run_mapping_job", +] diff --git a/application/web/web_main.py b/application/web/web_main.py index 094f818e8..4d0da4f97 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -1285,6 +1285,80 @@ 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 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( + { + "error": ( + "The AI service is temporarily rate-limited or out of quota. " + "Please try again in a minute." + ) + } + ), + 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 + 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) + + class CREFlow: """ "This class handles authentication with google's oauth""" 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")