brain-of-reese/tests/integration/test_kb_overview_api.py

"""Integration: KB overview (phase 31) — the ``<knowledge_base>`` section
of the chat system prompt.

Real Postgres (``podman compose up -d db``) seeded from
``tests/fixtures/docs/`` through the real importer; the chat path reuses
the deterministic capturing fake LLM from ``test_chat_api``
(token-overlap embeddings), so the stored row's journey —
``kb_overview`` row → per-turn PK lookup → ``<knowledge_base>`` section
of the **exact** captured system prompt (HIGH and LOW) — is verified
end-to-end without a network.

The byte-identity contract (phase 15 convention): with no row, the
captured system prompt equals the pre-phase construction
(``build_high_prompt`` / ``build_deflect_prompt`` with
``kb_overview=None``) — asserted with ``==``, not ``in``.

Requires:  podman compose up -d db
"""
from __future__ import annotations

import asyncio
import logging
from collections.abc import Iterator
from pathlib import Path

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select, text
from test_chat_api import FakeRagLLM, _stream_chat, _token_vec

from app.api import chat as chat_api
from app.main import app as fastapi_app
from app.models import Document, KbOverview
from app.rag.importer import import_sources
from app.rag.prompts import build_deflect_prompt, build_high_prompt
from app.rag.retriever import retrieve, weak_hit_titles

FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
OFF_TOPIC = "How do I bake sourdough bread?"

#: A multi-line, multi-bullet outline: the section must carry it whole
#: (well within ``BOR_KB_OVERVIEW_MAX_CHARS``) and the per-turn log line
#: records its length.
OVERVIEW = (
    "- Kubernetes cluster and node maintenance notes\n"
    "- Backup schedules and restore runbooks\n"
    "- Networking: static DNS and kafkabridge"
)


@pytest.fixture(autouse=True)
def clean_kb_overview(db) -> Iterator[None]:
    """The outline row + query log are global state: reset around every
    test so no test inherits another test's row."""
    db.execute(text("TRUNCATE kb_overview, query_log"))
    db.commit()
    yield
    db.execute(text("TRUNCATE kb_overview, query_log"))
    db.commit()


@pytest.fixture()
def seeded_kb(db) -> Iterator[FakeRagLLM]:
    """Fresh Postgres with the fixture docs imported (real pipeline)."""
    db.execute(text("TRUNCATE chunks, documents, query_log, kb_overview"))
    db.commit()
    llm = FakeRagLLM()
    summary = asyncio.run(import_sources([FIXTURES], llm, session=db))
    assert summary.added == 9  # A9 formats (phase 44 added tables.md); .hidden/ skipped
    yield llm
    db.execute(text("TRUNCATE chunks, documents, query_log, kb_overview"))
    db.commit()


def _seed_overview(db) -> None:
    db.add(KbOverview(id=1, content=OVERVIEW))
    db.commit()


def _cited_docs(db, frames: list[dict]) -> list[Document]:
    """The documents the done event cited, in citation order — the same
    list ``plan_turn`` passed to the prompt builder."""
    docs = []
    for s in frames[-1]["sources"]:
        doc = db.scalar(select(Document).where(Document.path == s["path"]))
        assert doc is not None, f"done source {s['path']!r} missing from the KB"
        docs.append(doc)
    return docs


def _turn_log_lines(caplog: pytest.LogCaptureFixture) -> list[str]:
    """The per-turn ``question=…`` log lines (PLAN §9) from this test."""
    return [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]


# ---------- no row → byte-identical to the pre-phase prompts ----------


