250 lines
9.5 KiB
Python
250 lines
9.5 KiB
Python
"""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 == 8 # A9 formats; .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()
|