MobilityOps/backend/app/services/knowledge/ragcore.py

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 <token>` (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,
        )