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

"""Phase 03 E2E (Playwright): the happy-path RAG chat turn.

Story: ``.agents/user_stories/chat-rag-answer.md``
Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_chat_rag.py -v --no-cov

Seeding reuses the real importer against ``tests/fixtures/docs/`` with the
deterministic mock embeddings (same pattern as the phase 02 story suite);
the mock LLM answers on-topic questions by quoting the question and the
document context, so the UI assertions are fully deterministic.
"""
from __future__ import annotations

import asyncio
import json
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
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"
QUESTION = "How is my Kubernetes cluster set up?"
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 test_on_topic_question_streams_grounded_answer(
    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="/")

    # Healthy KB: the offline banner must stay hidden.
    expect(page.locator("#kb-banner")).to_be_hidden()

    page.fill("#message-input", QUESTION)
    page.click("#send-btn")

    # User bubble (right, brand) shows the question.
    expect(page.locator(".msg.user .bubble")).to_contain_text(QUESTION)

    # Brain bubble streams in: the mock quotes the question and ends with a
    # deterministic marker — waiting on the marker proves content arrived.
    bubble = page.locator(".msg.brain .bubble")
    bubble.first.wait_for(state="visible", timeout=30_000)
    expect(bubble.first).to_contain_text(QUESTION, timeout=30_000)
    expect(bubble.first).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000)

    # Phase 119 (LOCKED A1): a zero-read grounded turn chips NOTHING —
    # the suggested kubernetes.md is seed context, not a citation chip
    # (the retired phase-118 A4 suggested+read union is gone; the
    # pre-phase-26 chip/contract pins retired with it). The grounding is
    # pinned by the query_log row below (LOCKED A3, untouched).
    expect(page.locator(".msg.brain .source-chip")).to_have_count(0)

    # Button recovers: enabled + "Send" (never stale).
    expect(page.locator("#send-btn")).to_be_enabled()
    expect(page.locator("#send-label")).to_have_text("Send")


def test_chat_logs_query(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", QUESTION)
    page.click("#send-btn")
    # The turn settles with the mock answer (phase 119 A1: no chip to
    # wait on — a zero-read turn chips nothing; the retired phase-118
    # A4 union is gone). The button recovery is the settle sync: the
    # client re-enables Send on the done frame, and the server writes
    # the query_log row just before yielding it.
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(
        "Deterministic mock answer for E2E", timeout=30_000
    )
    expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)

    # App still healthy after the turn.
    r = httpx.get(f"{app_url}/api/health", timeout=5)
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ok" and body["db"] == "up"

    # Durable record: exactly one query_log row for the turn.
    with SessionLocal() as db:
        rows = db.scalars(select(QueryLog)).all()
    assert len(rows) == 1
    row = rows[0]
    assert row.question == QUESTION
    assert row.deflected is False
    assert row.top_score > 0.0
    assert row.chunk_hits >= 1
    assert "docs/homelab/kubernetes.md" in row.sources
    assert row.latency_ms >= 0


def test_sse_stream_shape(app_url: str, mock_llm: int, db_ready: None) -> None:
    """Raw transport contract (PLAN §4): delta events, then one done."""
    _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": QUESTION}, 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, "answer must arrive as multiple deltas (streamed)"
    assert all(d.get("text") for d in deltas)
    assert "".join(d["text"] for d in deltas)  # non-empty answer

    done = [f for f in frames if f.get("type") == "done"]
    assert len(done) == 1
    assert frames[-1]["type"] == "done"  # done is the final event
    assert done[0]["deflected"] is False
    assert done[0]["suggestions"] == []
    # Phase 119 (LOCKED A1): done.sources = the READ docs only — this
    # plain turn read nothing, so the frame carries an EMPTY sources
    # list (the retired phase-118 A4 suggested+read union is gone; the
    # suggested doc's durable record lives in query_log, not the frame).
    assert done[0]["sources"] == [], done[0]["sources"]