brain-of-reese/tests/unit/test_chat_gate.py

"""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, KbOverview, QueryLog
from app.rag.llm import StreamPiece
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!"

#: A small KB outline standing in for the lite-generated one (phase 31).
KB_OVERVIEW = "- Homelab\n  - Kubernetes (k3s)\n- Deployments\n  - Borg backups"


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,
    cosine: float | None = None,
    fts_hit: bool = False,
    is_summary: bool = False,
) -> RetrievedChunk:
    """Fake candidate: *score* is the fused rank score; *cosine* (defaults to
    *score*) is the vector-similarity gate input."""
    return RetrievedChunk(
        chunk_id=uuid.uuid4(),
        position=-1 if is_summary else 0,
        content=doc.content[:32],
        score=score,
        document=doc,
        cosine=score if cosine is None else cosine,
        fts_hit=fts_hit,
        is_summary=is_summary,
    )


# ---------- 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 "<relevance>HIGH</relevance>" 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 "<relevance>LOW</relevance>" 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


# ---------- hybrid gate matrix (A8, revised: cosine AND fts) ----------


def test_gate_weak_cosine_with_fts_hit_still_answers() -> None:
    """cosine < threshold but a lexical hit ⇒ HIGH — the FTS-OR branch.
    This is the name-your-tool case: "kafkabridge" grounds despite weak
    vector overlap."""
    doc = _doc("Static DNS", "DNS_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.02, cosine=0.10, fts_hit=True)], _settings(threshold=0.30)
    )
    assert plan.deflected is False
    assert plan.top_score == pytest.approx(0.10)  # gate input is the cosine
    assert plan.fts_hits == 1
    assert "DNS_DOC_CONTENT" in plan.system_prompt
    assert plan.suggestions == []


def test_gate_weak_cosine_zero_fts_deflects() -> None:
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
    plan = chat_api.plan_turn([_chunk(doc, 0.02, cosine=0.10)], _settings(threshold=0.30))
    assert plan.deflected is True
    assert plan.top_score == pytest.approx(0.10)
    assert plan.fts_hits == 0


def test_gate_strong_cosine_without_fts_answers() -> None:
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
    plan = chat_api.plan_turn([_chunk(doc, 0.90, cosine=0.90)], _settings(threshold=0.30))
    assert plan.deflected is False
    assert plan.fts_hits == 0


def test_gate_fts_hits_counts_all_lexical_candidates() -> None:
    a = _doc("Alpha", "ALPHA_CONTENT")
    b = _doc("Beta", "BETA_CONTENT")
    chunks = [
        _chunk(a, 0.03, cosine=0.05, fts_hit=True),
        _chunk(a, 0.02, cosine=0.04, fts_hit=True),  # same doc, second chunk
        _chunk(b, 0.01, cosine=0.03),
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
    assert plan.deflected is False
    assert plan.fts_hits == 2  # per chunk, not per doc


def test_gate_lexical_only_chunk_does_not_inflate_cosine() -> None:
    """top_score stays the best *vector* cosine even when a lexical-only
    chunk (cosine 0.0 by construction) carries the highest fused score."""
    a = _doc("Alpha", "ALPHA_CONTENT")
    b = _doc("Beta", "BETA_CONTENT")
    chunks = [
        _chunk(a, 0.50, cosine=0.55),  # vector rank 1
        _chunk(b, 0.90, cosine=0.0, fts_hit=True),  # lexical rank 1 wins the ranking
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
    assert plan.top_score == pytest.approx(0.55)
    assert plan.deflected is False  # 0.55 >= 0.30 anyway
    # ranking follows the fused score: Beta's doc is the top source
    assert plan.docs[0].title == "Beta"


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


# ---------- summary hits (phase 30: summary → full source document) ----------


def test_summary_hit_on_selected_top_doc_counts() -> None:
    """HIGH branch: the top document was hit via its summary chunk ⇒ 1.

    Context assembly is unchanged (A7 revised): the *source* document's
    full content lands in the prompt, not the summary text alone.
    """
    a = _doc("Alpha", "ALPHA_FULL_SOURCE_CONTENT")
    b = _doc("Beta", "BETA_FULL_SOURCE_CONTENT")
    chunks = [
        _chunk(a, 0.90, is_summary=True),  # top doc reached through its summary
        _chunk(b, 0.50),
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
    assert plan.deflected is False
    assert plan.summary_hits == 1
    # The full source document is what the LLM sees (phase 24 contract).
    assert "ALPHA_FULL_SOURCE_CONTENT" in plan.system_prompt


def test_summary_hit_outside_top_n_selection_not_counted() -> None:
    """A summary chunk on a document outside the top-N (default 2) selection
    does not count — only hits that landed in the selected context do."""
    a = _doc("Alpha", "ALPHA_CONTENT")
    b = _doc("Beta", "BETA_CONTENT")
    c = _doc("Gamma", "GAMMA_CONTENT")
    chunks = [
        _chunk(a, 0.90),
        _chunk(b, 0.80),
        _chunk(c, 0.70, is_summary=True),  # 3rd-ranked doc — not selected
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
    assert plan.deflected is False
    assert [d.title for d in plan.docs] == ["Alpha", "Beta"]
    assert plan.summary_hits == 0


def test_low_branch_counts_summary_hit_on_selected_doc() -> None:
    """LOW (deflected) branch records ``summary_hits`` too: the weak hit's
    parent is still the selected (weak-hit) document."""
    a = _doc("Gamma", "GAMMA_DOC_CONTENT")
    b = _doc("Delta", "DELTA_DOC_CONTENT")
    chunks = [
        _chunk(a, 0.05, cosine=0.05, is_summary=True),  # weak cosine, no FTS
        _chunk(b, 0.03, cosine=0.03),
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.30))
    assert plan.deflected is True
    assert plan.summary_hits == 1


def test_no_summary_chunks_yields_zero_summary_hits() -> None:
    """Legacy chunks (``is_summary=false``) keep ``summary_hits == 0``."""
    a = _doc("Alpha", "ALPHA_CONTENT")
    b = _doc("Beta", "BETA_CONTENT")
    plan = chat_api.plan_turn([_chunk(a, 0.90), _chunk(b, 0.40)], _settings(threshold=0.30))
    assert plan.summary_hits == 0
    plan_low = chat_api.plan_turn([_chunk(a, 0.05, cosine=0.05)], _settings(threshold=0.30))
    assert plan_low.deflected is True
    assert plan_low.summary_hits == 0


# ---------- KB overview (phase 31: <knowledge_base> section + kb_chars) ----------


def test_plan_turn_high_injects_kb_overview() -> None:
    """HIGH branch: the stored outline lands in the prompt between
    ``<relevance>`` and ``<tuning>`` (or the ``<documents>`` body with no
    notes), and ``kb_chars`` records the outline's length."""
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.90)], _settings(threshold=0.30), kb_overview=KB_OVERVIEW
    )
    assert plan.deflected is False
    assert plan.kb_chars == len(KB_OVERVIEW)
    prompt = plan.system_prompt
    assert "<knowledge_base>" in prompt
    assert KB_OVERVIEW in prompt
    i_rel = prompt.index("<relevance>HIGH</relevance>")
    i_kb = prompt.index("<knowledge_base>")
    i_docs = prompt.index("<documents>")
    assert i_rel < i_kb < i_docs


