NuklearRabbitandClaude Sonnet 5
e5d8466266
The previous adapter targeted an endpoint shape RAGcore never actually exposed. health() now checks /health/ready and ask() posts to the real POST /v1/answers with Bearer auth and requested_space_ids, matching RAGcore's actual contract after this session's Bearer-auth and search/answer wiring work. Adds RAGCORE_SPACE_ID config/env plumbing (a question is meaningless without a knowledge space to scope it to) and 12 new adapter tests covering degradation paths: missing space id, connection errors, non-200 responses, malformed responses, not-answerable, and answerable-without-citations all fail closed to "insufficient evidence" rather than fabricating an answer. KNOWLEDGE_PROVIDER stays "demo" in production for now -- switching requires RAGcore's own search/answer application to actually be deployed and live-verified, tracked separately in PROJECT_STATE.md. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
348 lines
12 KiB
Python
348 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from app.core.config import get_settings
|
|
from app.services.knowledge.demo import DemoKnowledgeProvider
|
|
from app.services.knowledge.ragcore import RAGcoreKnowledgeProvider
|
|
|
|
|
|
def test_brief_exact_damage_question_in_all_three_languages():
|
|
# The exact validation questions from docs/fleet-ops-correction/current-gap-audit.md
|
|
# -- each must ground on the damage procedure as its *primary* (top-ranked) source,
|
|
# not merely appear somewhere in the top-3, and the source/version/section/excerpt
|
|
# must all come from that same-language document (never an English fallback).
|
|
provider = DemoKnowledgeProvider()
|
|
cases = {
|
|
"nl-BE": "Wat moet ik doen wanneer een voertuig beschadigd terugkomt?",
|
|
"en-GB": "What should I do when a vehicle returns with damage?",
|
|
"fr-BE": "Que dois-je faire lorsqu'un véhicule revient endommagé ?",
|
|
}
|
|
for language, question in cases.items():
|
|
answer = provider.ask(question, f"test-brief-{language}", language)
|
|
assert answer.evidence_state == "grounded", language
|
|
assert answer.sources, language
|
|
assert answer.sources[0].document_id == "damage-procedure", (
|
|
f"{language}: expected the damage procedure as the primary source, "
|
|
f"got {answer.sources[0].document_id!r}"
|
|
)
|
|
assert answer.answer
|
|
assert answer.sources[0].excerpt
|
|
|
|
|
|
def test_knowledge_procedures_never_mention_mobilityops_or_poc():
|
|
# Section 2 of docs/fleet-ops-correction/current-gap-audit.md: the visible brand
|
|
# name is exactly "Fleet Ops", and "PoC" must never appear in visible content --
|
|
# including the demo knowledge base, not just the frontend.
|
|
procedures_dir = Path(get_settings().knowledge_dir)
|
|
offenders = []
|
|
for path in sorted(procedures_dir.glob("*/*.md")):
|
|
text = path.read_text(encoding="utf-8")
|
|
if "MobilityOps" in text or re.search(r"\bPoC\b", text):
|
|
offenders.append(str(path))
|
|
assert offenders == []
|
|
|
|
|
|
def test_s6_damage_question_is_grounded_with_expected_sources():
|
|
provider = DemoKnowledgeProvider()
|
|
answer = provider.ask(
|
|
"What must I do when a vehicle returns with damage?", "test-correlation-1"
|
|
)
|
|
assert answer.evidence_state == "grounded"
|
|
document_ids = {s.document_id for s in answer.sources}
|
|
assert "damage-procedure" in document_ids
|
|
assert "vehicle-return-procedure" in document_ids
|
|
assert answer.answer # never empty for a grounded answer
|
|
for source in answer.sources:
|
|
assert source.excerpt
|
|
|
|
|
|
def test_unrelated_question_is_insufficient():
|
|
provider = DemoKnowledgeProvider()
|
|
answer = provider.ask("What is the capital of France?", "test-correlation-2")
|
|
assert answer.evidence_state == "insufficient"
|
|
assert "France" not in answer.answer # never fabricates an answer beyond the procedures
|
|
|
|
|
|
def test_demo_provider_health_reports_document_count():
|
|
provider = DemoKnowledgeProvider()
|
|
health = provider.health()
|
|
assert health.provider == "demo"
|
|
assert health.available is True
|
|
assert health.document_count == 11
|
|
|
|
|
|
def test_demo_provider_health_reports_document_count_per_language():
|
|
provider = DemoKnowledgeProvider()
|
|
for language in ("nl-BE", "en-GB", "fr-BE"):
|
|
assert provider.health(language).document_count == 11
|
|
|
|
|
|
def test_demo_provider_grounds_damage_question_in_dutch():
|
|
provider = DemoKnowledgeProvider()
|
|
answer = provider.ask(
|
|
"Wat moet ik doen als een voertuig terugkomt met schade?",
|
|
"test-correlation-nl",
|
|
"nl-BE",
|
|
)
|
|
assert answer.evidence_state == "grounded"
|
|
document_ids = {s.document_id for s in answer.sources}
|
|
assert "damage-procedure" in document_ids
|
|
|
|
|
|
def test_demo_provider_grounds_damage_question_in_french():
|
|
provider = DemoKnowledgeProvider()
|
|
answer = provider.ask(
|
|
"Que dois-je faire quand un véhicule revient avec des dommages ?",
|
|
"test-correlation-fr",
|
|
"fr-BE",
|
|
)
|
|
assert answer.evidence_state == "grounded"
|
|
document_ids = {s.document_id for s in answer.sources}
|
|
assert "damage-procedure" in document_ids
|
|
|
|
|
|
def test_demo_provider_insufficient_evidence_message_is_localized():
|
|
provider = DemoKnowledgeProvider()
|
|
nl_answer = provider.ask(
|
|
"Wat is de hoofdstad van Frankrijk?", "test-correlation-nl-2", "nl-BE"
|
|
)
|
|
fr_answer = provider.ask(
|
|
"Quelle est la capitale de la France ?", "test-correlation-fr-2", "fr-BE"
|
|
)
|
|
assert nl_answer.evidence_state == "insufficient"
|
|
assert fr_answer.evidence_state == "insufficient"
|
|
assert nl_answer.answer != fr_answer.answer
|
|
assert "France" not in nl_answer.answer
|
|
assert "France" not in fr_answer.answer
|
|
|
|
|
|
def test_ask_question_endpoint_grounded(ops_client):
|
|
response = ops_client.post(
|
|
"/api/v1/knowledge/questions",
|
|
json={"question": "What must I do when a vehicle returns with damage?"},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["evidence_state"] == "grounded"
|
|
assert body["provider"] == "demo"
|
|
assert len(body["sources"]) > 0
|
|
|
|
|
|
def test_ask_question_requires_authentication(client):
|
|
response = client.post("/api/v1/knowledge/questions", json={"question": "Anything?"})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_ask_question_is_audited_without_leaking_full_text(ops_client):
|
|
ops_client.post(
|
|
"/api/v1/knowledge/questions",
|
|
json={"question": "What must I do when a vehicle returns with damage?"},
|
|
)
|
|
events = ops_client.get(
|
|
"/api/v1/audit", params={"action": "knowledge_question_asked"}
|
|
).json()
|
|
assert len(events) >= 1
|
|
metadata = events[0]["metadata"]
|
|
assert "evidence_state" in metadata
|
|
assert "source_ids" in metadata
|
|
assert "question" not in metadata
|
|
|
|
|
|
def test_knowledge_status_endpoint(ops_client):
|
|
response = ops_client.get("/api/v1/knowledge/status")
|
|
assert response.status_code == 200
|
|
assert response.json()["provider"] == "demo"
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code: int, body: dict):
|
|
self.status_code = status_code
|
|
self._body = body
|
|
|
|
def json(self) -> dict:
|
|
return self._body
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, get_response=None, post_response=None, raise_on=None):
|
|
self._get_response = get_response
|
|
self._post_response = post_response
|
|
self._raise_on = raise_on
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def get(self, path):
|
|
if self._raise_on == "get":
|
|
raise httpx.ConnectError("no ragcore in this environment")
|
|
return self._get_response
|
|
|
|
def post(self, path, json=None):
|
|
if self._raise_on == "post":
|
|
raise httpx.ConnectError("no ragcore in this environment")
|
|
return self._post_response
|
|
|
|
|
|
def test_ragcore_provider_degrades_to_unavailable(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
|
|
monkeypatch.setattr(provider, "_client", lambda: _FakeClient(raise_on="post"))
|
|
answer = provider.ask("Anything?", "test-correlation-3")
|
|
assert answer.evidence_state == "unavailable"
|
|
assert answer.sources == []
|
|
|
|
|
|
def test_ragcore_provider_without_configured_space_is_unavailable_without_a_network_call(
|
|
monkeypatch,
|
|
):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "")
|
|
|
|
def fail_if_called():
|
|
raise AssertionError("should not call RAGcore without a configured space id")
|
|
|
|
monkeypatch.setattr(provider, "_client", fail_if_called)
|
|
answer = provider.ask("Anything?", "test-correlation-no-space")
|
|
assert answer.evidence_state == "unavailable"
|
|
|
|
|
|
def test_ragcore_provider_health_reports_ready_status(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(get_response=_FakeResponse(200, {"status": "ok"})),
|
|
)
|
|
health = provider.health()
|
|
assert health.provider == "ragcore"
|
|
assert health.available is True
|
|
|
|
|
|
def test_ragcore_provider_health_reports_degraded_status(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(get_response=_FakeResponse(200, {"status": "degraded"})),
|
|
)
|
|
health = provider.health()
|
|
assert health.available is False
|
|
|
|
|
|
def test_ragcore_provider_health_degrades_on_connection_error(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider, "_client", lambda: _FakeClient(raise_on="get"))
|
|
health = provider.health()
|
|
assert health.available is False
|
|
assert "unavailable" in health.detail.lower()
|
|
|
|
|
|
def _answers_body(**overrides) -> dict:
|
|
body = {
|
|
"answer": "Report damage and route the vehicle to maintenance.",
|
|
"answerability": "answerable",
|
|
"citations": [
|
|
{
|
|
"id": "cite-1",
|
|
"document_id": "doc-1",
|
|
"document_version_id": "version-1",
|
|
"title": "Damage handling procedure",
|
|
"section": "Detection",
|
|
"excerpt": "Inspect the vehicle for visible damage.",
|
|
}
|
|
],
|
|
}
|
|
body.update(overrides)
|
|
return body
|
|
|
|
|
|
def test_ragcore_provider_grounded_answer_maps_citations_to_sources(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(post_response=_FakeResponse(200, _answers_body())),
|
|
)
|
|
answer = provider.ask("What must I do about damage?", "test-correlation-grounded")
|
|
assert answer.evidence_state == "grounded"
|
|
assert answer.answer
|
|
assert len(answer.sources) == 1
|
|
source = answer.sources[0]
|
|
assert source.document_id == "doc-1"
|
|
assert source.title == "Damage handling procedure"
|
|
assert source.version == "version-1"
|
|
assert source.section == "Detection"
|
|
assert source.excerpt
|
|
|
|
|
|
def test_ragcore_provider_not_answerable_is_insufficient_and_never_fabricates(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(
|
|
post_response=_FakeResponse(
|
|
200,
|
|
_answers_body(
|
|
answer="This should never be shown.",
|
|
answerability="not_answerable",
|
|
citations=[],
|
|
),
|
|
)
|
|
),
|
|
)
|
|
answer = provider.ask("Unrelated question?", "test-correlation-insufficient")
|
|
assert answer.evidence_state == "insufficient"
|
|
assert answer.answer == ""
|
|
assert answer.sources == []
|
|
|
|
|
|
def test_ragcore_provider_answerable_without_citations_is_insufficient(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(
|
|
post_response=_FakeResponse(
|
|
200, _answers_body(answerability="answerable", citations=[])
|
|
)
|
|
),
|
|
)
|
|
answer = provider.ask("What must I do about damage?", "test-correlation-no-citations")
|
|
assert answer.evidence_state == "insufficient"
|
|
assert answer.sources == []
|
|
|
|
|
|
def test_ragcore_provider_non_200_response_is_unavailable(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(post_response=_FakeResponse(401, {"code": "AUTHENTICATION_REQUIRED"})),
|
|
)
|
|
answer = provider.ask("Anything?", "test-correlation-401")
|
|
assert answer.evidence_state == "unavailable"
|
|
|
|
|
|
def test_ragcore_provider_malformed_response_is_unavailable(monkeypatch):
|
|
provider = RAGcoreKnowledgeProvider()
|
|
monkeypatch.setattr(provider._settings, "ragcore_space_id", "space-1")
|
|
monkeypatch.setattr(
|
|
provider,
|
|
"_client",
|
|
lambda: _FakeClient(post_response=_FakeResponse(200, {"citations": "not-a-list"})),
|
|
)
|
|
answer = provider.ask("Anything?", "test-correlation-malformed")
|
|
assert answer.evidence_state == "unavailable"
|