brain-of-reese/tests/e2e/test_honest_deflection.py

"""Phase 04 E2E (Playwright): honest deflection when retrieval is weak.

Story: ``.agents/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
from e2e.auth_helpers import ADMIN_PASSWORD, login

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"
OFF_TOPIC = "How do I bake sourdough bread?"
# Phase 112 (A8 revised 2026-09-14, TODO L2a — the "Mongolia case"): a
# question the LLM knows (Gershwin) but the fixture KB does not cover.
# Unlike the plain no-hit deflection above, it carries WEAK FTS hits
# (the "compos" stem matches the compose fixture docs — fts_hits >= 1)
# while its best mock token-overlap cosine (~0.12) sits BELOW
# lexical_support_floor (0.15, the mock-calibrated conftest value). The
# pre-phase gate (fts>0 → HIGH) grounded it and injected irrelevant docs
# into the prompt; the revised gate (cosine corroboration) must keep it
# LOW. The mock keys on DEFLECT_MODE, so the test pins the gate, not
# model compliance.
OUT_OF_KB = "Who composed Rhapsody in Blue?"
# 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 == 13  # A9 formats (phase 47 added quadlet+j2)
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")
    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(43, 33, 16)"  # --accent-bg #2b2110 (dark theme)
    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)")
    # --brand-soft #2d0a0a / --brand-ink #fca5a5 (dark-red rebrand)
    assert chip_style["backgroundColor"] == "rgb(45, 10, 10)"
    assert chip_style["color"] == "rgb(252, 165, 165)"
    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)
    login(page, app_url, next="/")
    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

    # Shared chip component (phase 05): one tap submits — the click fills,
    # focuses AND sends. The chip's topic is one Brain really covers, so
    # this follow-up turn is a grounded (non-deflected) answer quoting the
    # question, and the composer is left empty (submitted, not queued).
    chips.first.click()
    expect(page.locator("#message-input")).to_have_value("")
    expect(page.locator(".msg.user .bubble")).to_have_count(2, timeout=30_000)
    expect(page.locator(".msg.user .bubble").nth(1)).to_have_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)

    # Phase 79: POST /api/chat is require_user-gated — the httpx client
    # signs in as the admin first (the form login's API side: 204 + the
    # signed session cookie in the jar).
    client = httpx.Client(timeout=60.0)
    r = client.post(f"{app_url}/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204

    frames: list[dict[str, Any]] = []
    with client.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


def test_out_of_kb_question_deflects_without_citations(
    app_url: str, mock_llm: int, db_ready: None
) -> None:
    """Phase 112 acceptance (TODO L2): a known-out-of-KB question whose
    weak lexical hits NO LONGER promote (the fts>0 / cosine<floor
    quadrant, pinned end-to-end) deflects with ZERO source citations —
    done.sources is empty (the UI chips nothing under a deflected
    answer) and 2-3 concrete alternative questions are offered.

    Raw SSE (like the done-event test above): the done frame is the
    contract surface; the mock's DEFLECT_MODE phrasing proves the
    server sent the LOW prompt (the gate, not the model, decides).
    """
    _reset_db(mock_llm, seed=True)

    client = httpx.Client(timeout=60.0)
    r = client.post(f"{app_url}/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204

    frames: list[dict[str, Any]] = []
    with client.stream(
        "POST", f"{app_url}/api/chat", json={"message": OUT_OF_KB}, 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"]
    answer = "".join(d["text"] for d in deltas)
    # The DEFLECT_MODE phrasing streamed ⇒ the LOW prompt reached the
    # model (the mock answers it only for the deflection system prompt).
    assert re.search(DEFLECT_PHRASE, answer, re.IGNORECASE)

    done = frames[-1]
    assert done["type"] == "done"
    assert done["deflected"] is True
    # No false citations (TODO L2): the weak hits never ride the wire as
    # sources — a deflected answer cites nothing.
    assert done["sources"] == []
    # 2-3 concrete alternative questions, all non-empty.
    assert 2 <= len(done["suggestions"]) <= 3
    assert all(s.strip() for s in done["suggestions"])

    # Durable record: the quadrant pinned end-to-end — the lexical leg
    # FIRED (fts_hits > 0, the pre-phase gate's promotion trigger) while
    # the vector signal never cleared lexical_support_floor, so the
    # revised gate deflected. The retrieval itself stays recorded
    # (query_log = observability, not citations).
    with SessionLocal() as db:
        row = db.scalars(select(QueryLog)).one()
    assert row.question == OUT_OF_KB
    assert row.deflected is True
    assert (row.fts_hits or 0) >= 1
    assert row.top_score < get_settings().lexical_support_floor
    assert row.sources