M41: harden trust boundaries and delivery

backend/app/services/data_quality.py

@@ -5,7 +5,7 @@ from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from difflib import SequenceMatcher
 
-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 from sqlalchemy.orm import Session
 
 from app.core.errors import AppError
@@ -26,6 +26,7 @@ from app.services.vehicle_status import (
 REQUIRED_CUSTOMER_FIELDS = ("first_name", "last_name")
 REQUIRED_VEHICLE_FIELDS = ("registration_number", "make", "model", "location")
 DUPLICATE_THRESHOLD = 70
+DATA_QUALITY_SCAN_LOCK_ID = 6_138_493_717_091_029_491
 
 
 def issue_due_at(detected_at: datetime, severity: str) -> datetime:
@@ -356,6 +357,9 @@ def _scan_odometer_regressions(db: Session, scan: ScanResult) -> None:
 def run_scan(
     db: Session, *, actor_label: str | None = None, actor_type: str = "user"
 ) -> ScanResult:
+    # The check-then-insert work below spans several rules. Serialise whole scans at the
+    # database boundary so API and n8n triggers cannot both observe an empty condition.
+    db.scalar(select(func.pg_advisory_xact_lock(DATA_QUALITY_SCAN_LOCK_ID)))
     scan = ScanResult()
     _scan_duplicate_customers(db, scan)
     _scan_missing_required_fields(db, scan)
@@ -375,8 +379,13 @@ def run_scan(
     return scan
 
 
-def _load_open_issue(db: Session, public_ref: str) -> DataQualityIssue:
-    issue = db.scalar(select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref))
+def _load_open_issue(
+    db: Session, public_ref: str, *, lock: bool = True
+) -> DataQualityIssue:
+    statement = select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref)
+    if lock:
+        statement = statement.with_for_update()
+    issue = db.scalar(statement)
     if issue is None:
         raise AppError("ISSUE_NOT_FOUND", "Data quality issue not found.", status_code=404)
     if issue.status != "open":
@@ -707,8 +716,10 @@ def resolve_booking_overlap(
     return issue
 
 
-def _load_vehicle_status_conflict_issue(db: Session, public_ref: str) -> DataQualityIssue:
-    issue = _load_open_issue(db, public_ref)
+def _load_vehicle_status_conflict_issue(
+    db: Session, public_ref: str, *, lock: bool = True
+) -> DataQualityIssue:
+    issue = _load_open_issue(db, public_ref, lock=lock)
     if issue.rule_type != "vehicle_status_conflict":
         raise AppError(
             "NOT_A_STATUS_CONFLICT_ISSUE",
@@ -724,7 +735,7 @@ def preview_vehicle_status_recommendation(
     """Non-mutating: computes and returns the recommendation only. Never resolves the
     issue, never writes an audit event, never queues automation -- safe to call as often
     as the UI needs (e.g. every time the panel is opened) with zero side effects."""
-    issue = _load_vehicle_status_conflict_issue(db, public_ref)
+    issue = _load_vehicle_status_conflict_issue(db, public_ref, lock=False)
     vehicle = db.scalar(select(Vehicle).where(Vehicle.id == issue.entity_id))
     if vehicle is None:
         raise AppError(

backend/app/services/knowledge/ragcore.py

@@ -92,6 +92,17 @@ def _rank_sources_for_concepts(sources: list[SourceCard], concepts: set[str]) ->
     )
 
 
+def _retrieval_score(result: dict) -> float | None:
+    scores = result.get("scores")
+    if not isinstance(scores, dict):
+        return None
+    for name in ("rerank", "fused"):
+        value = scores.get(name)
+        if isinstance(value, int | float) and not isinstance(value, bool):
+            return float(value)
+    return None
+
+
 class RAGcoreKnowledgeProvider:
     """Adapter for the central RAGcore service, against its real `/v1/*` contract
     (see `docs/contracts/openapi.yaml` in the RAGcore checkout -- RAGcore is built and
@@ -338,6 +349,8 @@ class RAGcoreKnowledgeProvider:
 
         try:
             results = body.get("results", [])
+            if not isinstance(results, list):
+                raise TypeError("results must be a list")
             sources = [
                 SourceCard(
                     document_id=str(result["citation"]["document_id"]),
@@ -351,10 +364,21 @@ class RAGcoreKnowledgeProvider:
         except (TypeError, KeyError, ValueError):
             return unavailable
 
-        sources = _deduplicate_sources(sources)
+        all_sources = _deduplicate_sources(sources)
         concepts = _question_concepts(question)
-        sources = _rank_sources_for_concepts(sources, concepts)
-        if not sources:
+        qualified_sources = [
+            SourceCard(
+                document_id=str(result["citation"]["document_id"]),
+                title=result["citation"]["title"],
+                version=str(result["citation"]["document_version_id"]),
+                section=result["citation"].get("section") or "",
+                excerpt=result["citation"]["excerpt"],
+            )
+            for result in results
+            if (_retrieval_score(result) or 0.0) >= self._settings.ragcore_min_search_score
+        ]
+        sources = _rank_sources_for_concepts(_deduplicate_sources(qualified_sources), concepts)
+        if not all_sources:
             return GroundedAnswer(
                 answer="",
                 evidence_state="insufficient",
@@ -363,11 +387,19 @@ class RAGcoreKnowledgeProvider:
                 correlation_id=correlation_id,
             )
 
-        if not concepts:
+        damage_evidence = any(
+            term
+            in (
+                f"{source.document_id} {source.title} {source.section} {source.excerpt}"
+            ).casefold()
+            for source in sources
+            for term in _DOMAIN_CONCEPTS["damage"]
+        )
+        if not concepts or not sources or ("damage" in concepts and not damage_evidence):
             return GroundedAnswer(
                 answer="",
                 evidence_state="insufficient",
-                sources=sources,
+                sources=all_sources,
                 provider=self.name,
                 correlation_id=correlation_id,
             )