diff --git a/.agent/phases/todo/02_story_import_documents.md b/.agent/phases/complete/02_story_import_documents.md
similarity index 100%
rename from .agent/phases/todo/02_story_import_documents.md
rename to .agent/phases/complete/02_story_import_documents.md
diff --git a/.agent/phases/todo/03_story_chat_rag.md b/.agent/phases/complete/03_story_chat_rag.md
similarity index 100%
rename from .agent/phases/todo/03_story_chat_rag.md
rename to .agent/phases/complete/03_story_chat_rag.md
diff --git a/README.md b/README.md
index 35a0bf1..eeff69b 100644
--- a/README.md
+++ b/README.md
@@ -210,9 +210,23 @@ served locally (no CDN), `BOR_ENVIRONMENT=production`.
`uv run python -m scripts.llm_probe`, update `BOR_EMBEDDING_DIM`, then
drop + recreate the chunks table (new migration or manual `TRUNCATE
chunks, documents`).
+- **Honest deflection (the amber “I haven't done anything like that”
+ bubble)** — every question passes the honesty gate: when the best
+ cosine similarity is below `BOR_RELEVANCE_THRESHOLD` (default `0.30`),
+ Brain switches to deflection mode instead of guessing. The LLM prompt
+ then carries weak-hit *titles only* (no document content), the reply
+ opens with “I haven't done anything like that”, the bubble renders
+ amber with “Maybe try” chips derived from the closest indexed titles,
+ the SSE `done` event carries `deflected: true` + `suggestions[]`, and
+ the `query_log` row records `deflected=true` + the weak `top_score`.
+ This is a feature, not a bug — the KB simply has no notes that close;
+ the chips always point at topics Brain really covers.
- **Answers deflect too often / too rarely** — tune
`BOR_RELEVANCE_THRESHOLD` (lower = answers more, higher = more honest
- deflection). Check `query_log` for the actual scores:
+ deflection): `0.0` ⇒ every question gets answered, even unknown topics
+ (expect confident-sounding guesses); `1.0` ⇒ everything deflects
+ (nothing but a perfect 1.0 score counts as relevant). After changing
+ it, check the real scores:
`psql … -c 'SELECT question, top_score, deflected FROM query_log ORDER BY created_at DESC LIMIT 20'`
- **KB offline banner in the chat** — Postgres isn't running:
`podman compose up -d db`.
diff --git a/app/api/chat.py b/app/api/chat.py
index c05f019..f61eff3 100644
--- a/app/api/chat.py
+++ b/app/api/chat.py
@@ -1,33 +1,39 @@
"""POST /api/chat — a RAG chat turn streamed over SSE (PLAN §3/§4).
Flow (LOCKED A7/A15): embed the question → pgvector cosine top-K chunks →
-distinct parent documents (full text, capped) → locked persona prompt
-(PLAN §6) → ``turbo`` streamed as ``delta`` events → final ``done`` event
-(``deflected``, ``sources``, ``suggestions``) + ``query_log`` row + the
-per-turn log line (PLAN §9). Mid-stream failures become a structured
-``error`` event; a pre-stream DB outage is a plain 503 JSON.
+the **honesty gate** (A8: best score < ``BOR_RELEVANCE_THRESHOLD`` ⇒
+deflection) → locked persona prompt (PLAN §6) → ``turbo`` streamed as
+``delta`` events → final ``done`` event (``deflected``, ``sources``,
+``suggestions``) + ``query_log`` row + the per-turn log line (PLAN §9).
+Mid-stream failures become a structured ``error`` event; a pre-stream DB
+outage is a plain 503 JSON.
-The honesty gate (LOW relevance → deflection) lands in phase 04; every
-turn in this phase is grounded (``deflected=false``).
+Honesty gate: a weak retrieval (score strictly below the threshold — or
+an empty KB) flips the turn to deflection mode: the LOW prompt carries
+weak-hit *titles only* (never document content) plus deterministic
+"Maybe try" chips, and the ``done`` event / ``query_log`` row record
+``deflected=true`` with the weak score.
"""
from __future__ import annotations
import json
import logging
import time
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Sequence
+from dataclasses import dataclass
from typing import Any
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse, StreamingResponse
from sqlalchemy.orm import Session
-from app.config import get_settings
+from app.config import Settings, get_settings
from app.db import db_available, get_db
-from app.models import QueryLog
+from app.models import Document, QueryLog
from app.rag.llm import EmbeddingError, LLMClient, LLMError
-from app.rag.prompts import build_high_prompt
-from app.rag.retriever import retrieve, select_documents
+from app.rag.prompts import build_deflect_prompt, build_high_prompt
+from app.rag.retriever import RetrievedChunk, retrieve, select_documents, weak_hit_titles
+from app.rag.suggestions import derive_suggestions
from app.schemas import ChatDoneEvent, ChatRequest, SourceRef
logger = logging.getLogger("app.chat")
@@ -52,6 +58,44 @@ def sse_event(payload: dict[str, Any]) -> str:
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+@dataclass
+class TurnPlan:
+ """What one chat turn sends to the LLM and reports on ``done``."""
+
+ top_score: float
+ deflected: bool
+ system_prompt: str
+ docs: list[Document] # cited sources (weak hits when deflected)
+ suggestions: list[str] # "Maybe try" chips (deflected turns only)
+
+
+def plan_turn(chunks: Sequence[RetrievedChunk], settings: Settings) -> TurnPlan:
+ """Apply the honesty gate (A8) and assemble prompt + context for a turn.
+
+ * ``top_score >= threshold`` → grounded: HIGH prompt with the full
+ top-N documents, no suggestions. A score exactly at the threshold
+ is an answer — the gate is strict (``score < threshold``).
+ * ``top_score < threshold`` (or no hits at all) → deflected: LOW
+ prompt (``DEFLECT_MODE``) with weak-hit titles only — never document
+ content — plus deterministic alternative-question chips derived
+ from those titles.
+ """
+ top_score = chunks[0].score if chunks else 0.0
+ if top_score >= settings.relevance_threshold:
+ docs = select_documents(
+ chunks, n=settings.top_n_docs, max_chars=settings.max_context_chars
+ )
+ return TurnPlan(top_score, False, build_high_prompt(docs), docs, [])
+ titles = weak_hit_titles(chunks)
+ return TurnPlan(
+ top_score,
+ True,
+ build_deflect_prompt(titles),
+ select_documents(chunks, n=settings.top_n_docs, max_chars=settings.max_context_chars),
+ derive_suggestions(titles, settings.suggestions),
+ )
+
+
@router.post("/chat")
async def chat(
request: ChatRequest,
@@ -88,10 +132,12 @@ async def chat(
return
embed_ms = int((time.monotonic() - t0) * 1000)
- # 2. Retrieve top-K chunks → top-N full parent documents.
+ # 2. Retrieve top-K chunks, then the honesty gate (A8) picks the
+ # HIGH (grounded) or LOW (deflected) prompt + context.
+ settings = get_settings()
try:
chunks = retrieve(db, question_vec)
- docs = select_documents(chunks)
+ plan = plan_turn(chunks, settings)
except Exception: # noqa: BLE001 — DB failure mid-turn
logger.exception("chat: retrieval failed question=%r", request.message)
yield sse_event(
@@ -101,15 +147,13 @@ async def chat(
}
)
return
-
- top_score = chunks[0].score if chunks else 0.0
- source_paths = [f"{d.source}/{d.path}" for d in docs]
+ source_paths = [f"{d.source}/{d.path}" for d in plan.docs]
messages = [
- {"role": "system", "content": build_high_prompt(docs)},
+ {"role": "system", "content": plan.system_prompt},
{"role": "user", "content": request.message},
]
- # 3. Stream the grounded answer.
+ # 3. Stream the answer (grounded, or an honest deflection).
try:
async for piece in llm.chat_stream(messages):
yield sse_event({"type": "delta", "text": piece})
@@ -126,9 +170,9 @@ async def chat(
db.add(
QueryLog(
question=request.message,
- top_score=top_score,
+ top_score=plan.top_score,
chunk_hits=len(chunks),
- deflected=False,
+ deflected=plan.deflected,
sources=", ".join(source_paths),
latency_ms=total_ms,
)
@@ -142,19 +186,19 @@ async def chat(
"sources=%r total_ms=%d",
request.message,
embed_ms,
- top_score,
- get_settings().relevance_threshold,
- False,
+ plan.top_score,
+ settings.relevance_threshold,
+ plan.deflected,
source_paths,
total_ms,
)
yield sse_event(
ChatDoneEvent(
- deflected=False,
+ deflected=plan.deflected,
sources=[
- SourceRef(source=d.source, path=d.path, title=d.title) for d in docs
+ SourceRef(source=d.source, path=d.path, title=d.title) for d in plan.docs
],
- suggestions=[],
+ suggestions=plan.suggestions,
).model_dump()
)
diff --git a/app/rag/retriever.py b/app/rag/retriever.py
index f137d30..071ada7 100644
--- a/app/rag/retriever.py
+++ b/app/rag/retriever.py
@@ -64,6 +64,22 @@ def retrieve(
]
+def weak_hit_titles(chunks: Sequence[RetrievedChunk]) -> list[str]:
+ """Distinct parent-document titles of *chunks*, best chunk score first.
+
+ Deflection mode (PLAN §6, A8) is built from these *titles only* — the
+ LOW prompt and the "Maybe try" chips never see document content.
+ """
+ titles: list[str] = []
+ seen: set[uuid.UUID] = set()
+ for rc in sorted(chunks, key=lambda c: c.score, reverse=True):
+ if rc.document.id in seen:
+ continue
+ seen.add(rc.document.id)
+ titles.append(rc.document.title)
+ return titles
+
+
def select_documents(
chunks: Sequence[RetrievedChunk],
n: int | None = None,
diff --git a/app/rag/suggestions.py b/app/rag/suggestions.py
new file mode 100644
index 0000000..9288976
--- /dev/null
+++ b/app/rag/suggestions.py
@@ -0,0 +1,57 @@
+"""Deflection suggestions — the "Maybe try" chips under a deflected answer.
+
+v1 behavior (phase 04, honest-deflection story): chips are derived
+deterministically from the weak-hit document titles, so Brain only ever
+points the user at topics it actually has indexed — never invented ones.
+A model-generated list could layer on top later; the title-derived path
+is the shipped, testable one (PLAN §6: deflection offers 2-3 alternative
+questions about things the docs DO cover).
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+#: The ``done`` event carries at most this many alternative questions.
+MAX_SUGGESTIONS = 3
+
+
+def derive_suggestions(
+ titles: Sequence[str],
+ fallback: Sequence[str] = (),
+ max_n: int = MAX_SUGGESTIONS,
+) -> list[str]:
+ """Build the alternative-question chips for a deflected turn.
+
+ One chip per weak-hit title (*titles* arrive in best-chunk-score
+ order from :func:`app.rag.retriever.weak_hit_titles`), phrased as a
+ question the knowledge base can ground. If fewer than *max_n* titles
+ are available, *fallback* (the onboarding suggestions) tops the list
+ up so the user still gets 2-3 real options. Whitespace is normalized,
+ duplicates (case-insensitive) are dropped, and the result contains
+ only non-empty strings — at most *max_n* of them.
+ """
+ out: list[str] = []
+ seen: set[str] = set()
+
+ def push(item: str) -> None:
+ item = " ".join(item.split())
+ if not item:
+ return
+ key = item.lower()
+ if key in seen:
+ return
+ seen.add(key)
+ out.append(item)
+
+ for title in titles:
+ if len(out) >= max_n:
+ break
+ title = " ".join(title.split())
+ if not title:
+ continue
+ push(f"What's in your notes about {title}?")
+ for question in fallback:
+ if len(out) >= max_n:
+ break
+ push(question)
+ return out
diff --git a/frontend/assets/app.js b/frontend/assets/app.js
index ae403ee..0a783d5 100644
--- a/frontend/assets/app.js
+++ b/frontend/assets/app.js
@@ -2,10 +2,11 @@
*
* Renders suggestions, shows KB health, and runs chat turns against
* POST /api/chat (SSE, PLAN §4): deltas render live into the Brain bubble,
- * the done event appends source chips, errors surface as a red banner.
- * The full feedback state machine lands with the loading-feedback story;
- * this keeps the "never stale" contract: the button is busy for the whole
- * turn and is always re-enabled at the end.
+ * the done event appends source chips (and "Maybe try" chips when the
+ * turn was deflected — honesty gate, phase 04), errors surface as a red
+ * banner. The full feedback state machine lands with the loading-feedback
+ * story; this keeps the "never stale" contract: the button is busy for the
+ * whole turn and is always re-enabled at the end.
* All DOM ids match frontend/index.html.
*/
@@ -199,6 +200,38 @@ function appendSources(wrap, sources) {
}
}
+/* "Maybe try:" chips under a deflected bubble (honesty gate, phase 04).
+ Same .suggestion-chip component as onboarding; clicking wires what
+ exists today — fill the input + focus. One-tap submit lands with the
+ phase 05 chip component. The group is accessible (role=list +
+ aria-label) and wraps cleanly at every width. */
+function appendMaybeTry(wrap, suggestions) {
+ if (!suggestions || !suggestions.length) return;
+ const body = wrap.querySelector(".msg-body");
+ const group = document.createElement("div");
+ group.className = "maybe-try";
+ group.setAttribute("role", "list");
+ group.setAttribute("aria-label", "Maybe try");
+ const label = document.createElement("span");
+ label.className = "visually-hidden";
+ label.textContent = "Maybe try:";
+ group.appendChild(label);
+ for (const s of suggestions) {
+ const btn = document.createElement("button");
+ btn.type = "button";
+ btn.className = "suggestion-chip";
+ btn.setAttribute("role", "listitem");
+ btn.textContent = s;
+ btn.addEventListener("click", () => {
+ input.value = s;
+ autoGrow();
+ input.focus();
+ });
+ group.appendChild(btn);
+ }
+ body.appendChild(group);
+}
+
function showErrorBanner(detail) {
banner.hidden = false;
banner.classList.add("is-error");
@@ -258,7 +291,10 @@ async function handleSend(e) {
removeTyping();
wrap = addMessage("brain", "…");
}
- if (ev.deflected) wrap.classList.add("is-deflected");
+ if (ev.deflected) {
+ wrap.classList.add("is-deflected");
+ appendMaybeTry(wrap, ev.suggestions);
+ }
appendSources(wrap, ev.sources);
} else if (ev.type === "error") {
throw new Error(ev.detail || "Something went wrong on my side.");
diff --git a/frontend/assets/styles.css b/frontend/assets/styles.css
index 7cf7708..00c6048 100644
--- a/frontend/assets/styles.css
+++ b/frontend/assets/styles.css
@@ -241,6 +241,18 @@ body {
}
.source-chip:hover { background: #e2e5fd; }
+/* "Maybe try" chips under a deflected bubble (phase 04). Unlike the
+ onboarding row (which scrolls horizontally on mobile), this group wraps
+ at every width: the chips are the actionable follow-up, not decoration.
+ The pills themselves reuse .suggestion-chip (>=44px, brand-soft/ink). */
+.maybe-try {
+ display: flex;
+ align-items: center;
+ flex-wrap: wrap;
+ gap: 0.45rem;
+ padding-inline: 0.25rem;
+}
+
/* typing indicator */
.typing { display: inline-flex; gap: 5px; padding: 0.9rem 1rem; }
.typing span {
diff --git a/tests/e2e/test_honest_deflection.py b/tests/e2e/test_honest_deflection.py
new file mode 100644
index 0000000..10c9fd9
--- /dev/null
+++ b/tests/e2e/test_honest_deflection.py
@@ -0,0 +1,197 @@
+"""Phase 04 E2E (Playwright): honest deflection when retrieval is weak.
+
+Story: ``.agent/user_stories/honest-deflection.md``
+Run in isolation (DB must be up: ``podman compose up -d db``):
+
+ uv run pytest tests/e2e/test_honest_deflection.py -v --no-cov
+
+The seeded KB (``tests/fixtures/docs``) + the mock LLM's genuine
+token-overlap embeddings make the honesty gate deterministic: the
+off-topic baking question scores far below ``BOR_RELEVANCE_THRESHOLD``,
+so Brain must deflect — amber bubble, "I haven't done anything like
+that", and ≥2 "Maybe try" chips about topics it really covers.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+from pathlib import Path
+from threading import Thread
+from typing import Any
+
+import httpx
+from playwright.sync_api import Page, expect
+from sqlalchemy import select, text
+
+from app.config import Settings, get_settings
+from app.db import SessionLocal
+from app.models import QueryLog
+from app.rag.importer import ImportSummary, import_sources
+from app.rag.llm import LLMClient
+
+REPO = Path(__file__).resolve().parents[2]
+FIXTURES = REPO / "tests" / "fixtures" / "docs"
+OFF_TOPIC = "How do I bake sourdough bread?"
+# The mock's deflection answer (tests/e2e/mock_llm.py) must match this.
+DEFLECT_PHRASE = r"haven't done anything like that"
+MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
+
+
+async def _import_fixtures(mock_port: int) -> ImportSummary:
+ kwargs: dict[str, Any] = {"_env_file": None, "llm_base_url": f"http://127.0.0.1:{mock_port}/v1"}
+ settings = Settings(**kwargs) # pyright: ignore[reportCallIssue]
+ return await import_sources([FIXTURES], LLMClient(settings))
+
+
+def _run_in_thread(coro: Any) -> Any:
+ """Run a coroutine on a worker thread.
+
+ Playwright's sync API keeps an asyncio loop running on the test thread,
+ so ``asyncio.run`` cannot be called directly from a test body.
+ """
+ box: dict[str, Any] = {}
+
+ def runner() -> None:
+ try:
+ box["value"] = asyncio.run(coro)
+ except BaseException as e: # noqa: BLE001 — re-raised on the test thread
+ box["error"] = e
+
+ t = Thread(target=runner)
+ t.start()
+ t.join()
+ if "error" in box:
+ raise box["error"]
+ return box["value"]
+
+
+def _reset_db(mock_port: int, seed: bool) -> ImportSummary | None:
+ """Truncate the KB (and query log), then optionally re-import fixtures."""
+ with SessionLocal() as db:
+ db.execute(text("TRUNCATE chunks, documents, query_log"))
+ db.commit()
+ if not seed:
+ return None
+ return _run_in_thread(_import_fixtures(mock_port))
+
+
+def _deflected_chips(page: Page) -> Any:
+ return page.locator(".msg.brain.is-deflected .maybe-try .suggestion-chip")
+
+
+def test_off_topic_question_deflects_honestly(
+ page: Page, app_url: str, mock_llm: int, db_ready: None
+) -> None:
+ summary = _reset_db(mock_llm, seed=True)
+ assert summary is not None and summary.added == 3
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+ expect(page.locator("#kb-banner")).to_be_hidden()
+
+ page.fill("#message-input", OFF_TOPIC)
+ page.click("#send-btn")
+ expect(page.locator(".msg.user .bubble")).to_contain_text(OFF_TOPIC)
+
+ # The answer bubble is the deflected one: amber, honest phrasing.
+ bubble = page.locator(".msg.brain.is-deflected .bubble").first
+ bubble.wait_for(state="visible", timeout=30_000)
+ expect(bubble).to_have_text(re.compile(DEFLECT_PHRASE, re.IGNORECASE), timeout=30_000)
+ expect(page.locator(".msg.brain.is-deflected")).to_have_count(1)
+
+ # Visually distinct from a normal answer (accent-bg / accent-line).
+ style = bubble.evaluate("el => getComputedStyle(el)")
+ assert style["backgroundColor"] == "rgb(255, 247, 232)" # --accent-bg #fff7e8
+ assert style["borderTopColor"] == "rgb(245, 158, 11)" # --accent-line #f59e0b
+
+ # ≥2 "Maybe try:" chips below the bubble, in an accessible group.
+ chips = _deflected_chips(page)
+ expect(chips.first).to_be_visible(timeout=30_000)
+ assert chips.count() >= 2, "deflection must offer 2-3 alternative chips"
+ group = page.locator(".msg.brain.is-deflected .maybe-try")
+ expect(group).to_have_count(1)
+ expect(group.first).to_have_attribute("aria-label", "Maybe try")
+ expect(group.first).to_have_attribute("role", "list")
+ # Chip component contract: brand pill, ≥44px touch target.
+ chip_style = chips.first.evaluate("el => getComputedStyle(el)")
+ assert chip_style["backgroundColor"] == "rgb(238, 240, 254)" # --brand-soft
+ assert chip_style["color"] == "rgb(55, 48, 163)" # --brand-ink
+ box = chips.first.bounding_box()
+ assert box is not None and box["height"] >= 44
+
+ # Button recovers (never stale).
+ expect(page.locator("#send-btn")).to_be_enabled()
+ expect(page.locator("#send-label")).to_have_text("Send")
+
+
+def test_deflection_suggestions_are_clickable(
+ page: Page, app_url: str, mock_llm: int, db_ready: None
+) -> None:
+ _reset_db(mock_llm, seed=True)
+ page.set_default_timeout(30_000)
+ page.goto(app_url)
+ page.fill("#message-input", OFF_TOPIC)
+ page.click("#send-btn")
+ chips = _deflected_chips(page)
+ expect(chips.first).to_be_visible(timeout=30_000)
+ chip_text = chips.first.inner_text().strip()
+ assert chip_text
+
+ # Phase 04 chip contract (wire what exists): click fills + focuses.
+ chips.first.click()
+ expect(page.locator("#message-input")).to_have_value(chip_text)
+ expect(page.locator("#message-input")).to_be_focused()
+
+ # Completing the question asks it: a new user bubble + a reply —
+ # and the chip's topic is one Brain really covers, so this turn is
+ # a grounded (non-deflected) answer quoting the question.
+ page.press("#message-input", "Enter")
+ expect(page.locator(".msg.user .bubble")).to_have_count(2, timeout=30_000)
+ expect(page.locator(".msg.user .bubble").nth(1)).to_contain_text(chip_text)
+ expect(page.locator(".msg.brain .bubble")).to_have_count(2, timeout=30_000)
+ second = page.locator(".msg.brain .bubble").nth(1)
+ expect(second).to_contain_text(chip_text, timeout=30_000)
+ expect(second).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000)
+ # Still exactly one deflected turn in the conversation.
+ expect(page.locator(".msg.brain.is-deflected")).to_have_count(1)
+
+ # Button recovers after the second turn (never stale).
+ expect(page.locator("#send-btn")).to_be_enabled()
+ expect(page.locator("#send-label")).to_have_text("Send")
+
+
+def test_deflected_done_event_and_query_log(app_url: str, mock_llm: int, db_ready: None) -> None:
+ """Raw SSE contract for a deflected turn + the durable query_log row."""
+ _reset_db(mock_llm, seed=True)
+
+ frames: list[dict[str, Any]] = []
+ with httpx.stream(
+ "POST", f"{app_url}/api/chat", json={"message": OFF_TOPIC}, timeout=60.0
+ ) as r:
+ assert r.status_code == 200
+ assert r.headers["content-type"].startswith("text/event-stream")
+ buf = ""
+ for part in r.iter_text():
+ buf += part
+ while "\n\n" in buf:
+ frame, buf = buf.split("\n\n", 1)
+ if frame.strip().startswith("data:"):
+ frames.append(json.loads(frame.strip().removeprefix("data:").strip()))
+ assert buf.strip() == "" # stream ends cleanly on a frame boundary
+
+ deltas = [f for f in frames if f.get("type") == "delta"]
+ assert len(deltas) >= 2 # the deflection is streamed too
+ done = [f for f in frames if f.get("type") == "done"]
+ assert len(done) == 1
+ assert frames[-1]["type"] == "done"
+ assert done[0]["deflected"] is True
+ assert 2 <= len(done[0]["suggestions"]) <= 3
+ assert all(s.strip() for s in done[0]["suggestions"])
+
+ # Durable record: deflected=true + the weak top_score.
+ with SessionLocal() as db:
+ row = db.scalars(select(QueryLog)).one()
+ assert row.question == OFF_TOPIC
+ assert row.deflected is True
+ assert 0.0 < row.top_score < get_settings().relevance_threshold
+ assert row.chunk_hits >= 1
diff --git a/tests/integration/test_chat_api.py b/tests/integration/test_chat_api.py
index 21e5b46..7f1fd41 100644
--- a/tests/integration/test_chat_api.py
+++ b/tests/integration/test_chat_api.py
@@ -32,6 +32,7 @@ from app.rag.llm import EmbeddingError, LLMError
FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
+OFF_TOPIC = "How do I bake sourdough bread?"
DIM = 768
_TOKEN_RE = re.compile(r"[a-z0-9]+")
@@ -164,11 +165,51 @@ def test_chat_writes_query_log_row(client, db, seeded_kb: FakeRagLLM) -> None:
total_chunks = db.scalar(select(func.count()).select_from(Chunk))
assert row.chunk_hits == min(get_settings().top_k_chunks, total_chunks)
assert row.top_score > 0.0 # genuine token-overlap cosine, best hit
+ assert row.top_score >= get_settings().relevance_threshold # why the gate answered
assert row.top_score <= 1.0
assert "docs/homelab/kubernetes.md" in row.sources
assert row.latency_ms >= 0
+def test_off_topic_question_deflects_honestly(client, db, seeded_kb: FakeRagLLM) -> None:
+ """Phase 04 contract: weak retrieval ⇒ honest deflection, no fake answer."""
+ fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
+ try:
+ _, _, frames = _stream_chat(client, OFF_TOPIC)
+ finally:
+ fastapi_app.dependency_overrides.clear()
+
+ assert not any(f.get("type") == "error" for f in frames)
+ deltas = [f for f in frames if f.get("type") == "delta"]
+ assert len(deltas) >= 2 # the LLM is still called (voice stays chippy)
+ done = frames[-1]
+ assert done["type"] == "done"
+ assert done["deflected"] is True
+ # 2-3 alternative chips, all non-empty, derived from real titles/topics.
+ assert 2 <= len(done["suggestions"]) <= 3
+ assert all(s.strip() for s in done["suggestions"])
+ assert any(
+ "Deploying a New Service" in s for s in done["suggestions"]
+ ), "the best weak-hit title must be offered as a chip"
+ assert done["sources"], "weak hits are still reported as the closest sources"
+
+ # The LLM saw the LOW prompt: DEFLECT_MODE + titles, never doc content.
+ (system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
+ assert user["content"] == OFF_TOPIC
+ assert "LOW" in system["content"]
+ assert "DEFLECT_MODE" in system["content"]
+ assert "HONESTY GATE" in system["content"]
+ assert "Talos Linux" not in system["content"] # full doc content never sent
+ assert "" not in system["content"]
+
+ # Durable record: deflected=true + the weak top_score.
+ row = db.scalars(select(QueryLog)).one()
+ assert row.question == OFF_TOPIC
+ assert row.deflected is True
+ assert 0.0 < row.top_score < get_settings().relevance_threshold
+ assert row.chunk_hits >= 1
+
+
def test_chat_empty_kb_streams_empty_sources(client, db) -> None:
db.execute(text("TRUNCATE chunks, documents, query_log"))
db.commit()
@@ -179,11 +220,17 @@ def test_chat_empty_kb_streams_empty_sources(client, db) -> None:
finally:
fastapi_app.dependency_overrides.clear()
+ # Nothing retrieved ⇒ nothing to pretend to know: honest deflection.
done = frames[-1]
assert done["type"] == "done"
- assert done["deflected"] is False
+ assert done["deflected"] is True
assert done["sources"] == []
+ assert 2 <= len(done["suggestions"]) <= 3 # onboarding fallback chips
+ (system, _user) = llm.seen_messages[0][0], llm.seen_messages[0][1]
+ assert "DEFLECT_MODE" in system["content"]
+ assert "nothing close at all" in system["content"]
row = db.scalars(select(QueryLog)).one()
+ assert row.deflected is True
assert row.top_score == 0.0
assert row.chunk_hits == 0
assert row.sources == ""
diff --git a/tests/unit/test_chat_gate.py b/tests/unit/test_chat_gate.py
new file mode 100644
index 0000000..8e7aa69
--- /dev/null
+++ b/tests/unit/test_chat_gate.py
@@ -0,0 +1,322 @@
+"""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) -> RetrievedChunk:
+ return RetrievedChunk(
+ chunk_id=uuid.uuid4(),
+ position=0,
+ content=doc.content[:32],
+ score=score,
+ document=doc,
+ )
+
+
+# ---------- 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 "HIGH" 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 "LOW" 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
+
+
+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 "LOW" 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 "" 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 "HIGH" 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 _FakeSession:
+ """Stands in for the DB session: records the QueryLog row it is given."""
+
+ 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
+
+
+@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)
+ 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, _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 "HIGH" 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)