M31: verify RAG inventory and polish attention queue

backend/app/api/routers/knowledge.py

@@ -124,6 +124,12 @@ def knowledge_status(
             "reported_synced_document_count": synced if isinstance(synced, int) else None,
             "reported_failed_document_count": failed if isinstance(failed, int) else None,
             "last_sync_at": latest_sync.occurred_at,
-            "statistics_state": "sync_reported",
+            # A persisted sync callback is useful additional provenance, but must not
+            # downgrade stronger provider-side verification to merely "reported".
+            "statistics_state": (
+                health.statistics_state
+                if health.statistics_state == "verified"
+                else "sync_reported"
+            ),
         }
     )

backend/app/services/knowledge/ragcore.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
+from threading import Lock
+from time import monotonic
 
 import httpx
 
 from app.core.config import get_settings
 from app.services.knowledge import EvidenceState, GroundedAnswer, KnowledgeHealth, SourceCard
-from app.services.knowledge.procedures import iter_procedure_documents
+from app.services.knowledge.procedures import ProcedureDocument, iter_procedure_documents
 
 _GROUNDED_ANSWERABILITY = {"answerable", "partially_answerable"}
 
@@ -24,6 +27,9 @@ _LEAD_ANSWER_TEMPLATE = {
 }
 _DEFAULT_LANGUAGE = "en-GB"
 _MAX_SOURCE_CARDS = 3
+_INDEX_LOOKUP_TIMEOUT_SECONDS = 2.0
+_INDEX_VERIFICATION_TTL_SECONDS = 300.0
+_INDEX_VERIFICATION_WORKERS = 6
 
 _DOMAIN_CONCEPTS: dict[str, tuple[str, ...]] = {
     "damage": ("damage", "damaged", "schade", "beschadigd", "dommage", "endommagé"),
@@ -118,6 +124,8 @@ class RAGcoreKnowledgeProvider:
 
     def __init__(self) -> None:
         self._settings = get_settings()
+        self._verification_cache: dict[str, tuple[float, int]] = {}
+        self._verification_lock = Lock()
 
     def _client(self) -> httpx.Client:
         headers = {}
@@ -130,6 +138,12 @@ class RAGcoreKnowledgeProvider:
         )
 
     def health(self, language: str = "en-GB") -> KnowledgeHealth:
+        documents = [
+            document
+            for document in iter_procedure_documents(Path(self._settings.knowledge_dir))
+            if document.language == language
+        ]
+        verified_document_count: int | None = None
         try:
             with self._client() as client:
                 response = client.get("/health/ready")
@@ -140,13 +154,20 @@ class RAGcoreKnowledgeProvider:
                     if available
                     else f"RAGcore degraded: {body.get('status', 'unknown')}"
                 )
+                if available:
+                    verified_document_count = self._verify_indexed_documents(
+                        client, language, documents
+                    )
+                    if verified_document_count is None:
+                        detail += " Index verification is temporarily unavailable."
+                    else:
+                        detail += (
+                            f" {verified_document_count}/{len(documents)} managed sources have "
+                            "an active published version with the expected content hash."
+                        )
         except (httpx.HTTPError, ValueError) as exc:
             available = False
             detail = f"RAGcore unavailable: {type(exc).__name__}: {exc}"
-        source_document_count = sum(
-            document.language == language
-            for document in iter_procedure_documents(Path(self._settings.knowledge_dir))
-        )
         return KnowledgeHealth(
             provider=self.name,
             available=available,
@@ -154,16 +175,70 @@ class RAGcoreKnowledgeProvider:
             tenant=self._settings.ragcore_tenant,
             workspace=self._settings.ragcore_workspace,
             collection=self._settings.ragcore_collection,
-            # RAGcore's retrieval API has no corpus-size endpoint. Unknown is explicit
-            # so the UI never turns this into the misleading claim "0 procedures".
-            document_count=None,
-            source_document_count=source_document_count,
+            # RAGcore deliberately has no browse/count endpoint. Fleet Ops instead
+            # verifies each managed source through its exact identity lookup and only
+            # counts an active published version whose content hash still matches.
+            document_count=verified_document_count,
+            source_document_count=len(documents),
             reported_synced_document_count=None,
             reported_failed_document_count=None,
             last_sync_at=None,
-            statistics_state="not_reported",
+            statistics_state=(
+                "verified" if verified_document_count is not None else "not_reported"
+            ),
         )
 
