M44: harden release integrity and assurance

backend/app/services/knowledge/ragcore.py

@@ -8,6 +8,7 @@ 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
 
@@ -137,6 +138,22 @@ class RAGcoreKnowledgeProvider:
         self._settings = get_settings()
         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 _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 = {}
@@ -264,9 +281,12 @@ class RAGcoreKnowledgeProvider:
         if not self._settings.ragcore_space_id:
             return unavailable
 
-        answered = self._ask_via_answers(question, correlation_id)
-        if answered is not None:
-            return answered
+        if self._answers_circuit_is_open():
+            KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", "circuit_open").inc()
+        else:
+            answered = self._ask_via_answers(question, correlation_id)
+            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
@@ -290,9 +310,13 @@ class RAGcoreKnowledgeProvider:
                     },
                 )
                 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:
@@ -311,6 +335,8 @@ class RAGcoreKnowledgeProvider:
             answerability = body.get("answerability", "not_answerable")
             is_grounded = answerability in _GROUNDED_ANSWERABILITY and sources
             evidence_state: EvidenceState = "grounded" if is_grounded else "insufficient"
+            KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", evidence_state).inc()
+            self._close_answers_circuit()
             return GroundedAnswer(
                 answer=body.get("answer", "") if evidence_state == "grounded" else "",
                 evidence_state=evidence_state,
@@ -319,6 +345,8 @@ class RAGcoreKnowledgeProvider:
                 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(
@@ -342,9 +370,11 @@ class RAGcoreKnowledgeProvider:
                     },
                 )
                 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:
@@ -362,8 +392,14 @@ class RAGcoreKnowledgeProvider:
                 for result in results
             ]
         except (TypeError, KeyError, ValueError):
+            KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "malformed").inc()
             return unavailable
 
+        for result in results:
+            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 = [
@@ -379,6 +415,7 @@ class RAGcoreKnowledgeProvider:
         ]
         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",
@@ -396,6 +433,7 @@ class RAGcoreKnowledgeProvider:
             for term in _DOMAIN_CONCEPTS["damage"]
         )
         if not concepts or not sources or ("damage" in concepts and not damage_evidence):
+            KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "insufficient").inc()
             return GroundedAnswer(
                 answer="",
                 evidence_state="insufficient",
@@ -407,6 +445,7 @@ class RAGcoreKnowledgeProvider:
         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",