def test_plan_turn_high_kb_section_ordered_before_tuning() -> None:
    """Both sections present: ``<relevance>`` → ``<knowledge_base>`` →
    ``<tuning>`` → ``<documents>`` (the locked phase-31 order)."""
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.90)],
        _settings(threshold=0.30),
        notes=["be concise"],
        kb_overview=KB_OVERVIEW,
    )
    prompt = plan.system_prompt
    i_rel = prompt.index("<relevance>HIGH</relevance>")
    i_kb = prompt.index("<knowledge_base>")
    i_kb_close = prompt.index("</knowledge_base>")
    i_tuning = prompt.index("<tuning>")
    i_docs = prompt.index("<documents>")
    assert i_rel < i_kb < i_kb_close < i_tuning < i_docs
    assert plan.kb_chars == len(KB_OVERVIEW)


def test_plan_turn_low_injects_kb_overview() -> None:
    """LOW (deflected) branch: the outline is injected there too, ahead
    of the DEFLECT_MODE body, and document content stays excluded."""
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_NEVER_SENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.10)], _settings(threshold=0.30), kb_overview=KB_OVERVIEW
    )
    assert plan.deflected is True
    assert plan.kb_chars == len(KB_OVERVIEW)
    prompt = plan.system_prompt
    assert "<knowledge_base>" in prompt
    assert KB_OVERVIEW in prompt
    i_rel = prompt.index("<relevance>LOW</relevance>")
    i_kb = prompt.index("<knowledge_base>")
    i_mode = prompt.index("DEFLECT_MODE")
    assert i_rel < i_kb < i_mode
    assert "TALOS_DOC_NEVER_SENT" not in prompt  # titles only, still


