from __future__ import annotations import hashlib import re import uuid 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.core.observability import KNOWLEDGE_PROVIDER_REQUESTS, KNOWLEDGE_RETRIEVAL_SCORE from app.services.knowledge import EvidenceState, GroundedAnswer, KnowledgeHealth, SourceCard from app.services.knowledge.procedures import ProcedureDocument, iter_procedure_documents _GROUNDED_ANSWERABILITY = {"answerable", "partially_answerable"} _ANSWERABILITY_STATES = _GROUNDED_ANSWERABILITY | { "not_answerable", "conflicting_evidence", } # Mirrors DemoKnowledgeProvider's own extractive template in spirit: a real cited # excerpt wrapped in a fixed sentence, never a generated summary. Used only as a # fallback when RAGcore's own /v1/answers (generation + citation validation) is # unavailable but its retrieval (/v1/search) still returns real, relevant, cited # results -- see ask() below. Unlike the demo corpus's own markdown frontmatter, RAGcore's # `document_version_id` is an opaque UUID, not a human-meaningful version string, so it # is deliberately left out of this sentence (it still appears on the source card itself). _LEAD_ANSWER_TEMPLATE = { "en-GB": 'Per "{title}": {excerpt}', "nl-BE": 'Volgens "{title}": {excerpt}', "fr-BE": "Selon « {title} » : {excerpt}", } _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é"), "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"), } _ANSWERABLE_INTENT_CONCEPTS = frozenset(_DOMAIN_CONCEPTS) - {"vehicle", "customer"} def _normalize_evidence_text(value: str) -> str: return " ".join(re.findall(r"\w+", value.casefold())) 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]: """Collapse duplicate chunks and re-uploaded document versions. RAGcore document/version UUIDs change across uploads, so they are not useful deduplication keys. Human-visible citation identity is the normalized title, section. Chunks from the same unsectioned document collapse into one card; distinct named sections remain independently citable. """ seen: set[tuple[str, str]] = set() unique: list[SourceCard] = [] for source in sources: key = ( source.title.strip().casefold(), source.section.strip().casefold(), ) if key in seen: continue seen.add(key) unique.append(source) if len(unique) == _MAX_SOURCE_CARDS: break 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 ), ) def _retrieval_score(result: object) -> float | None: if not isinstance(result, dict): return 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 owned separately, MobilityOps only ever talks to its documented HTTP API). Authenticates as a service account via `Authorization: Bearer ` (RAGcore's session-cookie auth is for its own browser admin UI only). Any connection error, timeout, non-2xx response, or malformed body degrades to `evidence_state: "unavailable"` rather than raising -- this is the adapter that actually exercises the architecture's reliability boundary: RAGcore failure disables knowledge answers only, never fabricates an answer, never affects the rest of the app. `/v1/answers` (RAGcore's own generation + citation-validation step) is tried first; if it is itself unavailable (non-2xx or unreachable -- as opposed to a real 200 classifying the question as insufficiently answerable), `ask()` falls back to RAGcore's `/v1/search` retrieval, which is a materially different, simpler pipeline stage with no generation step to fail. The fallback answer is always an extractive excerpt RAGcore's own search actually found, wrapped in the same fixed citation template `DemoKnowledgeProvider` uses -- never a fabricated summary. Retrieval is scoped to the stable RAGcore ``source_id`` values owned by Fleet Ops for the requested UI language. Returned internal document/version UUIDs are deliberately not treated as Fleet Ops identifiers: every citation must instead prove the complete managed-source chain (source id, URI, locator, checksum and extractive local text). """ name = "ragcore" def __init__(self) -> None: self._settings = get_settings() self._managed_documents_by_source_id: dict[str, ProcedureDocument] = {} for document in iter_procedure_documents(Path(self._settings.knowledge_dir)): self._managed_documents_by_source_id[document.source_id] = document self._verification_cache: dict[str, tuple[float, int]] = {} self._verification_lock = Lock() self._answers_circuit_lock = Lock() self._answers_circuit_open_until = 0.0 def _managed_source(self, citation: object, language: str) -> SourceCard | None: if not isinstance(citation, dict): return None source_id = citation.get("source_id") if not isinstance(source_id, str): return None document = self._managed_documents_by_source_id.get(str(source_id)) if document is None or document.language != language: return None if citation.get("locator") != f"{document.document_id}.md": return None if citation.get("source_uri") != f"ragcore://source/{document.source_id}": return None for field_name in ("id", "document_id", "document_version_id"): value = citation.get(field_name) if not isinstance(value, str): return None try: parsed = uuid.UUID(value) except (ValueError, TypeError, AttributeError): return None if parsed.int == 0: return None if not isinstance(citation.get("title"), str): return None section = citation.get("section") if section is not None and not isinstance(section, str): return None excerpt = citation.get("excerpt") if not isinstance(excerpt, str) or not excerpt.strip(): return None expected_hash = hashlib.sha256(excerpt.encode("utf-8")).hexdigest() if citation.get("excerpt_sha256") != expected_hash: return None # RAGcore owns chunking, but the cited excerpt must still be extractive evidence # from the authoritative local procedure. Token normalization tolerates Markdown # punctuation/whitespace while rejecting provider text that was never uploaded. normalized_excerpt = _normalize_evidence_text(excerpt) if not normalized_excerpt or normalized_excerpt not in _normalize_evidence_text( document.content ): return None return SourceCard( document_id=document.document_id, title=document.title, version=document.version, section=section or "", excerpt=excerpt, ) def _managed_source_ids(self, language: str) -> list[str]: return [ document.source_id for document in self._managed_documents_by_source_id.values() if document.language == language ] def _answers_circuit_is_open(self) -> bool: with self._answers_circuit_lock: return monotonic() < self._answers_circuit_open_until def _open_answers_circuit(self) -> None: with self._answers_circuit_lock: self._answers_circuit_open_until = monotonic() + max( 0.0, self._settings.ragcore_answers_circuit_breaker_seconds ) def _close_answers_circuit(self) -> None: with self._answers_circuit_lock: self._answers_circuit_open_until = 0.0 def _client(self) -> httpx.Client: headers = {} if self._settings.ragcore_api_token: headers["Authorization"] = f"Bearer {self._settings.ragcore_api_token}" return httpx.Client( base_url=self._settings.ragcore_base_url, headers=headers, timeout=self._settings.ragcore_http_timeout_seconds, ) 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") body = response.json() if not isinstance(body, dict): raise ValueError("health response is not a JSON object") available = response.status_code == 200 and body.get("status") == "ok" detail = ( "RAGcore reachable and ready." 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 exact active published document in the configured space." ) except (httpx.HTTPError, ValueError) as exc: available = False detail = f"RAGcore unavailable: {type(exc).__name__}: {exc}" return KnowledgeHealth( provider=self.name, available=available, detail=detail, tenant=self._settings.ragcore_tenant, workspace=self._settings.ragcore_workspace, collection=self._settings.ragcore_collection, # RAGcore deliberately has no browse/count endpoint. Fleet Ops instead # verifies each managed source through its exact identity lookup and only # counts an active document with a published active version. RAGcore's # content_sha256 describes its canonical parsed artifact, not the uploaded # source bytes, so comparing it with Fleet Ops's source hash would be false. 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=( "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" ) 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="", evidence_state="unavailable", sources=[], provider=self.name, correlation_id=correlation_id, ) if not self._settings.ragcore_space_id: return unavailable if self._answers_circuit_is_open(): KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", "circuit_open").inc() else: answered = self._ask_via_answers(question, correlation_id, language) if answered is not None: return answered # /v1/answers itself is unavailable (non-2xx or unreachable) -- fall back to # real retrieval rather than degrading straight to "unavailable". This never # fabricates an answer to the question: it only ever shows an actually-cited # excerpt RAGcore's own search already found, using the same extractive # citation-wrapper template DemoKnowledgeProvider uses, never RAGcore's # generation step. return self._ask_via_search_fallback(question, correlation_id, language) def _ask_via_answers( self, question: str, correlation_id: str, language: str ) -> GroundedAnswer | None: """Returns None (not a GroundedAnswer) when /v1/answers itself is unavailable, so the caller can fall back to search -- as opposed to a real 200 response classifying the question as insufficiently answerable, which is a genuine, final result, not a reason to fall back.""" try: with self._client() as client: response = client.post( "/v1/answers", json={ "query": question, "requested_space_ids": [self._settings.ragcore_space_id], "filters": {"source_ids": self._managed_source_ids(language)}, }, ) if response.status_code != 200: KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", "non_2xx").inc() self._open_answers_circuit() return None body = response.json() except (httpx.HTTPError, ValueError): KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", "error").inc() self._open_answers_circuit() return None try: if not isinstance(body, dict): raise TypeError("answer response must be an object") # Validate the provider's complete AnswerResponse contract before trusting # generated prose. These IDs and claim bindings are the evidence that RAGcore # ran its deterministic claim/citation validator; one unrelated but otherwise # valid citation must never make arbitrary answer text appear grounded. uuid.UUID(str(body["answer_id"])) uuid.UUID(str(body["retrieval_run_id"])) raw_citations = body.get("citations", []) if not isinstance(raw_citations, list): raise TypeError("citations must be a list") citation_ids: set[uuid.UUID] = set() for citation in raw_citations: if not isinstance(citation, dict): raise TypeError("citation must be an object") citation_ids.add(uuid.UUID(str(citation["id"]))) raw_claims = body["claims"] if not isinstance(raw_claims, list): raise TypeError("claims must be a list") claims_are_bound = bool(raw_claims) answer_text = body.get("answer") for claim in raw_claims: if not isinstance(claim, dict): raise TypeError("claim must be an object") claim_text = claim.get("text") claim_citation_ids = claim.get("citation_ids") if ( not isinstance(claim_text, str) or not claim_text.strip() or not isinstance(answer_text, str) or claim_text.strip() not in answer_text or not isinstance(claim_citation_ids, list) or not claim_citation_ids ): claims_are_bound = False continue try: bound_ids = {uuid.UUID(str(item)) for item in claim_citation_ids} except (TypeError, ValueError): claims_are_bound = False continue if not bound_ids.issubset(citation_ids): claims_are_bound = False mapped_sources = [ source for citation in raw_citations if (source := self._managed_source(citation, language)) is not None ] citations_are_managed = len(mapped_sources) == len(raw_citations) sources = _deduplicate_sources(mapped_sources) answerability = body.get("answerability", "not_answerable") if answerability not in _ANSWERABILITY_STATES: raise TypeError("answerability is invalid") if not isinstance(answer_text, str): raise TypeError("answer must be a string") is_grounded = ( answerability in _GROUNDED_ANSWERABILITY and bool(sources) and citations_are_managed and claims_are_bound and bool(answer_text.strip()) ) evidence_state: EvidenceState = "grounded" if is_grounded else "insufficient" KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", evidence_state).inc() self._close_answers_circuit() return GroundedAnswer( answer=answer_text if evidence_state == "grounded" else "", evidence_state=evidence_state, sources=sources if evidence_state == "grounded" else [], provider=self.name, correlation_id=correlation_id, ) except (TypeError, KeyError, ValueError): KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", "malformed").inc() self._open_answers_circuit() return None def _ask_via_search_fallback( self, question: str, correlation_id: str, language: str ) -> GroundedAnswer: unavailable = GroundedAnswer( answer="", evidence_state="unavailable", sources=[], provider=self.name, correlation_id=correlation_id, ) try: with self._client() as client: response = client.post( "/v1/search", json={ "query": question, "requested_space_ids": [self._settings.ragcore_space_id], "filters": {"source_ids": self._managed_source_ids(language)}, "max_results": 5, }, ) if response.status_code != 200: KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "non_2xx").inc() return unavailable body = response.json() except (httpx.HTTPError, ValueError): KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "error").inc() return unavailable try: if not isinstance(body, dict): raise TypeError("search response must be an object") results = body.get("results", []) if not isinstance(results, list): raise TypeError("results must be a list") sources = [ source for result in results if isinstance(result, dict) and (source := self._managed_source(result.get("citation"), language)) is not None ] except (TypeError, KeyError, ValueError): KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "malformed").inc() return unavailable for result in results: if not isinstance(result, dict): continue score = _retrieval_score(result) if score is not None: KNOWLEDGE_RETRIEVAL_SCORE.observe(score) all_sources = _deduplicate_sources(sources) concepts = _question_concepts(question) qualified_sources = [ source for result in results if isinstance(result, dict) and (_retrieval_score(result) or 0.0) >= self._settings.ragcore_min_search_score and (source := self._managed_source(result.get("citation"), language)) is not None ] sources = _rank_sources_for_concepts(_deduplicate_sources(qualified_sources), concepts) if not all_sources: KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "insufficient").inc() return GroundedAnswer( answer="", evidence_state="insufficient", sources=[], provider=self.name, correlation_id=correlation_id, ) required_concepts = concepts & _ANSWERABLE_INTENT_CONCEPTS evidence_text = " ".join( f"{source.document_id} {source.title} {source.section} {source.excerpt}" for source in sources ).casefold() covered_concepts = { concept for concept in required_concepts if any(term in evidence_text for term in _DOMAIN_CONCEPTS[concept]) } if not required_concepts or not sources or covered_concepts != required_concepts: KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "insufficient").inc() return GroundedAnswer( answer="", evidence_state="insufficient", sources=all_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) KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "grounded").inc() return GroundedAnswer( answer=answer, evidence_state="grounded", sources=sources, provider=self.name, correlation_id=correlation_id, )