412 lines
15 KiB
Python
412 lines
15 KiB
Python
"""Unit: the honesty gate (A8) — boundary, prompts, and suggestion chips.
|
|
|
|
Pure gate logic runs against fake retriever output (``RetrievedChunk``
|
|
rows from a fake retriever) with no Postgres and no network. The
|
|
endpoint-level tests drive ``POST /api/chat`` with the retriever, the DB
|
|
session, and the LLM all faked, so the whole deflection contract
|
|
(prompt → deltas → done event → query_log) is verified without a stack.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api import chat as chat_api
|
|
from app.config import Settings
|
|
from app.main import app as fastapi_app
|
|
from app.models import Document, QueryLog
|
|
from app.rag.retriever import RetrievedChunk, weak_hit_titles
|
|
from app.rag.suggestions import MAX_SUGGESTIONS, derive_suggestions
|
|
|
|
ANSWER = "I haven't done anything like that — try one of these instead!"
|
|
|
|
|
|
def _settings(threshold: float = 0.30) -> Settings:
|
|
return Settings(
|
|
_env_file=None, # pyright: ignore[reportCallIssue]
|
|
relevance_threshold=threshold,
|
|
)
|
|
|
|
|
|
def _doc(title: str, content: str) -> Document:
|
|
return Document(
|
|
id=uuid.uuid4(),
|
|
source="Homelab",
|
|
path=f"{title.lower().replace(' ', '-')}.md",
|
|
full_path="/tmp/doc.md",
|
|
title=title,
|
|
content=content,
|
|
content_hash="0" * 64,
|
|
)
|
|
|
|
|
|
def _chunk(
|
|
doc: Document, score: float, cosine: float | None = None, fts_hit: bool = False
|
|
) -> RetrievedChunk:
|
|
"""Fake candidate: *score* is the fused rank score; *cosine* (defaults to
|
|
*score*) is the vector-similarity gate input."""
|
|
return RetrievedChunk(
|
|
chunk_id=uuid.uuid4(),
|
|
position=0,
|
|
content=doc.content[:32],
|
|
score=score,
|
|
document=doc,
|
|
cosine=score if cosine is None else cosine,
|
|
fts_hit=fts_hit,
|
|
)
|
|
|
|
|
|
# ---------- gate boundary (fake retriever rows, no LLM) ----------
|
|
|
|
|
|
def test_gate_boundary_score_at_threshold_answers() -> None:
|
|
"""Score exactly at the threshold ⇒ HIGH (the gate is strict <)."""
|
|
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
|
|
plan = chat_api.plan_turn([_chunk(doc, 0.30)], _settings(threshold=0.30))
|
|
assert plan.deflected is False
|
|
assert plan.top_score == pytest.approx(0.30)
|
|
assert "<relevance>HIGH</relevance>" in plan.system_prompt
|
|
assert "DEFLECT_MODE" not in plan.system_prompt
|
|
assert "TALOS_DOC_CONTENT" in plan.system_prompt
|
|
assert plan.suggestions == []
|
|
|
|
|
|
def test_gate_boundary_just_below_threshold_deflects() -> None:
|
|
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
|
|
plan = chat_api.plan_turn([_chunk(doc, 0.2999)], _settings(threshold=0.30))
|
|
assert plan.deflected is True
|
|
assert plan.top_score == pytest.approx(0.2999)
|
|
assert "<relevance>LOW</relevance>" in plan.system_prompt
|
|
assert "DEFLECT_MODE" in plan.system_prompt
|
|
# Titles only: the full document content must never reach the LLM.
|
|
assert "TALOS_DOC_CONTENT" not in plan.system_prompt
|
|
assert "Kubernetes Homelab Cluster" in plan.system_prompt
|
|
|
|
|
|
def test_gate_is_env_tunable_via_settings() -> None:
|
|
doc = _doc("Backup Strategy", "BACKUP_DOC_CONTENT")
|
|
hits = [_chunk(doc, 0.30)]
|
|
assert chat_api.plan_turn(hits, _settings(threshold=0.35)).deflected is True
|
|
assert chat_api.plan_turn(hits, _settings(threshold=0.25)).deflected is False
|
|
|
|
|
|
# ---------- hybrid gate matrix (A8, revised: cosine AND fts) ----------
|
|
|
|
|
|
def test_gate_weak_cosine_with_fts_hit_still_answers() -> None:
|
|
"""cosine < threshold but a lexical hit ⇒ HIGH — the FTS-OR branch.
|
|
This is the name-your-tool case: "kafkabridge" grounds despite weak
|
|
vector overlap."""
|
|
doc = _doc("Static DNS", "DNS_DOC_CONTENT")
|
|
plan = chat_api.plan_turn(
|
|
[_chunk(doc, 0.02, cosine=0.10, fts_hit=True)], _settings(threshold=0.30)
|
|
)
|
|
assert plan.deflected is False
|
|
assert plan.top_score == pytest.approx(0.10) # gate input is the cosine
|
|
assert plan.fts_hits == 1
|
|
assert "DNS_DOC_CONTENT" in plan.system_prompt
|
|
assert plan.suggestions == []
|
|
|
|
|
|
def test_gate_weak_cosine_zero_fts_deflects() -> None:
|
|
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
|
|
plan = chat_api.plan_turn([_chunk(doc, 0.02, cosine=0.10)], _settings(threshold=0.30))
|
|
assert plan.deflected is True
|
|
assert plan.top_score == pytest.approx(0.10)
|
|
assert plan.fts_hits == 0
|
|
|
|
|
|
def test_gate_strong_cosine_without_fts_answers() -> None:
|
|
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
|
|
plan = chat_api.plan_turn([_chunk(doc, 0.90, cosine=0.90)], _settings(threshold=0.30))
|
|
assert plan.deflected is False
|
|
assert plan.fts_hits == 0
|
|
|
|
|
|
def test_gate_fts_hits_counts_all_lexical_candidates() -> None:
|
|
a = _doc("Alpha", "ALPHA_CONTENT")
|
|
b = _doc("Beta", "BETA_CONTENT")
|
|
chunks = [
|
|
_chunk(a, 0.03, cosine=0.05, fts_hit=True),
|
|
_chunk(a, 0.02, cosine=0.04, fts_hit=True), # same doc, second chunk
|
|
_chunk(b, 0.01, cosine=0.03),
|
|
]
|
|
plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
|
|
assert plan.deflected is False
|
|
assert plan.fts_hits == 2 # per chunk, not per doc
|
|
|
|
|
|
def test_gate_lexical_only_chunk_does_not_inflate_cosine() -> None:
|
|
"""top_score stays the best *vector* cosine even when a lexical-only
|
|
chunk (cosine 0.0 by construction) carries the highest fused score."""
|
|
a = _doc("Alpha", "ALPHA_CONTENT")
|
|
b = _doc("Beta", "BETA_CONTENT")
|
|
chunks = [
|
|
_chunk(a, 0.50, cosine=0.55), # vector rank 1
|
|
_chunk(b, 0.90, cosine=0.0, fts_hit=True), # lexical rank 1 wins the ranking
|
|
]
|
|
plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
|
|
assert plan.top_score == pytest.approx(0.55)
|
|
assert plan.deflected is False # 0.55 >= 0.30 anyway
|
|
# ranking follows the fused score: Beta's doc is the top source
|
|
assert plan.docs[0].title == "Beta"
|
|
|
|
|
|
def test_gate_zero_chunks_deflects_with_fallback_chips() -> None:
|
|
plan = chat_api.plan_turn([], _settings())
|
|
assert plan.deflected is True
|
|
assert plan.top_score == 0.0
|
|
assert "nothing close at all" in plan.system_prompt
|
|
# No weak hits ⇒ onboarding fallback fills the chips.
|
|
assert 2 <= len(plan.suggestions) <= MAX_SUGGESTIONS
|
|
|
|
|
|
# ---------- prompt content (LOW vs HIGH) ----------
|
|
|
|
|
|
def test_low_prompt_has_titles_only_no_content() -> None:
|
|
a = _doc("Kubernetes Homelab Cluster", "ALPHA_DOC_CONTENT")
|
|
b = _doc("Backup Strategy", "BETA_DOC_CONTENT")
|
|
plan = chat_api.plan_turn([_chunk(b, 0.10), _chunk(a, 0.20)], _settings())
|
|
prompt = plan.system_prompt
|
|
assert "<relevance>LOW</relevance>" in prompt
|
|
assert "DEFLECT_MODE" in prompt
|
|
assert "HONESTY GATE" in prompt # the LOW rule is what the model follows
|
|
assert "- Kubernetes Homelab Cluster" in prompt
|
|
assert "- Backup Strategy" in prompt
|
|
assert "ALPHA_DOC_CONTENT" not in prompt
|
|
assert "BETA_DOC_CONTENT" not in prompt
|
|
assert "<documents>" not in prompt
|
|
|
|
|
|
def test_high_path_unaffected() -> None:
|
|
a = _doc("Kubernetes Homelab Cluster", "ALPHA_DOC_CONTENT")
|
|
b = _doc("Backup Strategy", "BETA_DOC_CONTENT")
|
|
plan = chat_api.plan_turn([_chunk(a, 0.90), _chunk(b, 0.40)], _settings())
|
|
assert plan.deflected is False
|
|
assert plan.suggestions == []
|
|
assert "<relevance>HIGH</relevance>" in plan.system_prompt
|
|
assert "DEFLECT_MODE" not in plan.system_prompt
|
|
assert "ALPHA_DOC_CONTENT" in plan.system_prompt
|
|
assert "BETA_DOC_CONTENT" in plan.system_prompt
|
|
assert [d.title for d in plan.docs] == ["Kubernetes Homelab Cluster", "Backup Strategy"]
|
|
|
|
|
|
# ---------- weak_hit_titles (fake retriever mapping) ----------
|
|
|
|
|
|
def test_weak_hit_titles_dedupe_and_rank_by_best_score() -> None:
|
|
a = _doc("Kubernetes Homelab Cluster", "AAA")
|
|
b = _doc("Backup Strategy", "BBB")
|
|
chunks = [_chunk(b, 0.5), _chunk(a, 0.2), _chunk(a, 0.9)]
|
|
assert weak_hit_titles(chunks) == ["Kubernetes Homelab Cluster", "Backup Strategy"]
|
|
|
|
|
|
# ---------- suggestions derivation ----------
|
|
|
|
|
|
def test_suggestions_derived_from_titles_in_order() -> None:
|
|
got = derive_suggestions(
|
|
["Kubernetes Homelab Cluster", "Backup Strategy", "Deploying a New Service"],
|
|
fallback=["should not appear"],
|
|
)
|
|
assert len(got) == 3
|
|
assert all(s.strip() for s in got)
|
|
assert "Kubernetes Homelab Cluster" in got[0]
|
|
assert "Backup Strategy" in got[1]
|
|
assert "Deploying a New Service" in got[2]
|
|
|
|
|
|
def test_suggestions_capped_at_three() -> None:
|
|
got = derive_suggestions([f"Title {i}" for i in range(6)], fallback=["F"])
|
|
assert len(got) == MAX_SUGGESTIONS == 3
|
|
|
|
|
|
def test_suggestions_top_up_from_fallback_when_titles_thin() -> None:
|
|
got = derive_suggestions(
|
|
["Backup Strategy"],
|
|
fallback=["How is my Kubernetes cluster set up?", "What's my backup strategy?"],
|
|
)
|
|
assert len(got) == 3
|
|
assert got[0] == "What's in your notes about Backup Strategy?"
|
|
assert got[1] == "How is my Kubernetes cluster set up?"
|
|
|
|
|
|
def test_suggestions_dedupes_and_ignores_blank() -> None:
|
|
got = derive_suggestions(
|
|
["Backup Strategy", "backup strategy", " "],
|
|
fallback=["What's my backup strategy?", " "],
|
|
)
|
|
# "backup strategy" is a case-insensitive dup; blank title/fallback are
|
|
# skipped — including ones that only look blank after formatting. Only
|
|
# two valid items remain, and the list never pads with junk.
|
|
assert got == [
|
|
"What's in your notes about Backup Strategy?",
|
|
"What's my backup strategy?",
|
|
]
|
|
assert all("about ?" not in s and s == s.strip() for s in got)
|
|
|
|
|
|
def test_suggestions_empty_input_yields_fallback_only() -> None:
|
|
assert derive_suggestions([], fallback=[]) == []
|
|
got = derive_suggestions([], fallback=["One?", "Two?"])
|
|
assert got == ["One?", "Two?"]
|
|
|
|
|
|
# ---------- endpoint-level gate (fake retriever + fake LLM + fake session) ----------
|
|
|
|
|
|
class _CannedLLM:
|
|
"""Records the messages it is given; streams a canned answer."""
|
|
|
|
def __init__(self, answer: str = ANSWER) -> None:
|
|
self.settings = Settings(_env_file=None) # pyright: ignore[reportCallIssue]
|
|
self.embed_batches = 0
|
|
self.answer = answer
|
|
self.seen: list[list[dict[str, str]]] = []
|
|
|
|
async def embed_one(self, _text: str) -> list[float]:
|
|
return [0.0] * 768
|
|
|
|
async def chat_stream(self, messages: list[dict[str, str]]):
|
|
self.seen.append(messages)
|
|
for i in range(0, len(self.answer), 12):
|
|
yield self.answer[i : i + 12]
|
|
|
|
|
|
class _FakeSteeringResult:
|
|
"""Empty steering-note result (no stored notes in these unit tests)."""
|
|
|
|
def all(self) -> list[Any]:
|
|
return []
|
|
|
|
|
|
class _FakeSession:
|
|
"""Stands in for the DB session: records the QueryLog row it is given.
|
|
|
|
``scalars`` always yields no steering notes (phase 15) so the chat
|
|
turn's ``load_steering_notes`` call stays a no-op here.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.added: list[Any] = []
|
|
self.commits = 0
|
|
|
|
def add(self, obj: Any) -> None:
|
|
self.added.append(obj)
|
|
|
|
def commit(self) -> None:
|
|
self.commits += 1
|
|
|
|
def scalars(self, _stmt: Any) -> _FakeSteeringResult:
|
|
return _FakeSteeringResult()
|
|
|
|
|
|
@pytest.fixture()
|
|
def gate_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[_FakeSession, _CannedLLM]]:
|
|
"""``POST /api/chat`` with retriever, session, and LLM all faked."""
|
|
monkeypatch.setattr(chat_api, "db_available", lambda: True)
|
|
session = _FakeSession()
|
|
llm = _CannedLLM()
|
|
monkeypatch.setitem(fastapi_app.dependency_overrides, chat_api.get_db, lambda: session)
|
|
monkeypatch.setitem(fastapi_app.dependency_overrides, chat_api.get_llm, lambda: llm)
|
|
# These tests assert against a specific gate threshold; keep it stable
|
|
# regardless of the production default (0.62) or any .env.
|
|
monkeypatch.setattr(
|
|
chat_api,
|
|
"get_settings",
|
|
lambda: Settings(_env_file=None, relevance_threshold=0.30), # pyright: ignore[reportCallIssue]
|
|
)
|
|
yield session, llm
|
|
fastapi_app.dependency_overrides.clear()
|
|
|
|
|
|
def _ask(client: TestClient, message: str) -> list[dict[str, Any]]:
|
|
with client.stream("POST", "/api/chat", json={"message": message}) as r:
|
|
assert r.status_code == 200
|
|
frames: list[dict[str, Any]] = []
|
|
buf = ""
|
|
for part in r.iter_text():
|
|
buf += part
|
|
while "\n\n" in buf:
|
|
frame, buf = buf.split("\n\n", 1)
|
|
frame = frame.strip()
|
|
if frame.startswith("data:"):
|
|
frames.append(json.loads(frame.removeprefix("data:").strip()))
|
|
assert buf.strip() == ""
|
|
return frames
|
|
|
|
|
|
def _fake_retriever(chunks: list[RetrievedChunk]) -> Any:
|
|
def retrieve(_db: Any, _question: str, _vec: list[float]) -> list[RetrievedChunk]:
|
|
return chunks
|
|
|
|
return retrieve
|
|
|
|
|
|
def test_endpoint_just_below_threshold_deflects(
|
|
client: TestClient,
|
|
gate_env: tuple[_FakeSession, _CannedLLM],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session, llm = gate_env
|
|
doc = _doc("Deploying a New Service", "DOC_CONTENT_NEVER_SENT")
|
|
monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.2999)]))
|
|
|
|
frames = _ask(client, "How do I bake sourdough bread?")
|
|
|
|
deltas = [f for f in frames if f["type"] == "delta"]
|
|
assert "".join(d["text"] for d in deltas) == ANSWER # the LLM was still called
|
|
done = frames[-1]
|
|
assert done["type"] == "done"
|
|
assert done["deflected"] is True
|
|
assert 2 <= len(done["suggestions"]) <= MAX_SUGGESTIONS # title chip + fallback
|
|
assert all(s.strip() for s in done["suggestions"])
|
|
assert any("Deploying a New Service" in s for s in done["suggestions"])
|
|
|
|
# The LLM saw the LOW prompt: DEFLECT_MODE + titles, never doc content.
|
|
(system, user) = llm.seen[0][0], llm.seen[0][1]
|
|
assert user["content"] == "How do I bake sourdough bread?"
|
|
assert "DEFLECT_MODE" in system["content"]
|
|
assert "DOC_CONTENT_NEVER_SENT" not in system["content"]
|
|
|
|
# Durable record: deflected + the weak score.
|
|
(row,) = session.added
|
|
assert isinstance(row, QueryLog)
|
|
assert row.deflected is True
|
|
assert row.top_score == pytest.approx(0.2999)
|
|
assert session.commits == 1
|
|
|
|
|
|
def test_endpoint_score_at_threshold_answers(
|
|
client: TestClient,
|
|
gate_env: tuple[_FakeSession, _CannedLLM],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session, llm = gate_env
|
|
doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_SENT")
|
|
monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.30)]))
|
|
|
|
frames = _ask(client, "How is my Kubernetes cluster set up?")
|
|
|
|
done = frames[-1]
|
|
assert done["type"] == "done"
|
|
assert done["deflected"] is False
|
|
assert done["suggestions"] == []
|
|
assert done["sources"] and done["sources"][0]["title"] == "Kubernetes Homelab Cluster"
|
|
|
|
(system, _user) = llm.seen[0][0], llm.seen[0][1]
|
|
assert "<relevance>HIGH</relevance>" in system["content"]
|
|
assert "DEFLECT_MODE" not in system["content"]
|
|
assert "TALOS_DOC_SENT" in system["content"]
|
|
|
|
(row,) = session.added
|
|
assert isinstance(row, QueryLog)
|
|
assert row.deflected is False
|
|
assert row.top_score == pytest.approx(0.30)
|