def test_plan_turn_empty_overview_keeps_prompt_and_zero_kb_chars() -> None:
    """No outline (None/empty/blank) → ``kb_chars == 0`` and a prompt
    byte-identical to the no-overview build in both branches."""
    high_doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_CONTENT")
    low_doc = _doc("Backup Strategy", "BACKUP_DOC_CONTENT")
    for chunks, kb in (
        ([_chunk(high_doc, 0.90)], None),
        ([_chunk(high_doc, 0.90)], ""),
        ([_chunk(high_doc, 0.90)], "   \n\t "),
        ([_chunk(low_doc, 0.10)], None),
        ([_chunk(low_doc, 0.10)], ""),
    ):
        plan = chat_api.plan_turn(chunks, _settings(threshold=0.30), kb_overview=kb)
        baseline = chat_api.plan_turn(chunks, _settings(threshold=0.30))
        assert plan.kb_chars == 0
        assert baseline.kb_chars == 0
        assert plan.system_prompt == baseline.system_prompt  # byte-identical
        assert "<knowledge_base>" not in plan.system_prompt


# ---------- 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 "<relevance>LOW</relevance>" 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 "<documents>" 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 "<relevance>HIGH</relevance>" 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 StreamPiece("content", self.answer[i : i + 12])


class _FakeSteeringResult:
    """Empty steering-note result (no stored notes in these unit tests)."""

    def all(self) -> list[Any]:
        return []


class _FakeSession:
    """Stands in for the DB session: records the QueryLog row it is given.

    ``scalars`` always yields no steering notes (phase 15) so the chat
    turn's ``load_steering_notes`` call stays a no-op here, and ``get``
    returns the single ``kb_overview`` row when one is configured
    (phase 31) — ``None`` by default, i.e. no stored outline.
    """

    def __init__(self, kb_overview: str = "") -> None:
        self.added: list[Any] = []
        self.commits = 0
        self.kb_overview = kb_overview

    def add(self, obj: Any) -> None:
        self.added.append(obj)

    def commit(self) -> None:
        self.commits += 1

    def scalars(self, _stmt: Any) -> _FakeSteeringResult:
        return _FakeSteeringResult()

    def get(self, model: Any, pk: Any) -> Any:
        if model is KbOverview and self.kb_overview:
            return KbOverview(id=1, content=self.kb_overview)
        return None


@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)
    # These tests assert against a specific gate threshold; keep it stable
    # regardless of the production default (0.62) or any .env.
    monkeypatch.setattr(
        chat_api,
        "get_settings",
        lambda: Settings(_env_file=None, relevance_threshold=0.30),  # pyright: ignore[reportCallIssue]
    )
    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, _question: str, _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 "<relevance>HIGH</relevance>" 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)


# ---------- endpoint: KB overview row (phase 31) ----------


def test_endpoint_stored_kb_row_injected_into_system_prompt(
    client: TestClient,
    gate_env: tuple[_FakeSession, _CannedLLM],
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A non-empty ``kb_overview`` row (read via one PK lookup) reaches
    the LLM's system prompt in both modes, and the per-turn log line
    records ``kb_chars=N`` (PLAN §9)."""
    session, llm = gate_env
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_SENT")
    monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.30)]))

    session.kb_overview = f"  {KB_OVERVIEW}  "  # the loader trims it
    with caplog.at_level("INFO", logger="app.chat"):
        _ask(client, "How is my Kubernetes cluster set up?")

    (system, _user) = llm.seen[0][0], llm.seen[0][1]
    assert "<knowledge_base>" in system["content"]
    assert KB_OVERVIEW in system["content"]
    assert system["content"].index("<relevance>HIGH</relevance>") < system["content"].index(
        "<knowledge_base>"
    ) < system["content"].index("<documents>")
    log_lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert log_lines and f"kb_chars={len(KB_OVERVIEW)}" in log_lines[-1]


def test_endpoint_no_kb_row_prompt_unchanged(
    client: TestClient,
    gate_env: tuple[_FakeSession, _CannedLLM],
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """No ``kb_overview`` row → the section is absent (byte-identical to
    the pre-phase prompt) and the log line records ``kb_chars=0``."""
    session, llm = gate_env
    assert session.kb_overview == ""  # fixture default: no stored row
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_SENT")
    monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.30)]))

    with caplog.at_level("INFO", logger="app.chat"):
        _ask(client, "How is my Kubernetes cluster set up?")

    (system, _user) = llm.seen[0][0], llm.seen[0][1]
    assert "<knowledge_base>" not in system["content"]
    log_lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert log_lines and "kb_chars=0" in log_lines[-1]