M10: harden knowledge trust and telemetry

backend/app/api/routers/demo.py

@@ -110,7 +110,7 @@ def demo_reset(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Demo reset is disabled on this deployment.",
         )
-    result = reset_and_seed(db)
+    result = reset_and_seed(db, preserve_integration_telemetry=True)
     integrity = scenario_integrity_report(db)
     record_audit_event(
         db,

backend/app/seed_loader.py

@@ -80,9 +80,17 @@ def _read_csv(name: str) -> list[dict[str, str]]:
         return list(csv.DictReader(handle))
 
 
-def clear_all(db: Session) -> None:
+_PERSISTENT_TELEMETRY_ACTIONS = (
+    "mcp_tool_request",
+    "n8n_return_followup_recorded",
+    "n8n_workflow_failure_registered",
+    "n8n_procedures_synced",
+    "knowledge_question_asked",
+)
+
+
+def clear_all(db: Session, *, preserve_integration_telemetry: bool = False) -> None:
     for model in (
-        AuditEvent,
         OutboxEvent,
         IdempotencyRecord,
         DataQualityIssue,
@@ -94,6 +102,10 @@ def clear_all(db: Session) -> None:
         User,
     ):
         db.execute(delete(model))
+    if preserve_integration_telemetry:
+        db.execute(delete(AuditEvent).where(AuditEvent.action.not_in(_PERSISTENT_TELEMETRY_ACTIONS)))
+    else:
+        db.execute(delete(AuditEvent))
 
 
 def load_seed(db: Session) -> SeedResult:
@@ -428,10 +440,12 @@ def load_seed(db: Session) -> SeedResult:
     return SeedResult(counts=counts, anchor_date=today, seeded_at=seeded_at)
 
 
-def reset_and_seed(db: Session) -> SeedResult:
+def reset_and_seed(
+    db: Session, *, preserve_integration_telemetry: bool = False
+) -> SeedResult:
     from app.services.data_quality import run_scan
 
-    clear_all(db)
+    clear_all(db, preserve_integration_telemetry=preserve_integration_telemetry)
     result = load_seed(db)
     db.commit()
     scan = run_scan(db)

backend/app/services/knowledge/ragcore.py

@@ -21,6 +21,54 @@ _LEAD_ANSWER_TEMPLATE = {
 }
 _DEFAULT_LANGUAGE = "en-GB"
 
+_DOMAIN_CONCEPTS: dict[str, tuple[str, ...]] = {
+    "damage": ("damage", "damaged", "schade", "beschadigd", "dommage", "endommagé"),
+    "vehicle": ("vehicle", "car", "voertuig", "wagen", "véhicule", "voiture"),
+    "return": ("return", "returned", "retour", "terugbrengen", "restitution"),
+    "fuel": ("fuel", "brandstof", "carburant"),
+    "booking": ("booking", "reservation", "boeking", "réservation"),
+    "odometer": ("odometer", "mileage", "kilometer", "kilométrage", "compteur"),
+    "customer": ("customer", "client", "klant"),
+    "cleaning": ("cleaning", "reiniging", "poetsen", "nettoyage"),
+    "maintenance": ("maintenance", "service", "onderhoud", "entretien"),
+    "conflict": ("conflict", "overlap", "overlapping", "conflit", "chevauchement"),
+    "checkout": ("checkout", "departure", "vertrek", "départ"),
+    "availability": ("available", "availability", "beschikbaar", "disponible", "disponibilité"),
+    "technical": ("technical", "warning", "technisch", "waarschuwing", "technique", "alerte"),
+}
+
+
+def _question_concepts(question: str) -> set[str]:
+    normalized = question.casefold()
+    return {
+        concept
+        for concept, terms in _DOMAIN_CONCEPTS.items()
+        if any(term in normalized for term in terms)
+    }
+
+
+def _deduplicate_sources(sources: list[SourceCard]) -> list[SourceCard]:
+    seen: set[tuple[str, str]] = set()
+    unique: list[SourceCard] = []
+    for source in sources:
+        key = (source.document_id, source.section)
+        if key in seen:
+            continue
+        seen.add(key)
+        unique.append(source)
+    return unique
+
+
+def _rank_sources_for_concepts(sources: list[SourceCard], concepts: set[str]) -> list[SourceCard]:
+    if "damage" not in concepts:
+        return sources
+    return sorted(
+        sources,
+        key=lambda source: (
+            0 if "damage" in f"{source.document_id} {source.title}".casefold() else 1
+        ),
+    )
+
 
 class RAGcoreKnowledgeProvider:
     """Adapter for the central RAGcore service, against its real `/v1/*` contract
@@ -199,6 +247,9 @@ class RAGcoreKnowledgeProvider:
         except (TypeError, KeyError, ValueError):
             return unavailable
 
+        sources = _deduplicate_sources(sources)
+        concepts = _question_concepts(question)
+        sources = _rank_sources_for_concepts(sources, concepts)
         if not sources:
             return GroundedAnswer(
                 answer="",
@@ -208,6 +259,15 @@ class RAGcoreKnowledgeProvider:
                 correlation_id=correlation_id,
             )
 
+        if not concepts:
+            return GroundedAnswer(
+                answer="",
+                evidence_state="insufficient",
+                sources=sources,
+                provider=self.name,
+                correlation_id=correlation_id,
+            )
+
         template = _LEAD_ANSWER_TEMPLATE.get(language, _LEAD_ANSWER_TEMPLATE[_DEFAULT_LANGUAGE])
         lead = sources[0]
         answer = template.format(title=lead.title, excerpt=lead.excerpt)

backend/tests/test_auth.py

@@ -65,3 +65,22 @@ def test_logout_invalidates_session(ops_client):
 def test_logout_without_a_session_is_safe(client):
     response = client.post("/api/v1/demo/logout")
     assert response.status_code == 200
+
+
+def test_demo_reset_preserves_integration_telemetry(ops_client, client):
+    from app.core.config import get_settings
+
+    settings = get_settings()
+    probe = client.get(
+        "/api/v1/integrations/mcp/operations-summary",
+        headers={
+            "X-Service-Token": settings.mcp_hub_service_token,
+            "X-Client-Id": "reset-probe",
+        },
+    )
+    assert probe.status_code == 200
+    assert ops_client.post("/api/v1/demo/reset").status_code == 200
+
+    assert client.post("/api/v1/demo/login", json={"role": "operations_manager"}).status_code == 200
+    events = client.get("/api/v1/audit", params={"action": "mcp_tool_request"}).json()
+    assert any(event["actor_label"] == "reset-probe" for event in events)

backend/tests/test_knowledge.py

@@ -450,3 +450,56 @@ def test_ragcore_provider_fallback_with_no_search_results_is_insufficient(monkey
     assert answer.evidence_state == "insufficient"
     assert answer.answer == ""
     assert answer.sources == []
+
+
+def test_ragcore_search_fallback_rejects_out_of_domain_question(monkeypatch):
+    provider = RAGcoreKnowledgeProvider()
+    monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
+    monkeypatch.setattr(
+        provider,
+        "_client",
+        lambda: _FakeClient(
+            post_responses={
+                "/v1/answers": _FakeResponse(503, {}),
+                "/v1/search": _FakeResponse(200, _search_body()),
+            }
+        ),
+    )
+    answer = provider.ask(
+        "Who won the football world cup in 1998?",
+        "test-correlation-out-of-domain",
+    )
+    assert answer.evidence_state == "insufficient"
+    assert answer.answer == ""
+
+
+def test_ragcore_search_fallback_prefers_damage_procedure(monkeypatch):
+    provider = RAGcoreKnowledgeProvider()
+    monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
+    search = _search_body()
+    search["results"].append(
+        {
+            "citation": {
+                "document_id": "damage-procedure",
+                "document_version_id": "version-2",
+                "title": "damage-procedure.md",
+                "section": "Damage",
+                "excerpt": "Record damage and keep the vehicle blocked.",
+            },
+            "rank": 2,
+            "scores": {"fused": 0.01},
+        }
+    )
+    monkeypatch.setattr(
+        provider,
+        "_client",
+        lambda: _FakeClient(
+            post_responses={
+                "/v1/answers": _FakeResponse(503, {}),
+                "/v1/search": _FakeResponse(200, search),
+            }
+        ),
+    )
+    answer = provider.ask("Wat moet ik doen bij schade?", "test-correlation-damage", "nl-BE")
+    assert answer.evidence_state == "grounded"
+    assert answer.sources[0].document_id == "damage-procedure"