def test_no_row_high_prompt_byte_identical_to_pre_phase(
    client: TestClient, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
    """No ``kb_overview`` row: the captured HIGH system prompt EQUALS the
    pre-phase construction exactly — the section is absent, not empty."""
    caplog.set_level(logging.INFO, logger="app.chat")
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[-1]["deflected"] is False
    (system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
    assert user["content"] == QUESTION
    expected = build_high_prompt(_cited_docs(db, frames), notes=[], kb_overview=None)
    assert system["content"] == expected
    assert "<knowledge_base>" not in system["content"]

    lines = _turn_log_lines(caplog)
    assert lines and "kb_chars=0" in lines[-1]


def test_no_row_low_prompt_byte_identical_to_pre_phase(
    client: TestClient, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
    """No row, off-topic question: the captured LOW (deflection) prompt
    EQUALS the pre-phase construction exactly."""
    caplog.set_level(logging.INFO, logger="app.chat")
    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 frames[-1]["deflected"] is True
    (system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
    assert user["content"] == OFF_TOPIC
    # Reconstruct the LOW prompt the way plan_turn does — the pre-phase
    # construction (kb_overview=None), the same deterministic retrieval.
    chunks = retrieve(db, OFF_TOPIC, _token_vec(OFF_TOPIC))
    expected = build_deflect_prompt(
        weak_hit_titles(chunks), notes=[], kb_overview=None
    )
    assert system["content"] == expected
    assert "<knowledge_base>" not in system["content"]
    assert "DEFLECT_MODE" in system["content"]

    lines = _turn_log_lines(caplog)
    assert lines and "kb_chars=0" in lines[-1]


# ---------- row present → section in BOTH prompts, exactly ----------


def test_row_high_prompt_carries_kb_section_exactly(
    client: TestClient, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
    """Stored row: the captured HIGH prompt EQUALS the construction with
    the outline — section present, ordered before ``<documents>``."""
    _seed_overview(db)
    caplog.set_level(logging.INFO, logger="app.chat")
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[-1]["deflected"] is False
    (system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
    expected = build_high_prompt(
        _cited_docs(db, frames), notes=[], kb_overview=OVERVIEW
    )
    assert system["content"] == expected

    # Section shape + order: <relevance> → <knowledge_base> → <documents>.
    prompt = system["content"]
    assert (
        prompt.index("<relevance>HIGH</relevance>")
        < prompt.index("<knowledge_base>")
        < prompt.index(OVERVIEW)
        < prompt.index("</knowledge_base>")
        < prompt.index("<documents>")
    )

    # The per-turn log line records the outline's length (PLAN §9).
    lines = _turn_log_lines(caplog)
    assert lines and f"kb_chars={len(OVERVIEW)}" in lines[-1]


def test_row_low_prompt_carries_kb_section_exactly(
    client: TestClient, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
    """Stored row, off-topic question: the LOW prompt EQUALS the
    construction with the outline — the section is in the deflection
    prompt too (real alternatives, not hallucinated ones)."""
    _seed_overview(db)
    caplog.set_level(logging.INFO, logger="app.chat")
    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 frames[-1]["deflected"] is True
    (system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
    chunks = retrieve(db, OFF_TOPIC, _token_vec(OFF_TOPIC))
    expected = build_deflect_prompt(
        weak_hit_titles(chunks), notes=[], kb_overview=OVERVIEW
    )
    assert system["content"] == expected

    prompt = system["content"]
    assert (
        prompt.index("<relevance>LOW</relevance>")
        < prompt.index("<knowledge_base>")
        < prompt.index(OVERVIEW)
        < prompt.index("</knowledge_base>")
        < prompt.index("DEFLECT_MODE")
    )
    # Deflection still sees titles only — never document content.
    assert "Talos Linux" not in prompt

    lines = _turn_log_lines(caplog)
    assert lines and f"kb_chars={len(OVERVIEW)}" in lines[-1]


def test_row_reread_every_turn_and_deleted_row_stops_it(
    client: TestClient, db, seeded_kb: FakeRagLLM
) -> None:
    """The row is read per turn (not cached): it steers every turn until
    it is deleted, and the following turn is section-free again."""
    _seed_overview(db)
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _stream_chat(client, QUESTION)
        _stream_chat(client, QUESTION)
        assert len(seeded_kb.seen_messages) == 2
        for messages in seeded_kb.seen_messages:
            assert "<knowledge_base>" in messages[0]["content"]
            assert OVERVIEW in messages[0]["content"]

        # Delete the row → the next turn's prompt drops the section.
        db.execute(text("TRUNCATE kb_overview"))
        db.commit()
        _stream_chat(client, QUESTION)
        assert len(seeded_kb.seen_messages) == 3
        assert "<knowledge_base>" not in seeded_kb.seen_messages[-1][0]["content"]
    finally:
        fastapi_app.dependency_overrides.clear()