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 datetime import UTC, datetime
from typing import TYPE_CHECKING, 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.agent import AGENT_TOOLS
from app.rag.llm import StreamPiece
from app.rag.retriever import RetrievedChunk, weak_hit_titles
from app.rag.suggestions import MAX_SUGGESTIONS, derive_suggestions
from tests.conftest import ADMIN_PASSWORD

if TYPE_CHECKING:
    from app.rag.scaffolding import ScaffoldingFilter

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, floor: float | None = None) -> Settings:
    if floor is None:
        floor = threshold * 0.5  # half the threshold — keeps existing tests green
    return Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        relevance_threshold=threshold,
        lexical_support_floor=floor,
    )


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,
        # Phase 106, D5: the HIGH block formats the row's created_at
        # UTC date part — the detached fixture carries it (the NOT NULL
        # DB column guarantees it for real rows).
        created_at=datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC),
    )


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 corroborated by cosine >= floor
    ⇒ HIGH — the FTS-OR branch. This is the name-your-tool case:
    "kafkabridge" grounds despite weak vector overlap.

    A8 revised 2026-09-14: FTS alone no longer promotes; cosine must also
    clear lexical_support_floor (here 0.15 = half of threshold 0.30)."""
    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, floor=0.05),  # floor=0.05 so 0.10 >= floor
    )
    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, floor=0.03))
    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
    # Phase 113 (the usefulness bar): Beta's doc ranks first by fused
    # score, but a lexical-only doc (cosine 0.0 by construction) cannot
    # clear the bar — it lands in the RELATED tier, never the cited one.
    assert plan.docs[0].title == "Alpha"
    assert plan.related_docs[0].title == "Beta"


# ---------- lexical support floor (A8 revised 2026-09-14) ----------


def test_gate_fts_hit_below_floor_deflects() -> None:
    """The Mongolia case: FTS hit with cosine below lexical_support_floor
    → LOW (deflected). The lexical-only hit no longer promotes to HIGH.
    This is the regression pin for phase 112."""
    doc = _doc("Capital Quest", "QUEST_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.05, cosine=0.10, fts_hit=True)],
        _settings(threshold=0.62),
    )
    assert plan.deflected is True
    assert plan.top_score == pytest.approx(0.10)
    assert plan.fts_hits == 1
    assert "DEFLECT_MODE" in plan.system_prompt
    assert "QUEST_DOC_CONTENT" not in plan.system_prompt
    assert "Capital Quest" in plan.system_prompt  # title only
    assert plan.suggestions  # derived from weak-hit titles


def test_gate_fts_hit_at_floor_answers() -> None:
    """FTS hit with cosine exactly at lexical_support_floor → HIGH.
    The floor is inclusive (>=), not strict (<)."""
    doc = _doc("Capital Quest", "QUEST_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.40, cosine=0.35, fts_hit=True)],
        _settings(threshold=0.62),
    )
    assert plan.deflected is False
    assert plan.top_score == pytest.approx(0.35)
    assert plan.fts_hits == 1
    assert "QUEST_DOC_CONTENT" in plan.system_prompt
    assert plan.suggestions == []


def test_gate_fts_hit_above_floor_below_threshold_answers() -> None:
    """FTS hit with cosine between floor and threshold → HIGH.
    The corroborated-lexical path fires."""
    doc = _doc("Capital Quest", "QUEST_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.50, cosine=0.50, fts_hit=True)],
        _settings(threshold=0.62),
    )
    assert plan.deflected is False
    assert plan.top_score == pytest.approx(0.50)
    assert plan.fts_hits == 1
    assert "QUEST_DOC_CONTENT" in plan.system_prompt
    assert plan.suggestions == []


def test_gate_fts_hit_above_code_default_floor_answers() -> None:
    """The quadrant table's "0.50 with default settings" row: the CODE
    defaults (``test_lexical_support_floor_validation_default`` pins them:
    threshold 0.62 / floor 0.35) — fts>0 + cosine 0.50 >= 0.35 → HIGH.
    Named literally (not via the helper's half-threshold floor) so the
    production-default path is pinned on its own."""
    doc = _doc("Capital Quest", "QUEST_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.50, cosine=0.50, fts_hit=True)],
        _settings(threshold=0.62, floor=0.35),
    )
    assert plan.deflected is False
    assert plan.top_score == pytest.approx(0.50)
    assert plan.fts_hits == 1
    assert "QUEST_DOC_CONTENT" in plan.system_prompt
    assert plan.suggestions == []


def test_gate_high_cosine_overrides_fts_deflection() -> None:
    """Strong cosine (>= threshold) → HIGH regardless of FTS status.
    The cosine-primary path is unchanged."""
    doc = _doc("Capital Quest", "QUEST_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.90, cosine=0.80, fts_hit=True)],
        _settings(threshold=0.62),
    )
    assert plan.deflected is False
    assert plan.top_score == pytest.approx(0.80)
    assert plan.fts_hits == 1
    assert "QUEST_DOC_CONTENT" in plan.system_prompt
    assert plan.suggestions == []


def test_gate_fts_no_cosine_deflects() -> None:
    """FTS hit with cosine = 0.0 → LOW (the extreme Mongolia case)."""
    doc = _doc("Capital Quest", "QUEST_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.90, cosine=0.0, fts_hit=True)],
        _settings(threshold=0.62),
    )
    assert plan.deflected is True
    assert plan.top_score == 0.0
    assert plan.fts_hits == 1
    assert "DEFLECT_MODE" in plan.system_prompt


def test_gate_multiple_fts_below_floor_deflects() -> None:
    """Multiple FTS hits, all below lexical_support_floor → LOW.
    The gate requires the BEST cosine to clear the floor, not just any hit."""
    a = _doc("Alpha Quest", "ALPHA_CONTENT")
    b = _doc("Beta Quest", "BETA_CONTENT")
    chunks = [
        _chunk(a, 0.30, cosine=0.20, fts_hit=True),
        _chunk(b, 0.25, cosine=0.15, fts_hit=True),
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.62))
    assert plan.deflected is True
    assert plan.fts_hits == 2
    assert "DEFLECT_MODE" in plan.system_prompt


def test_gate_one_fts_above_floor_answers() -> None:
    """Multiple chunks, one FTS hit above floor → HIGH.
    The best cosine (from the corroborated hit) clears the floor."""
    a = _doc("Alpha Quest", "ALPHA_CONTENT")
    b = _doc("Beta Quest", "BETA_CONTENT")
    chunks = [
        _chunk(a, 0.30, cosine=0.20, fts_hit=True),  # below floor
        _chunk(b, 0.25, cosine=0.40, fts_hit=True),  # above floor
    ]
    plan = chat_api.plan_turn(chunks, _settings(threshold=0.62))
    assert plan.deflected is False
    assert plan.fts_hits == 2
    assert "ALPHA_CONTENT" in plan.system_prompt
    assert "BETA_CONTENT" in plan.system_prompt


# ---------- config validation (lexical_support_floor) ----------


def test_lexical_support_floor_validation_floor_above_threshold_fails() -> None:
    """lexical_support_floor > relevance_threshold is rejected at startup."""
    with pytest.raises(ValueError, match="lexical_support_floor"):
        Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            relevance_threshold=0.62,
            lexical_support_floor=0.70,
        )


def test_lexical_support_floor_validation_negative_fails() -> None:
    """Negative lexical_support_floor is rejected."""
    with pytest.raises(ValueError, match="lexical_support_floor"):
        Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            lexical_support_floor=-0.1,
        )


def test_lexical_support_floor_validation_at_threshold_succeeds() -> None:
    """lexical_support_floor == relevance_threshold is legal."""
    s = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        relevance_threshold=0.62,
        lexical_support_floor=0.62,
    )
    assert s.lexical_support_floor == 0.62


def test_lexical_support_floor_validation_default() -> None:
    """Default lexical_support_floor is 0.35."""
    import os
    # Conftest sets BOR_RELEVANCE_THRESHOLD=0.30 and BOR_LEXICAL_SUPPORT_FLOOR=0.15.
    # We need the CODE defaults, so clear both and let the class defaults apply.
    saved_relevance = os.environ.pop("BOR_RELEVANCE_THRESHOLD", None)
    saved_floor = os.environ.pop("BOR_LEXICAL_SUPPORT_FLOOR", None)
    try:
        s = Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
        assert s.lexical_support_floor == 0.35
        assert s.relevance_threshold == 0.62
    finally:
        if saved_relevance is not None:
            os.environ["BOR_RELEVANCE_THRESHOLD"] = saved_relevance
        if saved_floor is not None:
            os.environ["BOR_LEXICAL_SUPPORT_FLOOR"] = saved_floor


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


# ---------- usefulness bar tiering (phase 113, LOCKED A2/A4) ----------


def _bar_settings(
    threshold: float = 0.62,
    lex_floor: float = 0.35,
    source_floor: float = 0.35,
    related_cap: int = 2,
    top_n: int = 2,
) -> Settings:
    """Explicit code defaults (production calibration) — the env's mock-
    calibrated floor (tests/conftest.py) is overridden per test."""
    return Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        relevance_threshold=threshold,
        lexical_support_floor=lex_floor,
        source_usefulness_floor=source_floor,
        related_max_docs=related_cap,
        top_n_docs=top_n,
    )


def test_plan_turn_high_tiers_strong_plus_weak() -> None:
    """Grounded turn: the bar-clearing doc is cited (and in the prompt),
    the weak 2nd doc loses its citation slot and lands in related_docs —
    the recurring incident's fix at the plan level."""
    strong = _doc("Kubernetes Homelab Cluster", "STRONG_DOC_CONTENT")
    weak = _doc("Backup Strategy", "WEAK_DOC_CONTENT")
    chunks = [_chunk(strong, 0.90, cosine=0.80), _chunk(weak, 0.80, cosine=0.20)]
    plan = chat_api.plan_turn(chunks, _bar_settings())
    assert plan.deflected is False
    assert [d.title for d in plan.docs] == ["Kubernetes Homelab Cluster"]
    assert [d.title for d in plan.related_docs] == ["Backup Strategy"]
    # The HIGH prompt carries the cited doc's content only.
    assert "STRONG_DOC_CONTENT" in plan.system_prompt
    assert "WEAK_DOC_CONTENT" not in plan.system_prompt


def test_plan_turn_high_single_strong_doc_yields_one_cited() -> None:
    """top_n_docs is a CEILING, not a quota: one strong doc ⇒ one cited doc,
    an empty related tier (LOCKED A2)."""
    strong = _doc("Kubernetes Homelab Cluster", "STRONG_DOC_CONTENT")
    plan = chat_api.plan_turn([_chunk(strong, 0.90, cosine=0.80)], _bar_settings())
    assert plan.deflected is False
    assert [d.title for d in plan.docs] == ["Kubernetes Homelab Cluster"]
    assert plan.related_docs == []


def test_plan_turn_low_weak_hits_fall_to_related() -> None:
    """Deflected turn: nothing clears the bar ⇒ the cited tier is empty
    and the weak hits fall to related_docs (the done frame's home for
    their visibility). The LOW prompt is unchanged (titles only)."""
    a = _doc("Alpha", "ALPHA_DOC_NEVER_SENT")
    b = _doc("Beta", "BETA_DOC_NEVER_SENT")
    chunks = [_chunk(a, 0.30, cosine=0.20), _chunk(b, 0.20, cosine=0.15)]
    plan = chat_api.plan_turn(chunks, _bar_settings())
    assert plan.deflected is True
    assert plan.docs == []  # no citation slot below the bar
    assert [d.title for d in plan.related_docs] == ["Alpha", "Beta"]  # rank order
    assert "ALPHA_DOC_NEVER_SENT" not in plan.system_prompt
    assert "Beta" in plan.system_prompt  # weak-hit titles still carried
    assert plan.suggestions  # chips unchanged


def test_plan_turn_related_cap_zero_kills_the_related_tier() -> None:
    """related_max_docs=0 is the kill switch: weak docs are scored but
    neither cited nor related (the pre-phase-113 visibility, minus the
    false citation — a deflected turn cites nothing)."""
    a = _doc("Alpha", "AAA")
    b = _doc("Beta", "BBB")
    chunks = [_chunk(a, 0.30, cosine=0.20), _chunk(b, 0.20, cosine=0.15)]
    plan = chat_api.plan_turn(chunks, _bar_settings(related_cap=0))
    assert plan.deflected is True
    assert plan.docs == []
    assert plan.related_docs == []


def test_plan_turn_floor_zero_keeps_legacy_cited_docs() -> None:
    """source_usefulness_floor=0 disables the bar: plan.docs is the legacy
    rank-ordered top-N (any cosine, incl. 0.0 lexical-only) and the
    related tier is empty."""
    a = _doc("Alpha", "AAA")
    b = _doc("Beta", "BBB")
    chunks = [
        _chunk(a, 0.90, cosine=0.0, fts_hit=True),  # lexical-only, rank 1
        _chunk(b, 0.80, cosine=0.10),
    ]
    plan = chat_api.plan_turn(
        chunks, _bar_settings(source_floor=0.0, lex_floor=0.05)
    )
    assert plan.deflected is False  # 0.10 + the fts hit clears the 0.05 lex floor
    assert [d.title for d in plan.docs] == ["Alpha", "Beta"]  # legacy order
    assert plan.related_docs == []


def test_plan_turn_lexically_grounded_below_source_floor_has_no_cited_docs() -> None:
    """The degenerate operator config (citation bar STRICTER than the
    grounding bar): a turn grounded by a corroborated-lexical hit whose
    cosine sits between the two floors has an EMPTY cited tier — the HIGH
    prompt carries no document content (the tools remain the escape
    hatch). The bar is a citation filter, not a gate input."""
    doc = _doc("Static DNS", "DNS_DOC_CONTENT")
    plan = chat_api.plan_turn(
        [_chunk(doc, 0.50, cosine=0.35, fts_hit=True)],
        _bar_settings(threshold=0.62, lex_floor=0.30, source_floor=0.50),
    )
    assert plan.deflected is False  # 0.35 >= lex floor 0.30, fts fired
    assert plan.docs == []  # 0.35 < source floor 0.50 — no citation slot
    assert "DNS_DOC_CONTENT" not in plan.system_prompt
    # The doc still SCORED — it rides the related tier (the "nearby docs"
    # row), it is not invisible.
    assert [d.title for d in plan.related_docs] == ["Static DNS"]


def test_plan_turn_related_tier_capped_in_rank_order() -> None:
    """Grounded turn, four bar-clearing docs, ceiling 2: cited = the top-2
    in rank order; related = the next two (the ceiling overflow, any
    cosine), capped at related_max_docs."""
    docs_in = [
        _doc(f"Doc {i}", f"DOC_CONTENT_{i}") for i in range(4)
    ]
    chunks = [
        _chunk(d, 0.9 - 0.1 * i, cosine=0.8 - 0.05 * i) for i, d in enumerate(docs_in)
    ]
    plan = chat_api.plan_turn(chunks, _bar_settings(top_n=2, related_cap=2))
    assert plan.deflected is False
    assert [d.title for d in plan.docs] == ["Doc 0", "Doc 1"]
    assert [d.title for d in plan.related_docs] == ["Doc 2", "Doc 3"]


# ---------- 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
    # Phase 71 (owner-permitted 2026-09-03): the deflection plain-text
    # line — the LOW turn offers no tools, so any tool markup there is
    # always wrong (prevention at the prompt; the filter + recovery is
    # the backstop).
    assert "Reply in plain text only — you have no tools in this mode." in prompt
    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
    # Phase 71: the deflection plain-text line never leaks into HIGH.
    assert "Reply in plain text only" 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.

    Never emits tool calls, so a grounded turn through the phase-37 agent
    loop ends after the single (tools-offered) request; *seen_tools*
    records each request's ``tools`` value for the phase-37 wiring pins.
    """

    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]]] = []
        self.seen_tools: list[list[dict[str, Any]] | None] = []

    async def embed_one(self, _text: str) -> list[float]:
        return [0.0] * 768

    async def chat_stream(
        self,
        messages: list[dict[str, str]],
        tools: list[dict[str, Any]] | None = None,
        scaffolding: ScaffoldingFilter | None = None,  # phase 71 pass-through
    ):
        self.seen.append(messages)
        self.seen_tools.append(tools)
        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 __enter__(self) -> _FakeSession:
        return self

    def __exit__(self, *args: Any) -> None:
        pass

    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(autouse=True)
def _admin_signed_in(client: TestClient) -> None:
    """Phase 79 (task 03): ``POST /api/chat`` is user-gated — the
    endpoint-level tests run as the signed-in ADMIN, so the shared
    ``client`` logs in once per test. The admin session short-circuits
    ``require_user`` before any DB touch, so the fake-session wiring in
    ``gate_env`` is untouched."""
    r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"


@pytest.fixture()
def gate_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[_FakeSession, _CannedLLM]]:
    """``POST /api/chat`` with retriever, session, and LLM all faked.

    SEC-14-04: the chat endpoint uses short-lived sessions via
    ``SessionLocal()`` — monkeypatch ``chat_api.SessionLocal`` instead
    of overriding ``get_db``.
    """
    monkeypatch.setattr(chat_api, "db_available", lambda: True)
    session = _FakeSession()
    llm = _CannedLLM()
    monkeypatch.setattr(chat_api, "SessionLocal", 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"])
    # Phase 112 (A8 revised, TODO L2): a deflected turn cites nothing —
    # done.sources is the citation surface (the UI chips every entry as
    # "the answer used this"), and the weak hits are scored docs, not
    # citations.
    assert done["sources"] == []

    # 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. The retrieval itself
    # stays recorded (observability unchanged — query_log records
    # retrieval, not citations; the phase-113 A3 precedent).
    (row,) = session.added
    assert isinstance(row, QueryLog)
    assert row.deflected is True
    assert row.top_score == pytest.approx(0.2999)
    assert row.sources  # the weak-hit doc's path, for threshold tuning
    assert session.commits == 1


def test_endpoint_grounded_turn_runs_agent_loop_with_tools(
    client: TestClient,
    gate_env: tuple[_FakeSession, _CannedLLM],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 37: a grounded endpoint turn runs the agent loop — the
    single no-tool-call request carries ``AGENT_TOOLS`` (the default
    round cap keeps the tools offered), no ``tool`` frames stream, and
    the ``done`` event is the plain retrieval shape (the tool-free
    answer is byte-identical)."""
    _session, llm = gate_env
    doc = _doc("Kubernetes Homelab Cluster", "TALOS_DOC_SENT")
    monkeypatch.setattr(chat_api, "retrieve", _fake_retriever([_chunk(doc, 0.90)]))

    frames = _ask(client, "How is my Kubernetes cluster set up?")

    assert frames[-1]["type"] == "done"
    assert frames[-1]["deflected"] is False
    assert not any(f["type"] == "tool" for f in frames)
    assert len(llm.seen) == 1
    assert llm.seen_tools == [AGENT_TOOLS]  # one request, tools offered
    # The system prompt is the HIGH prompt with the <tools> instructions
    # (phase 70: the harness-aligned ls/read/grep copy — new names in,
    # old phase-37/68 names out).
    (system, _user) = llm.seen[0][0], llm.seen[0][1]
    assert "<relevance>HIGH</relevance>" in system["content"]
    assert "<tools>" in system["content"]
    for tool in ("`ls`", "`grep`", "`read`"):
        assert tool in system["content"]
    for old in ("list_documents", "read_document", "search_documents"):
        assert old not in system["content"]


def test_endpoint_deflected_turn_never_offers_tools(
    client: TestClient,
    gate_env: tuple[_FakeSession, _CannedLLM],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 37: a deflected endpoint turn keeps the direct
    ``chat_stream`` — the single request carries no ``tools`` key
    (``seen_tools == [None]``), A8 byte-identical."""
    _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?")

    assert frames[-1]["type"] == "done"
    assert frames[-1]["deflected"] is True
    assert not any(f["type"] == "tool" for f in frames)
    assert len(llm.seen) == 1
    assert llm.seen_tools == [None]
    (system, _user) = llm.seen[0][0], llm.seen[0][1]
    assert "<tools>" not in system["content"]  # the LOW prompt never carries it
    # Phase 70: the rewritten <tools> copy stays out of the deflected path
    # (the LOW prompt is byte-identical to the pre-phase text).
    assert "You may extend your context with three tools" not in system["content"]


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]