+    def _verify_indexed_documents(
+        self, client: httpx.Client, language: str, documents: list[ProcedureDocument]
+    ) -> int | None:
+        if not self._settings.ragcore_space_id or not documents:
+            return None
+
+        now = monotonic()
+        with self._verification_lock:
+            cached = self._verification_cache.get(language)
+            if cached is not None and now - cached[0] < _INDEX_VERIFICATION_TTL_SECONDS:
+                return cached[1]
+
+        def is_verified(document: ProcedureDocument) -> bool:
+            response = client.get(
+                "/v1/documents",
+                params={
+                    "source_id": document.source_id,
+                    "external_id": f"{document.document_id}.md",
+                },
+                timeout=_INDEX_LOOKUP_TIMEOUT_SECONDS,
+            )
+            if response.status_code != 200:
+                raise RuntimeError("RAGcore document verification failed")
+            body = response.json()
+            items = body.get("items") if isinstance(body, dict) else None
+            if not isinstance(items, list) or len(items) != 1 or not isinstance(items[0], dict):
+                return False
+            item = items[0]
+            active_version = item.get("active_version")
+            return bool(
+                item.get("space_id") == self._settings.ragcore_space_id
+                and item.get("source_id") == document.source_id
+                and item.get("external_id") == f"{document.document_id}.md"
+                and item.get("status") == "active"
+                and isinstance(active_version, dict)
+                and active_version.get("status") == "published"
+                and active_version.get("content_sha256") == document.content_hash
+            )
+
+        try:
+            with ThreadPoolExecutor(
+                max_workers=min(_INDEX_VERIFICATION_WORKERS, len(documents))
+            ) as executor:
+                verified_count = sum(executor.map(is_verified, documents))
+        except (httpx.HTTPError, RuntimeError, TypeError, ValueError):
+            return None
+
+        with self._verification_lock:
+            self._verification_cache[language] = (monotonic(), verified_count)
+        return verified_count
+
     def ask(self, question: str, correlation_id: str, language: str = "en-GB") -> GroundedAnswer:
         unavailable = GroundedAnswer(
             answer="",

backend/tests/test_knowledge.py

@@ -9,6 +9,7 @@ from app.api.routers import knowledge as knowledge_router
 from app.core.config import get_settings
 from app.services.knowledge import KnowledgeHealth
 from app.services.knowledge.demo import DemoKnowledgeProvider
+from app.services.knowledge.procedures import iter_procedure_documents
 from app.services.knowledge.ragcore import RAGcoreKnowledgeProvider
 
 
@@ -198,6 +199,39 @@ def test_ragcore_status_separates_sync_report_from_unverifiable_index(
     assert status["statistics_state"] == "sync_reported"
 
 
+def test_ragcore_status_preserves_stronger_verified_index_evidence(client, ops_client, monkeypatch):
+    class VerifiedRagcoreProvider:
+        def health(self, language="en-GB"):
+            return KnowledgeHealth(
+                provider="ragcore",
+                available=True,
+                detail="ready and verified",
+                tenant="fleet-ops",
+                workspace="operations",
+                collection="internal-procedures",
+                document_count=11,
+                source_document_count=11,
+                reported_synced_document_count=None,
+                reported_failed_document_count=None,
+                last_sync_at=None,
+                statistics_state="verified",
+            )
+
+    monkeypatch.setattr(knowledge_router, "get_knowledge_provider", VerifiedRagcoreProvider)
+    settings = get_settings()
+    sync = client.post(
+        "/api/v1/integrations/n8n/procedures-sync-result",
+        json={"execution_id": "rag-verified-statistics-test", "synced": 33, "failed": 0},
+        headers={"X-Service-Token": settings.n8n_callback_token},
+    )
+    assert sync.status_code == 200
+
+    status = ops_client.get("/api/v1/knowledge/status?language=nl-BE").json()
+    assert status["document_count"] == 11
+    assert status["reported_synced_document_count"] == 33
+    assert status["statistics_state"] == "verified"
+
+
 class _FakeResponse:
     def __init__(self, status_code: int, body: dict):
         self.status_code = status_code
@@ -208,7 +242,14 @@ class _FakeResponse:
 
 
 class _FakeClient:
-    def __init__(self, get_response=None, post_response=None, post_responses=None, raise_on=None):
+    def __init__(
+        self,
+        get_response=None,
+        post_response=None,
+        post_responses=None,
+        raise_on=None,
+        get_handler=None,
+    ):
         self._get_response = get_response
         self._post_response = post_response
         # Maps a path (e.g. "/v1/search") to its own response, for tests that need
@@ -217,6 +258,7 @@ class _FakeClient:
         # existing single-endpoint test keeps working unchanged.
         self._post_responses = post_responses or {}
         self._raise_on = raise_on
+        self._get_handler = get_handler
 
     def __enter__(self):
         return self
@@ -224,9 +266,13 @@ class _FakeClient:
     def __exit__(self, *args):
         return False
 
-    def get(self, path):
+    def get(self, path, params=None, timeout=None):
         if self._raise_on == "get":
             raise httpx.ConnectError("no ragcore in this environment")
+        if self._get_handler is not None:
+            return self._get_handler(path, params, timeout)
+        if path == "/v1/documents":
+            return _FakeResponse(503, {})
         return self._get_response
 
     def post(self, path, json=None):
@@ -273,6 +319,50 @@ def test_ragcore_provider_health_reports_ready_status(monkeypatch):
     assert health.statistics_state == "not_reported"
 
 
+def test_ragcore_provider_verifies_published_documents_and_caches_count(monkeypatch):
+    provider = RAGcoreKnowledgeProvider()
+    monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
+    procedure_documents = [
+        document
+        for document in iter_procedure_documents(Path(provider._settings.knowledge_dir))
+        if document.language == "en-GB"
+    ]
+    documents = {document.source_id: document for document in procedure_documents}
+    lookups: list[str] = []
+
+    def get_handler(path, params, _timeout):
+        if path == "/health/ready":
+            return _FakeResponse(200, {"status": "ok"})
+        document = documents[params["source_id"]]
+        lookups.append(document.source_id)
+        return _FakeResponse(
+            200,
+            {
+                "items": [
+                    {
+                        "space_id": "space-1",
+                        "source_id": document.source_id,
+                        "external_id": f"{document.document_id}.md",
+                        "status": "active",
+                        "active_version": {
+                            "status": "published",
+                            "content_sha256": document.content_hash,
+                        },
+                    }
+                ]
+            },
+        )
+
+    monkeypatch.setattr(provider, "_client", lambda: _FakeClient(get_handler=get_handler))
+
+    first = provider.health("en-GB")
+    second = provider.health("en-GB")
+    assert first.document_count == 11
+    assert first.statistics_state == "verified"
+    assert second.document_count == 11
+    assert len(lookups) == 11
+
+
 def test_ragcore_provider_health_reports_degraded_status(monkeypatch):
     provider = RAGcoreKnowledgeProvider()
     monkeypatch.setattr(