"""Integration: POST /api/chat — the RAG turn end-to-end.

Real Postgres (compose) seeded from ``tests/fixtures/docs/`` through the
real importer; the LLM client is a deterministic in-process fake
(token-overlap embeddings, canned streamed answer), so no network is
needed and the cosine ordering is meaningful: the Kubernetes question
retrieves the Kubernetes document.

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

import asyncio
import hashlib
import json
import logging
import math
import re
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast

import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError
from sqlalchemy import delete, func, select, text
from sqlalchemy.orm import Session

from app.api import chat as chat_api
from app.config import Settings, get_settings
from app.main import app as fastapi_app
from app.models import Chunk, Document, GitSource, QueryLog
from app.rag import agent
from app.rag.agent import AGENT_TOOLS, READ_TRUNCATION_NOTICE
from app.rag.importer import import_sources
from app.rag.llm import EmbeddingError, LLMError, StreamPiece, ToolCallPiece, ToolResultPiece
from app.rag.prompts import build_high_prompt
from app.rag.retriever import TRUNCATION_MARKER
from app.schemas import ChatDoneEvent, SourceRef
from tests.conftest import ADMIN_PASSWORD

#: The fixture documents' fixed creation date (phase 106, D5): the
#: ``read`` result's second line is the row's ``created_at`` UTC date
#: part — a fixed value keeps the read-result pins deterministic
#: (instead of the ``now()`` server default of a bare insert).
_FIXTURE_CREATED_AT = datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC)


if TYPE_CHECKING:
    from app.rag.scaffolding import ScaffoldingFilter

FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
OFF_TOPIC = "How do I bake sourdough bread?"
DIM = 768
_TOKEN_RE = re.compile(r"[a-z0-9]+")


def _token_vec(text: str) -> list[float]:
    """Bag-of-words unit vector — same algorithm as the E2E mock, so the
    cosine behaviour here matches what the story E2E sees."""
    vec = [0.0] * DIM
    for tok in _TOKEN_RE.findall(text.lower()):
        vec[int(hashlib.md5(tok.encode()).hexdigest(), 16) % DIM] += 1.0
    norm = math.sqrt(sum(v * v for v in vec)) or 1.0
    return [v / norm for v in vec]


class FakeRagLLM:
    """Duck-typed :class:`app.rag.llm.LLMClient` stand-in for the chat path."""

    def __init__(
        self,
        answer: str = "Hey — you've got this! Talos, Cilium, three nodes. 🧠",
        thinking: str = "",
        embed_error: Exception | None = None,
        stream_error: Exception | None = None,
        fail_mid_stream: bool = False,
        tool_script: list[list[StreamPiece | ToolCallPiece]] | None = None,
        embed_fail_count: int = 0,
        stream_fail_count: int = 0,
        answer_sequence: list[str] | None = None,
    ) -> None:
        self.settings = Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
        self.embed_batches = 0
        self.answer = answer
        self.thinking = thinking
        self.embed_error = embed_error
        self.stream_error = stream_error
        self.fail_mid_stream = fail_mid_stream
        #: Phase 71: per-request canned answers (the recovery matrix):
        #: request *i* (0-based, in ``seen_messages`` order) yields
        #: ``answer_sequence[i]``; once exhausted it falls back to
        #: ``answer``. ``None`` keeps the single-``answer`` behavior.
        self.answer_sequence = answer_sequence
        #: Phase 67: the first N ``embed_one`` calls raise an
        #: ``EmbeddingError`` (then succeed) — a dead-then-recovered
        #: embeddings endpoint for the retry loop.
        self.embed_fail_count = embed_fail_count
        #: Phase 67: the first N ``chat_stream`` requests die with an
        #: ``LLMError`` BEFORE any piece (then succeed) — a dead-then-
        #: recovered answer endpoint for the pre-first-piece retry rule.
        self.stream_fail_count = stream_fail_count
        self.question_embeds: list[str] = []
        #: Phase 74: assistant history messages may carry
        #: ``reasoning_content`` — the dict values stay strings, but the
        #: key set is wider than the pre-phase ``{role, content}`` shape.
        self.seen_messages: list[list[dict[str, Any]]] = []
        #: Every request's ``tools`` value (phase 37) — ``None`` is the
        #: pre-phase request shape (the key is absent from the payload).
        self.seen_tools: list[list[dict[str, Any]] | None] = []
        #: Canned per-agent-round piece lists (phase 37): ``tool_script[i]``
        #: is yielded for the *i*-th request that carries a non-None
        #: ``tools`` parameter (a request the agent loop is offering tools
        #: on). A request without tools — the deflected direct path, the
        #: cap-forced answer request, or the kill-switch
        #: (``agent_max_rounds=0``) single-request path — always yields the
        #: thinking + answer stream below, so a deflected turn through this
        #: fake is byte-identical to the plain fake's output.
        self.tool_script: list[list[StreamPiece | ToolCallPiece]] = list(tool_script or [])

    async def embed(self, texts: list[str]) -> list[list[float]]:
        self.embed_batches += 1
        return [_token_vec(t) for t in texts]

    async def chat(
        self, messages: list[dict[str, str]], model: str | None = None
    ) -> str:
        """Deterministic ``lite`` stand-in for the import-time summaries
        (phase 30) — same convention as ``tests.fakes.FakeEmbedder.chat``."""
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")

    async def embed_one(self, text: str) -> list[float]:
        if self.embed_error is not None:
            raise self.embed_error
        if self.embed_fail_count > 0:
            self.embed_fail_count -= 1
            self.question_embeds.append(text)
            raise EmbeddingError("simulated embeddings endpoint failure")
        self.question_embeds.append(text)
        return _token_vec(text)

    def _answer_for_request(self) -> str:
        """The canned answer for the request that was just recorded
        (phase 71 ``answer_sequence``; ``None`` → the single answer)."""
        if self.answer_sequence is None:
            return self.answer
        index = len(self.seen_messages) - 1
        if index < len(self.answer_sequence):
            return self.answer_sequence[index]
        return self.answer

    async def chat_stream(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
        scaffolding: ScaffoldingFilter | None = None,
    ):
        """Typed stream (phase 17): ``thinking`` slices (same 12-char
        cadence as content) **before** the content pieces. With the
        default ``thinking=""`` this yields content-only pieces — today's
        behavior, new yield type. Phase 37: *tools* is the agent loop's
        ``tools=…`` passthrough (recorded in ``seen_tools``); a request
        with tools consumes the next ``tool_script`` entry, if any.
        Phase 71: *scaffolding* mirrors ``LLMClient.chat_stream`` — the
        canned content pieces are fed through the caller's filter (an
        empty clean result yields nothing) and the held tail is flushed
        on normal completion, so a scaffolding-only canned answer streams
        zero content pieces and leaves ``stripped_chars`` behind for the
        recovery policy to key on. ``None`` (e.g. pre-phase callers) keeps
        the byte-identical raw path."""
        self.seen_messages.append(messages)
        self.seen_tools.append(tools)
        if self.stream_error is not None:
            raise self.stream_error
        if self.stream_fail_count > 0:
            self.stream_fail_count -= 1
            raise LLMError("simulated pre-piece endpoint failure")
        mid_stream_drop = False
        raw: list[StreamPiece | ToolCallPiece]
        if tools is not None and self.tool_script:
            raw = self.tool_script.pop(0)
        elif self.fail_mid_stream:
            raw = [StreamPiece("content", "partial ")]
            mid_stream_drop = True
        else:
            answer = self._answer_for_request()
            raw = cast(
                "list[StreamPiece | ToolCallPiece]",
                [
                    StreamPiece("thinking", self.thinking[i : i + 12])
                    for i in range(0, len(self.thinking), 12)
                ]
                + [
                    StreamPiece("content", answer[i : i + 12])
                    for i in range(0, len(answer), 12)
                ],
            )
        if scaffolding is None:
            for piece in raw:
                yield piece
            if mid_stream_drop:
                raise LLMError("mid-stream dropout")
            return
        for piece in raw:
            if isinstance(piece, StreamPiece) and piece.kind == "content":
                cleaned = scaffolding.feed(piece.text)
                if cleaned:
                    yield StreamPiece("content", cleaned)
            else:
                yield piece
        if mid_stream_drop:
            # The tail is NOT flushed on a failed stream — the real
            # client only flushes a cleanly completed one.
            raise LLMError("mid-stream dropout")
        tail = scaffolding.flush()
        if tail:
            yield StreamPiece("content", tail)


@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"))
    db.commit()
    llm = FakeRagLLM()
    summary = asyncio.run(import_sources([FIXTURES], llm, session=db))
    assert summary.added == 13  # A9 formats (phase 47 added quadlet+j2); .hidden/ skipped
    yield llm
    db.execute(text("TRUNCATE chunks, documents, query_log"))
    db.commit()


@pytest.fixture(autouse=True)
def _admin_signed_in(client: TestClient) -> None:
    """Phase 79 (task 03): ``POST /api/chat`` is user-gated — every turn
    in this module runs as the signed-in ADMIN, so the shared ``client``
    logs in once per test (the TestClient cookie jar carries the session
    for every request of the test). The anonymous 401 contract itself is
    pinned in ``test_auth_api.py``."""
    r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"


def _stream_chat(client: TestClient, message: str) -> tuple[int, str, list[dict[str, Any]]]:
    with client.stream("POST", "/api/chat", json={"message": message}) as r:
        assert r.status_code == 200
        assert r.headers["content-type"].startswith("text/event-stream")
        buf = ""
        frames: list[dict[str, Any]] = []
        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() == "", "stream must end on a frame boundary"
    return r.status_code, r.headers["content-type"], frames


def test_chat_streams_deltas_then_done_with_sources(client, db, seeded_kb: FakeRagLLM) -> None:
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    deltas = [f for f in frames if f.get("type") == "delta"]
    assert len(deltas) >= 2  # genuinely streamed
    assert "".join(d["text"] for d in deltas) == seeded_kb.answer
    assert not any(f.get("type") == "error" for f in frames)

    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"] == []
    sources = done[0]["sources"]
    assert sources, "done must carry the cited sources"
    assert sources[0]["path"] == "homelab/kubernetes.md"
    assert sources[0]["source"] == "docs"
    assert sources[0]["title"] == "Kubernetes Homelab Cluster"

    # The LLM received the locked HIGH prompt with the FULL document text.
    (system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
    assert user["content"] == QUESTION
    assert "<relevance>HIGH</relevance>" in system["content"]
    assert "DEFLECT_MODE" not in system["content"]
    assert "<documents>" in system["content"]
    assert "Talos Linux" in system["content"]  # full doc, not just the chunk
    assert "HONESTY GATE" in system["content"]


def test_chat_streams_thinking_before_deltas(client, db, seeded_kb: FakeRagLLM) -> None:
    """Phase 17: ``thinking`` frames precede every ``delta`` frame and
    reassemble to the model's reasoning; the ``done`` contract is
    unchanged."""
    thinker = FakeRagLLM(
        thinking=(
            "Step 1: parse the question. Step 2: check the kubernetes doc. "
            "Step 3: name Talos, Cilium, three nodes. Step 4: answer."
        )
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: thinker
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    thinking = [f for f in frames if f.get("type") == "thinking"]
    deltas = [f for f in frames if f.get("type") == "delta"]
    assert len(thinking) >= 1  # genuinely streamed
    assert len(deltas) >= 2
    # Every thinking frame precedes every delta frame.
    ordered = [f["type"] for f in frames if f["type"] in ("thinking", "delta")]
    assert ordered == ["thinking"] * len(thinking) + ["delta"] * len(deltas)
    assert all(set(f.keys()) == {"type", "text"} for f in thinking)
    assert "".join(f["text"] for f in thinking) == thinker.thinking
    assert "".join(d["text"] for d in deltas) == thinker.answer

    # Done still last; sources unchanged by the thinking extension.
    done = frames[-1]
    assert done["type"] == "done"
    assert done["deflected"] is False
    assert done["suggestions"] == []
    assert done["sources"][0]["path"] == "homelab/kubernetes.md"
    assert done["sources"][0]["source"] == "docs"
    assert not any(f.get("type") == "error" for f in frames)


def test_chat_thinking_suppressed_when_disabled(
    client, db, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Phase 17 kill-switch: ``BOR_STREAM_THINKING=0`` drops every
    ``thinking`` frame; the delta stream is byte-identical to the
    thinking-free case."""
    thinker = FakeRagLLM(thinking="hidden reasoning that must never reach the wire")
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: thinker
    # Same honesty gate the conftest/module already use (mock-calibrated
    # 0.30 from the environment) — only the kill-switch changes.
    live = get_settings()
    monkeypatch.setattr(
        chat_api,
        "get_settings",
        lambda: Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            relevance_threshold=live.relevance_threshold,
            stream_thinking=False,
        ),
    )
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert not any(f.get("type") == "thinking" for f in frames)
    deltas = [f for f in frames if f.get("type") == "delta"]
    assert "".join(d["text"] for d in deltas) == thinker.answer
    assert frames[-1]["type"] == "done"
    assert not any(f.get("type") == "error" for f in frames)


def test_chat_writes_query_log_row(client, db, seeded_kb: FakeRagLLM) -> None:
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    rows = db.scalars(select(QueryLog)).all()
    assert len(rows) == 1
    row = rows[0]
    assert row.question == QUESTION
    assert row.deflected is False
    total_chunks = db.scalar(select(func.count()).select_from(Chunk))
    # chunk_hits is the fused candidate set (cosine top-N ∪ FTS top-N).
    assert 1 <= row.chunk_hits <= total_chunks
    assert row.top_score > 0.0  # genuine token-overlap cosine, best hit
    assert row.top_score <= 1.0
    assert "docs/homelab/kubernetes.md" in row.sources
    assert row.latency_ms >= 0
    # Why the gate answered (A8 revised): cosine over the threshold OR a
    # lexical hit. The mock-calibrated threshold (0.30, see tests/conftest.py)
    # makes the cosine branch true here; the FTS branch is covered too —
    # "kubernetes" / "cluster" match the doc's tsvector.
    thr = get_settings().relevance_threshold
    assert row.top_score >= thr or (row.fts_hits or 0) > 0
    assert (row.fts_hits or 0) >= 1  # the lexical branch really fired


def test_off_topic_question_deflects_honestly(client, db, seeded_kb: FakeRagLLM) -> None:
    """Phase 04 contract: weak retrieval ⇒ honest deflection, no fake answer."""
    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 not any(f.get("type") == "error" for f in frames)
    deltas = [f for f in frames if f.get("type") == "delta"]
    assert len(deltas) >= 2  # the LLM is still called (voice stays chippy)
    done = frames[-1]
    assert done["type"] == "done"
    assert done["deflected"] is True
    # 2-3 alternative chips, all non-empty, derived from real titles/topics.
    assert 2 <= len(done["suggestions"]) <= 3
    assert all(s.strip() for s in done["suggestions"])
    assert any(
        "Deploying a New Service" in s for s in done["suggestions"]
    ), "the best weak-hit title must be offered as a chip"
    # 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. (Pre-phase: they rode the wire as sources.)
    assert done["sources"] == []

    # The LLM saw the LOW prompt: DEFLECT_MODE + titles, never doc content.
    (system, user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
    assert user["content"] == OFF_TOPIC
    assert "<relevance>LOW</relevance>" in system["content"]
    assert "DEFLECT_MODE" in system["content"]
    assert "HONESTY GATE" in system["content"]
    assert "Talos Linux" not in system["content"]  # full doc content never sent
    assert "<documents>" not in system["content"]

    # Durable record: deflected=true + the weak top_score. Deflection is
    # only reached when the cosine is under the threshold AND no chunk
    # FTS-matches the question — so fts_hits must be zero here.
    row = db.scalars(select(QueryLog)).one()
    assert row.question == OFF_TOPIC
    assert row.deflected is True
    assert 0.0 < row.top_score < get_settings().relevance_threshold
    assert row.fts_hits == 0
    assert row.chunk_hits >= 1
    # The retrieval stays durably recorded for threshold tuning
    # (observability unchanged — query_log records retrieval, not
    # citations; the done frame's [] above is the citation surface).
    assert row.sources


def test_done_frame_carries_related_tier_on_grounded_turn(
    client, db, seeded_kb: FakeRagLLM
) -> None:
    """Phase 113 (LOCKED A4): a grounded turn's done frame carries the
    related tier — the ranked docs beyond the cited ceiling, capped at
    ``related_max_docs`` (2), disjoint from the cited list. The durable
    record keeps the FULL retrieval (cited + related, LOCKED A3)."""
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    done = frames[-1]
    assert done["deflected"] is False
    sources = {(s["source"], s["path"]) for s in done["sources"]}
    related = done["related"]
    assert related, "the 2nd-and-lower scored docs ride the related tier"
    assert len(related) <= get_settings().related_max_docs
    # The related tier never overlaps the cited list (the dedupe is by
    # (source, path) — the same pattern as the cited docs).
    related_keys = {(s["source"], s["path"]) for s in related}
    assert sources.isdisjoint(related_keys)
    # Rank order: the cited top-2 are the kubernetes doc and the template;
    # the next ranked doc is the ssh aliases file.
    assert related[0]["path"] == "homelab/ssh/ssh_aliases.txt"
    # Every ref carries the chip identity fields (the UI row reuses them).
    assert all(s["title"] for s in related)

    # Durable record: the full retrieval (cited + related) is logged.
    row = db.scalars(select(QueryLog)).one()
    assert "docs/homelab/ssh/ssh_aliases.txt" in row.sources
    assert "docs/homelab/kubernetes.md" in row.sources


def test_deflected_done_frame_carries_weak_hits_in_related(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Phase 113: on a deflected turn nothing clears the bar — the cited
    tier is empty (done.sources stays [], the phase-112 contract) and the
    weak hits fall to the related tier (their visibility home). The
    durable record still carries the retrieval (LOCKED A3)."""
    monkeypatch.setenv("BOR_SOURCE_USEFULNESS_FLOOR", "0.20")
    get_settings.cache_clear()
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, OFF_TOPIC)
        done = frames[-1]
        assert done["deflected"] is True
        assert done["sources"] == []  # a deflected answer cites nothing
        # The weak hits (the sourdough question's best mock cosines are
        # ~0.11/0.04 — both below the 0.20 bar) ride the related tier, in
        # rank order, capped at related_max_docs.
        related = done["related"]
        assert len(related) <= get_settings().related_max_docs
        assert [s["path"] for s in related][:2] == [
            "deployments/new-service.md",
            "homelab/quadlet/lan.network",
        ]
        assert all(s["title"] for s in related)
        assert done["suggestions"]  # the "Maybe try" chips are unchanged

        # Durable record: the weak retrieval stays logged for tuning (A3).
        row = db.scalars(select(QueryLog)).one()
        assert row.deflected is True
        assert row.sources  # the weak-hit paths, for threshold tuning
    finally:
        # The cache clear is LAST — an assertion that calls get_settings()
        # after the clear would re-populate the lru_cache with the
        # monkeypatched value and leak it into the next test.
        fastapi_app.dependency_overrides.clear()
        get_settings.cache_clear()


def test_related_doc_read_by_agent_is_cited_not_related(
    client, db, seeded_kb: FakeRagLLM
) -> None:
    """Phase 113 × phase 37: an agent-read doc is a citation by definition
    — when the agent ``read``s a doc that would otherwise ride the related
    tier, it joins done.sources (deduped, last) and is EXCLUDED from
    done.related (a "nearby doc" that was actually used must not read as
    nearby)."""
    scripted = FakeRagLLM(
        tool_script=[
            [
                ToolCallPiece(
                    id="call_1",
                    name="read",
                    arguments={"path": "docs/homelab/ssh/ssh_aliases.txt"},
                )
            ]
        ]
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    done = frames[-1]
    assert done["deflected"] is False
    sources = [(s["source"], s["path"]) for s in done["sources"]]
    assert sources[-1] == ("docs", "homelab/ssh/ssh_aliases.txt")  # read ⇒ cited
    related = [(s["source"], s["path"]) for s in done["related"]]
    assert ("docs", "homelab/ssh/ssh_aliases.txt") not in related
    assert set(sources).isdisjoint(set(related))
    # The OTHER related-tier doc (gitlab, rank 4) stays in the tier.
    assert ("docs", "homelab/container_gitlab/gitlab.md") in related


def test_keyword_question_grounded_by_lexical_hit_despite_weak_cosine(
    client, db, seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 09: a name-your-tool question the vector model barely ranks
    ("kafkabridge" only appears in static-dns.json) must still be grounded
    via the FTS branch — HIGH when cosine >= lexical_support_floor AND
    fts_hits > 0 (A8 revised 2026-09-14).

    The conftest floor (0.15) is above the mock's cosine (~0.134), so we
    lower the floor here so the corroborated-lexical path fires."""
    from app.config import get_settings  # noqa: E402

    monkeypatch.setenv("BOR_LEXICAL_SUPPORT_FLOOR", "0.10")
    # get_settings is lru_cached — clear the cache so the new env var takes effect.
    get_settings.cache_clear()
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, "How does kafkabridge work?")
        done = frames[-1]
        assert done["type"] == "done"
        assert done["deflected"] is False  # weak cosine, but a lexical hit
        assert done["suggestions"] == []
        sources = done["sources"]
        assert sources and sources[0]["path"] == "homelab/networking/static-dns.json"

        (system, _user) = seeded_kb.seen_messages[0][0], seeded_kb.seen_messages[0][1]
        assert "<relevance>HIGH</relevance>" in system["content"]  # grounded prompt

        row = db.scalars(select(QueryLog)).one()
        assert row.deflected is False
        assert row.top_score < get_settings().relevance_threshold  # weak vector score
        assert (row.fts_hits or 0) >= 1  # …and it is the FTS hit that grounds it
        assert "docs/homelab/networking/static-dns.json" in row.sources
    finally:
        # The cache clear is LAST — an assertion that calls get_settings()
        # after the clear would re-populate the lru_cache with the
        # monkeypatched value and leak it into the next test.
        fastapi_app.dependency_overrides.clear()
        get_settings.cache_clear()


def test_chat_empty_kb_streams_empty_sources(client, db) -> None:
    db.execute(text("TRUNCATE chunks, documents, query_log"))
    db.commit()
    llm = FakeRagLLM()
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: llm
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    # Nothing retrieved ⇒ nothing to pretend to know: honest deflection.
    done = frames[-1]
    assert done["type"] == "done"
    assert done["deflected"] is True
    assert done["sources"] == []
    assert 2 <= len(done["suggestions"]) <= 3  # onboarding fallback chips
    (system, _user) = llm.seen_messages[0][0], llm.seen_messages[0][1]
    assert "DEFLECT_MODE" in system["content"]
    assert "nothing close at all" in system["content"]
    row = db.scalars(select(QueryLog)).one()
    assert row.deflected is True
    assert row.top_score == 0.0
    assert row.chunk_hits == 0
    assert row.sources == ""


def test_chat_embed_failure_yields_error_event(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Phase 67: a dead embeddings endpoint retries on the configured
    budget — one ``retry`` frame per restart (the attempt about to be
    tried, 1-based) — and settles on the existing terminal error frame;
    no query_log row. Zero delay keeps the exhaustion path fast."""
    broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    retries = live.llm_retries
    assert [f["type"] for f in frames] == ["retry"] * retries + ["error"]
    assert [f["attempt"] for f in frames if f["type"] == "retry"] == list(
        range(2, retries + 2)
    )
    assert all(
        f["max_attempts"] == retries + 1 for f in frames if f["type"] == "retry"
    )
    assert "embedding" in frames[-1]["detail"]
    assert db.scalars(select(QueryLog)).all() == []


def test_chat_mid_stream_failure_yields_error_after_partial_deltas(client, db) -> None:
    broken = FakeRagLLM(fail_mid_stream=True)
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert [f["type"] for f in frames] == ["delta", "error"]
    assert "dropped the connection" in frames[1]["detail"]
    # No done event, no log row for a turn that never completed.
    assert db.scalars(select(QueryLog)).all() == []


def test_error_event_matches_contract_shape(
    client, db, seeded_kb, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The SSE error event (PLAN §4) is exactly ``{type, detail}`` — the
    client's loading-feedback state machine (phase 06) keys off this shape
    to flip to the error state and re-enable the send button.
    ``llm_retries=0`` keeps this a single-attempt turn: the contract under
    test is the error frame itself, not the phase-67 retry loop."""
    broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=0)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert len(frames) == 1
    event = frames[0]
    assert set(event.keys()) == {"type", "detail"}
    assert event["type"] == "error"
    assert isinstance(event["detail"], str) and event["detail"]


def test_chat_db_down_returns_503_json(client, monkeypatch) -> None:
    monkeypatch.setattr(chat_api, "db_available", lambda: False)
    r = client.post("/api/chat", json={"message": "hello"})
    assert r.status_code == 503
    assert "offline" in r.json()["detail"]


def test_chat_retrieval_failure_yields_error_event(client, db, seeded_kb, monkeypatch) -> None:
    def boom(*_a: Any, **_k: Any) -> Any:
        raise RuntimeError("db exploded")

    monkeypatch.setattr(chat_api, "retrieve", boom)
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert [f["type"] for f in frames] == ["error"]
    assert "offline mid-question" in frames[0]["detail"]
    assert db.scalars(select(QueryLog)).all() == []


class _BrokenCommitSession:
    """Pass-through session whose ``commit()`` raises (query_log failure)."""

    def __init__(self, real: Any) -> None:
        self._real = real

    def __enter__(self) -> _BrokenCommitSession:
        return self

    def __exit__(self, *args: Any) -> None:
        self._real.close()

    def commit(self) -> None:
        raise RuntimeError("query_log commit failed")

    def __getattr__(self, name: str) -> Any:
        return getattr(self._real, name)


def test_chat_query_log_failure_still_sends_done(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """SEC-14-04: even when the query_log write fails, the answer still
    goes out. The chat endpoint uses short-lived sessions (SessionLocal)
    for query_log writes — monkeypatch SessionLocal to return a broken
    session that fails on commit."""
    from app.db import SessionLocal as real_SessionLocal

    def broken_session_factory():
        real = real_SessionLocal()
        return _BrokenCommitSession(real)

    monkeypatch.setattr(chat_api, "SessionLocal", broken_session_factory)
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    # The answer (and the done event) went out despite the log-row failure.
    assert [f["type"] for f in frames if f["type"] == "delta"]
    assert frames[-1]["type"] == "done"
    assert frames[-1]["deflected"] is False


# ---------- phase 37: agent document tools on grounded turns ----------


async def _collect_run_agent(
    llm: FakeRagLLM,
    db: Session,
    system_prompt: str,
    settings: Settings,
    seed_docs: list[Document],
) -> tuple[list[Any], agent.AgentHolder]:
    """Consume one ``run_agent`` turn, returning the yielded pieces (in
    order) and the holder. Phase 95 (task 01): the direct agent-loop
    drive — the agent-loop yield order on the real prompt path, the
    complement of the endpoint-level ``tool_result`` SSE tests below
    (task 02)."""
    holder = agent.AgentHolder()
    pieces: list[Any] = []
    async for piece in agent.run_agent(
        llm,  # pyright: ignore[reportArgumentType]  # duck-typed LLMClient
        lambda: db,  # SEC-14-04: session factory (integration tests reuse the fixture session)
        system_prompt=system_prompt,
        user_message=QUESTION,
        seed_docs=seed_docs,
        settings=settings,
        holder=holder,
    ):
        pieces.append(piece)
    return pieces, holder


def test_read_cap_truncates_and_yields_tool_result_on_real_prompt_path(
    db,
) -> None:
    """Phase 95 (task 01): on the REAL prompt path (a real Postgres
document + the real ``build_high_prompt``), a ``read`` of a document
LONGER than ``settings.read_max_chars`` truncates the result the model
sees — first ``cap`` chars + the shared :data:`TRUNCATION_MARKER` + the
pinned grep-pointer notice — and ``run_agent`` yields exactly ONE
``ToolResultPiece``: AFTER the read's ``tool`` frame (the matching
``ToolCallPiece``) and BEFORE the next model round. The endpoint-level
``tool_result`` SSE frame is asserted separately below (task 02); this
pins the agent-loop yield order on the real prompt path."""
    cap = 100
    content = "K" * (cap + 40)  # 40 chars over the cap
    doc = Document(
        id=uuid.uuid4(),
        source="docs",
        path="big.md",
        full_path="/tmp/big.md",
        title="Big Doc",
        content=content,
        content_hash="1" * 64,
        created_at=_FIXTURE_CREATED_AT,
    )
    db.add(doc)
    db.commit()
    try:
        # The real prompt path: the actual HIGH prompt for the one doc.
        system_prompt = build_high_prompt([doc])
        settings = Settings(_env_file=None, read_max_chars=cap)  # pyright: ignore[reportCallIssue]
        scripted = FakeRagLLM(
            tool_script=[
                [
                    ToolCallPiece(
                        id="call_1", name="read", arguments={"path": "docs/big.md"}
                    )
                ]
                # the answer request (tools still offered, script
                # exhausted) falls back to the thinking + answer stream
            ]
        )
        pieces, holder = asyncio.run(
            _collect_run_agent(scripted, db, system_prompt, settings, seed_docs=[])
        )
        # The model's context carried the truncated read — the first cap
        # chars, then the shared marker + the pinned grep-pointer notice
        # (so a downstream grep is the model's path to the rest). The
        # fake records the (mutated-in-place) messages list, so the same
        # tool message is aliased across requests — they all carry the
        # same content; take the last.
        tool_msgs = [
            m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
        ]
        assert tool_msgs, "the executed read must be appended as a tool message"
        body = tool_msgs[-1]["content"]
        # Phase 106, D5: the date rides every read — the SECOND line
        # (first line byte-identical — the mock's header contract).
        assert body.startswith(
            "Document docs/big.md:\ndate: 2024-06-15\n" + content[:cap]
        )
        assert TRUNCATION_MARKER in body
        assert (
            READ_TRUNCATION_NOTICE.format(shown=cap, total=len(content)) in body
        )
        # The yield order: the read's ToolCallPiece, then the ONE
        # ToolResultPiece, then the next round's answer content.
        kinds: list[str] = []
        for p in pieces:
            if isinstance(p, ToolCallPiece):
                kinds.append("toolcall")
            elif isinstance(p, ToolResultPiece):
                kinds.append("toolresult")
            elif isinstance(p, StreamPiece):
                kinds.append(p.kind)
        assert kinds.count("toolresult") == 1
        assert kinds.index("toolcall") < kinds.index("toolresult")
        assert kinds.index("toolresult") < kinds.index("content")
        # The piece carries (argument, shown, total) — the raw
        # source/path the model passed (what the tool frame carries), the
        # cap kept, the true length.
        (result_piece,) = [p for p in pieces if isinstance(p, ToolResultPiece)]
        assert result_piece.name == "read"
        assert result_piece.argument == "docs/big.md"
        assert result_piece.truncated is True
        assert result_piece.chars_shown == cap
        assert result_piece.chars_total == len(content)
        # Holder accounting: a truncated read is still a SUCCESSFUL call
        # (counted + added to context); the tuple is the signal only.
        assert holder.tool_calls == 1
        assert holder.read_docs == [doc]
        assert holder.read_truncations == [("docs/big.md", cap, len(content))]
    finally:
        db.delete(doc)
        db.commit()


def test_read_at_or_under_cap_yields_no_tool_result_on_real_prompt_path(
    db,
) -> None:
    """Phase 95 (task 01): the complement — a ``read`` of a document at or
under the cap on the real prompt path is byte-identical to the
pre-phase-95 agent loop: NO ``ToolResultPiece``, no holder entry, no
marker in the model's context."""
    cap = 100
    content = "K" * cap  # exactly at the cap → fits, not truncated
    doc = Document(
        id=uuid.uuid4(),
        source="docs",
        path="fits.md",
        full_path="/tmp/fits.md",
        title="Fits Doc",
        content=content,
        content_hash="2" * 64,
        created_at=_FIXTURE_CREATED_AT,
    )
    db.add(doc)
    db.commit()
    try:
        system_prompt = build_high_prompt([doc])
        settings = Settings(_env_file=None, read_max_chars=cap)  # pyright: ignore[reportCallIssue]
        scripted = FakeRagLLM(
            tool_script=[
                [
                    ToolCallPiece(
                        id="call_1", name="read", arguments={"path": "docs/fits.md"}
                    )
                ]
            ]
        )
        pieces, holder = asyncio.run(
            _collect_run_agent(scripted, db, system_prompt, settings, seed_docs=[])
        )
        # No ToolResultPiece, no holder entry.
        assert not any(isinstance(p, ToolResultPiece) for p in pieces)
        assert holder.read_truncations == []
        # The model's context is the whole document, the pre-phase-95
        # read result plus the phase-106 D5 date line (no marker, no
        # notice). (The fake aliases the mutated messages list, so
        # take the last tool msg.)
        tool_msgs = [
            m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
        ]
        assert tool_msgs, "the executed read must be appended as a tool message"
        assert (
            tool_msgs[-1]["content"]
            == "Document docs/fits.md:\ndate: 2024-06-15\n" + content
        )
        assert TRUNCATION_MARKER not in tool_msgs[-1]["content"]
        # Still a successful read.
        assert holder.tool_calls == 1
        assert holder.read_docs == [doc]
    finally:
        db.delete(doc)
        db.commit()


def _insert_big_doc(db, content: str) -> Document:
    """One bare ``documents`` row (no chunks — the ``read`` lookup is a
    (source, path) identity match, not a retrieval) for the SSE-level
    read-cap tests: a document the model can only reach through the
    ``read`` tool."""
    doc = Document(
        id=uuid.uuid4(),
        source="docs",
        path="big-read.md",
        full_path="/tmp/big-read.md",
        title="Big Read Doc",
        content=content,
        content_hash="3" * 64,
        created_at=_FIXTURE_CREATED_AT,
    )
    db.add(doc)
    db.commit()
    return doc


def test_truncated_read_streams_tool_result_frame_after_tool_frame(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95 (task 02, the A15 extension): a grounded turn whose
    scripted ``read`` hits a document LONGER than ``read_max_chars``
    (the cap lowered via the settings override — the task 01
    ``Settings(_env_file=None, read_max_chars=…)`` pattern) streams the
    ``tool`` → ``tool_result`` → ``delta…`` → ``done`` sequence: EXACTLY
    ONE ``tool_result`` frame, AFTER the matching ``tool`` frame (the
    line is already on screen) and BEFORE the next round's first frame,
    with the right shape and counts (``chars_shown`` = the cap,
    ``chars_total`` = the true length). The model's context carried the
    truncated read (marker + pinned grep-pointer notice); the read is
    still cited (a truncated read is a successful call)."""
    cap = 100
    content = "K" * (cap + 150)
    doc = _insert_big_doc(db, content)
    live = get_settings()
    monkeypatch.setattr(
        chat_api,
        "get_settings",
        lambda: Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            relevance_threshold=live.relevance_threshold,
            read_max_chars=cap,
        ),
    )
    scripted = FakeRagLLM(
        tool_script=[
            [
                ToolCallPiece(
                    id="call_1", name="read", arguments={"path": "docs/big-read.md"}
                )
            ]
            # the answer request still carries the tools (1 round < the
            # default cap of 10); the script is exhausted, so the fake
            # falls back to the thinking + answer stream
        ]
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()
        db.delete(doc)
        db.commit()

    types = [f["type"] for f in frames]
    assert "error" not in types
    tool_i = types.index("tool")
    tool_result_i = types.index("tool_result")
    # Exactly one tool_result frame…
    assert types.count("tool_result") == 1
    # …AFTER the matching tool frame and BEFORE the next model round's
    # first frame (the answer's deltas): tool → tool_result → delta…
    assert tool_i + 1 == tool_result_i
    assert tool_result_i < min(i for i, t in enumerate(types) if t == "delta")
    # The frame's exact shape: the additive seventh event type carries
    # the name/argument of the matching tool frame + the counts.
    frame = frames[tool_result_i]
    assert set(frame) == {
        "type",
        "name",
        "argument",
        "truncated",
        "chars_shown",
        "chars_total",
    }
    assert frame["name"] == frames[tool_i]["name"] == "read"
    assert frame["argument"] == frames[tool_i]["argument"] == "docs/big-read.md"
    assert frame["truncated"] is True
    assert frame["chars_shown"] == cap  # the cap kept
    assert frame["chars_total"] == len(content)  # the true length
    # The LLM's context carried the honest truncation: first cap chars +
    # the shared marker + the pinned grep-pointer notice (the fake
    # aliases the mutated messages list — take the last tool msg).
    tool_msgs = [
        m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
    ]
    assert tool_msgs
    body = tool_msgs[-1]["content"]
    # Phase 106, D5: the date rides every read — the SECOND line.
    assert body.startswith(f"Document docs/big-read.md:\ndate: 2024-06-15\n{content[:cap]}")
    assert TRUNCATION_MARKER in body
    assert READ_TRUNCATION_NOTICE.format(shown=cap, total=len(content)) in body
    # The truncated read is still a SUCCESSFUL call — cited in done.
    done = frames[-1]
    assert done["type"] == "done" and done["deflected"] is False
    assert ("docs", "big-read.md") in [(s["source"], s["path"]) for s in done["sources"]]
    assert ("docs", "homelab/kubernetes.md") in [
        (s["source"], s["path"]) for s in done["sources"]
    ]


def test_untruncated_read_streams_no_tool_result_frame(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95 (task 02): the complement at the SSE level — a ``read``
    of a document AT OR UNDER the cap (the same long document, cap
    raised past its true length) streams NO ``tool_result`` frame (one
    frame = one noteworthy event; the six pre-existing event types are
    byte-identical), the ``tool`` frame is unchanged, and the model's
    context is the WHOLE document (no marker, no notice)."""
    content = "K" * 250
    doc = _insert_big_doc(db, content)
    live = get_settings()
    monkeypatch.setattr(
        chat_api,
        "get_settings",
        lambda: Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            relevance_threshold=live.relevance_threshold,
            read_max_chars=10_000,  # far over the doc's true length
        ),
    )
    scripted = FakeRagLLM(
        tool_script=[
            [
                ToolCallPiece(
                    id="call_1", name="read", arguments={"path": "docs/big-read.md"}
                )
            ]
        ]
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()
        db.delete(doc)
        db.commit()

    types = [f["type"] for f in frames]
    assert "error" not in types
    assert types.count("tool_result") == 0  # one frame = one noteworthy event
    assert types.count("tool") == 1
    (tool_frame,) = [f for f in frames if f["type"] == "tool"]
    assert set(tool_frame) == {"type", "name", "argument"}  # byte-identical shape
    assert tool_frame["argument"] == "docs/big-read.md"
    assert frames[-1]["type"] == "done" and frames[-1]["deflected"] is False
    # The model saw the WHOLE document — no marker, no notice.
    tool_msgs = [
        m for r in scripted.seen_messages for m in r if m.get("role") == "tool"
    ]
    assert tool_msgs
    assert (
        tool_msgs[-1]["content"]
        == "Document docs/big-read.md:\ndate: 2024-06-15\n" + content
    )
    assert TRUNCATION_MARKER not in tool_msgs[-1]["content"]


def test_grounded_turn_streams_tool_frames_and_cites_read_doc(
    client, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
    """(a) Grounded turn with tool calls: the event sequence is
    ``thinking?/tool/tool/delta…/done``; ``done.sources`` and the
    ``query_log`` row include the read document (deduped, order
    preserved); the per-turn log line carries ``tool_calls=2``.
    Phase 45: the agent loop keeps offering the tools for the whole
    turn — the round cap (not per-tool budgets) is the bound."""
    scripted = FakeRagLLM(
        tool_script=[
            [
                StreamPiece("thinking", "Let me list what is indexed…"),
                ToolCallPiece(id="call_1", name="ls", arguments={}),
            ],
            [
                ToolCallPiece(
                    id="call_2",
                    name="read",
                    arguments={"path": "docs/homelab/backups.md"},
                )
            ],
            # the answer request still carries the tools (2 rounds < the
            # default cap of 10); the fake's tool_script is exhausted, so
            # it falls back to the thinking + answer stream
        ]
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    types = [f["type"] for f in frames]
    assert types[0] == "thinking"
    assert types[1] == "tool" and types[2] == "tool"  # the two executed calls
    assert "error" not in types
    assert types[3:-1] == ["delta"] * (len(types) - 4)  # deltas, then done last
    assert frames[-1]["type"] == "done"

    list_frame, read_frame = frames[1], frames[2]
    assert set(list_frame) == {"type", "name", "argument"}
    assert list_frame["name"] == "ls"
    assert list_frame["argument"] is None  # no ``path`` argument was passed
    assert set(read_frame) == {"type", "name", "argument"}
    assert read_frame["name"] == "read"
    # Phase 70: the frame's argument is the single string the model
    # passed — the combined ``source/path``.
    assert read_frame["argument"] == "docs/homelab/backups.md"

    deltas = [f for f in frames if f["type"] == "delta"]
    assert len(deltas) >= 2  # genuinely streamed
    assert "".join(d["text"] for d in deltas) == scripted.answer

    done = frames[-1]
    assert done["deflected"] is False
    # done.sources = the retrieval docs + the read doc, deduped, order kept.
    sources = [(s["source"], s["path"]) for s in done["sources"]]
    assert sources[-1] == ("docs", "homelab/backups.md")  # the read doc is cited
    assert ("docs", "homelab/kubernetes.md") in sources  # …after the retrieval docs
    assert len(sources) == len(set(sources))  # deduped by (source, path)
    assert done["sources"][-1]["title"] == "Backup Strategy"

    # Phase 45: the tools stay offered on every request — the round cap
    # (not spent budgets) bounds the loop, and the model answered while
    # still being offered the tools (2 rounds < default cap 10).
    assert len(scripted.seen_messages) == 3
    assert scripted.seen_tools[0] == AGENT_TOOLS
    assert scripted.seen_tools[1] == AGENT_TOOLS
    assert scripted.seen_tools[2] == AGENT_TOOLS

    # The query_log row carries the same combined source list.
    (row,) = db.scalars(select(QueryLog)).all()
    assert row.deflected is False
    assert "docs/homelab/kubernetes.md" in row.sources
    assert row.sources.endswith(", docs/homelab/backups.md")  # the read doc, last

    # The required per-turn log line (PLAN §9 extension) counts both calls
    # and lists the combined sources (retrieval + read).
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "tool_calls=2" in lines[-1]
    assert "'docs/homelab/kubernetes.md'" in lines[-1]
    assert "'docs/homelab/backups.md'" in lines[-1]
    assert "scaffold_stripped=0" in lines[-1]  # phase 71: uniform clean-turn field


def test_grounded_turn_streams_grep_tool_frames(
    client, db, seeded_kb: FakeRagLLM
) -> None:
    """Phase 68 (renamed ``grep`` in phase 70): a scripted ``grep`` call
    streams as ``{type: "tool", name: "grep", argument: <pattern>}`` —
    the raw pattern is the frame's ``argument`` (the UI renders the
    "searching for" line from it). A non-string pattern — a model error
    the backend refuses — yields ``argument: null``. A grep adds no
    source: ``done.sources`` stays the retrieval docs (locked A5)."""
    scripted = FakeRagLLM(
        tool_script=[
            [
                ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "Cilium"}),
            ],
            [
                ToolCallPiece(
                    id="call_2",
                    name="grep",
                    arguments={"pattern": 42},  # model error: non-string
                ),
            ],
            # the answer request still carries the tools (2 rounds < the
            # default cap of 10); the fake's tool_script is exhausted, so
            # it falls back to the thinking + answer stream
        ]
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    types = [f["type"] for f in frames]
    assert "error" not in types
    assert len(scripted.seen_tools) == 3  # both greps executed (rounds)

    tool_frames = [f for f in frames if f["type"] == "tool"]
    assert len(tool_frames) == 2
    first, second = tool_frames
    assert set(first) == {"type", "name", "argument"}
    assert first["name"] == "grep"
    assert first["argument"] == "Cilium"  # the raw pattern
    assert set(second) == {"type", "name", "argument"}
    assert second["name"] == "grep"
    assert second["argument"] is None  # the non-string pattern → null

    # The greps still answered: deltas, then a grounded done.
    assert [f for f in frames if f["type"] == "delta"]
    done = frames[-1]
    assert done["type"] == "done" and done["deflected"] is False
    paths = [s["path"] for s in done["sources"]]
    assert "homelab/kubernetes.md" in paths  # retrieval docs, unchanged
    assert "homelab/backups.md" not in paths  # a grep adds no source


def test_tool_frames_carry_the_model_arguments_regardless_of_execution(
    client, db, seeded_kb: FakeRagLLM, caplog: pytest.LogCaptureFixture
) -> None:
    """Phase 70 pins: the frame's ``argument`` is the single string
    argument the model passed — an ``ls`` frame carries the scope when
    the model gave one (null only when it is omitted, pinned above) —
    and frame emission is execution-independent: a rejected call (an
    unknown ``read`` path) still streams its frame with the model's
    argument as-is. The rejected read adds no source (``done.sources``
    stays the retrieval docs), and rejected calls count nothing
    (``tool_calls=1`` — only the executed scoped ``ls``)."""
    # The scoped ``ls`` source-name check reads the registry — insert a
    # row resolving to ``docs`` (the fixture's source name) and delete
    # it again afterwards.
    src = GitSource(url="https://github.com/reese/docs.git", kind="git")
    db.add(src)
    db.commit()
    try:
        scripted = FakeRagLLM(
            tool_script=[
                [ToolCallPiece(id="call_1", name="ls", arguments={"path": "docs"})],
                [
                    ToolCallPiece(
                        id="call_2", name="read", arguments={"path": "docs/homelab/nope.md"}
                    )
                ],
            ]
        )
        fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
        try:
            caplog.set_level(logging.INFO, logger="app.chat")
            _, _, frames = _stream_chat(client, QUESTION)
        finally:
            fastapi_app.dependency_overrides.clear()
    finally:
        db.execute(delete(GitSource).where(GitSource.id == src.id))
        db.commit()

    types = [f["type"] for f in frames]
    assert "error" not in types
    # Both calls stream a frame — the rejected read included.
    tool_frames = [f for f in frames if f["type"] == "tool"]
    assert len(tool_frames) == 2
    ls_frame, read_frame = tool_frames
    assert set(ls_frame) == {"type", "name", "argument"}
    assert ls_frame["name"] == "ls"
    assert ls_frame["argument"] == "docs"  # the model's scope, as passed
    assert set(read_frame) == {"type", "name", "argument"}
    assert read_frame["name"] == "read"
    # The rejected call's frame still carries the model's argument as
    # passed — frame emission is execution-independent.
    assert read_frame["argument"] == "docs/homelab/nope.md"

    # The rejected read adds no source — done.sources stays retrieval.
    done = frames[-1]
    assert done["type"] == "done" and done["deflected"] is False
    paths = [s["path"] for s in done["sources"]]
    assert "homelab/kubernetes.md" in paths  # retrieval docs, unchanged
    assert "homelab/nope.md" not in paths  # the refused read cites nothing

    # The rejected call counts nothing — only the executed scoped ls.
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "tool_calls=1" in lines[-1]


def test_deflected_turn_stays_byte_identical_without_tools(
    client, db, seeded_kb: FakeRagLLM
) -> None:
    """(b) Deflected turn: the agent loop never runs — no ``tool``
    frames, and the frame sequence is byte-identical to the plain fake's
    direct-``chat_stream`` output even for a fake scripted to call tools
    (its script is never consumed). The LLM was called once, without a
    ``tools`` key."""
    scripted = FakeRagLLM(
        tool_script=[
            [ToolCallPiece(id="call_1", name="ls", arguments={})],
            [
                ToolCallPiece(
                    id="call_2",
                    name="read",
                    arguments={"path": "docs/homelab/backups.md"},
                )
            ],
            [StreamPiece("content", "never used — the agent never runs")],
        ]
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, baseline = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        _, _, frames = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames == baseline  # byte-identical to the direct path
    assert not any(f["type"] == "tool" for f in frames)
    assert frames[-1]["type"] == "done" and frames[-1]["deflected"] is True
    assert len(scripted.tool_script) == 3  # the script was never consumed
    assert len(scripted.seen_messages) == 1
    assert scripted.seen_tools == [None]  # one request, no tools key

    # The read document never sneaks into the deflected turn's record.
    (row,) = [
        r
        for r in db.scalars(select(QueryLog)).all()
        if r.question == OFF_TOPIC
    ][-1:]
    assert row.deflected is True
    assert "backups.md" not in row.sources


def test_zero_max_rounds_reproduce_pre_phase_single_request(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """(c) ``BOR_AGENT_MAX_ROUNDS=0``: no ``tool`` frames, exactly one
    request **without** a ``tools`` key (the pre-phase request shape),
    ``done.sources`` unchanged, and ``tool_calls=0`` in the log line —
    the kill switch survives the phase-45 budget removal."""
    scripted = FakeRagLLM(
        tool_script=[
            [ToolCallPiece(id="call_1", name="ls", arguments={})],
            [
                ToolCallPiece(
                    id="call_2",
                    name="read",
                    arguments={"path": "docs/homelab/backups.md"},
                )
            ],
        ]
    )
    live = get_settings()
    monkeypatch.setattr(
        chat_api,
        "get_settings",
        lambda: Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            relevance_threshold=live.relevance_threshold,
            agent_max_rounds=0,
        ),
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert not any(f["type"] == "tool" for f in frames)
    assert "error" not in [f["type"] for f in frames]
    done = frames[-1]
    assert done["type"] == "done"
    assert done["deflected"] is False
    paths = [s["path"] for s in done["sources"]]
    assert "homelab/kubernetes.md" in paths  # retrieval docs, unchanged
    assert "homelab/backups.md" not in paths  # nothing was read

    # Exactly one request, and it carried no ``tools`` key at all — the
    # scripted tool calls were never even offered a chance.
    assert len(scripted.seen_messages) == 1
    assert scripted.seen_tools == [None]
    assert len(scripted.tool_script) == 2  # never consumed

    (row,) = db.scalars(select(QueryLog)).all()
    assert "docs/homelab/kubernetes.md" in row.sources
    assert "backups.md" not in row.sources
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "tool_calls=0" in lines[-1]
    assert "scaffold_stripped=0" in lines[-1]  # phase 71: uniform clean-turn field


def test_tool_execution_db_failure_yields_error_event(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A tool call that hits a dead DB mid-stream gets the same structured
    ``error`` event as the pre-stream retrieval path — never a severed
    stream (the "never stale" contract, PLAN §7.4)."""
    scripted = FakeRagLLM(
        tool_script=[[ToolCallPiece(id="call_1", name="ls", arguments={})]]
    )

    def boom(*_a: Any, **_k: Any) -> Any:
        raise RuntimeError("db exploded mid tool call")

    # Phase 94: the no-arg ``ls`` executes through ``ls_top`` — the
    # failure hook moves with the rewrite (same contract: the tool
    # frame goes out first, the structured error ends the turn).
    monkeypatch.setattr(agent, "ls_top", boom)
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: scripted
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    # The ``tool`` frame went out first (the model requested the call);
    # the failed execution ends the turn with the structured error event.
    assert [f["type"] for f in frames] == ["tool", "error"]
    assert frames[0]["name"] == "ls"
    assert "offline mid-question" in frames[1]["detail"]
    assert db.scalars(select(QueryLog)).all() == []  # no row for a failed turn


# ---------- phase 67: LLM retries before the first token ----------


def _retry_settings(live: Settings, **overrides: Any) -> Settings:
    """Settings for the retry tests: the live (mock-calibrated) threshold
    plus the phase-67 knobs, with a ZERO delay so the suite never sleeps.
    (The 5 s default is unit-pinned in ``tests/unit/test_config.py``.)"""
    kwargs: dict[str, Any] = {
        "relevance_threshold": live.relevance_threshold,
        "llm_retry_delay": 0.0,
    }
    kwargs.update(overrides)
    return Settings(_env_file=None, **kwargs)  # pyright: ignore[reportCallIssue]


def test_embed_failure_retries_then_turn_completes(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A dead-then-recovered embeddings endpoint: one SSE ``retry`` frame
    (the attempt about to be tried, 1-based) ahead of the normal answer
    frames; the turn completes and the per-turn log line counts the
    retry (``retries=1``)."""
    flaky = FakeRagLLM(embed_fail_count=1)
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=1)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[0] == {"type": "retry", "attempt": 2, "max_attempts": 2}
    assert not any(f["type"] == "error" for f in frames)
    deltas = [f for f in frames if f["type"] == "delta"]
    assert len(deltas) >= 2
    assert "".join(d["text"] for d in deltas) == flaky.answer
    assert frames[-1]["type"] == "done"
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "retries=1" in lines[-1]
    assert "scaffold_stripped=0" in lines[-1]  # phase 71: uniform clean-turn field


def test_embed_failure_exhausts_retries_then_terminal_error(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A dead embeddings endpoint (``llm_retries=2`` → 3 attempts): one
    ``retry`` frame per restart (attempts 2 and 3 of 3), then the
    EXISTING terminal error frame — the copy is unchanged, no query_log
    row."""
    dead = FakeRagLLM(embed_fail_count=99)  # every attempt fails
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=2)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: dead
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert [f["type"] for f in frames] == ["retry", "retry", "error"]
    assert [f["attempt"] for f in frames if f["type"] == "retry"] == [2, 3]
    assert all(f["max_attempts"] == 3 for f in frames if f["type"] == "retry")
    assert "embedding" in frames[-1]["detail"]
    assert db.scalars(select(QueryLog)).all() == []


def test_deflected_stream_retries_before_the_first_piece(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Deflected answer stream: the first attempt dies before any piece,
    the restart streams — a ``retry`` frame ahead of the deltas, the
    request restarted with the same messages (no tools key), and the
    per-turn log line counts the retry."""
    flaky = FakeRagLLM(stream_fail_count=1)
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=2)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _, _, frames = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[0] == {"type": "retry", "attempt": 2, "max_attempts": 3}
    rest = frames[1:]
    assert all(f["type"] in ("delta", "done") for f in rest)
    assert "".join(f["text"] for f in rest if f["type"] == "delta") == flaky.answer
    assert rest[-1]["type"] == "done" and rest[-1]["deflected"] is True
    assert len(flaky.seen_messages) == 2  # the request was restarted
    assert flaky.seen_tools == [None, None]  # …byte-identical (no tools key)
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "retries=1" in lines[-1]
    assert "scaffold_stripped=0" in lines[-1]  # phase 71: uniform clean-turn field


def test_deflected_stream_failure_after_first_frame_is_terminal(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Locked A2: a stream failure AFTER the first output frame is
    terminal — no ``retry`` frame, the existing error copy, no row (a
    partial answer is never redone)."""
    broken = FakeRagLLM(fail_mid_stream=True)
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=3)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
    try:
        _, _, frames = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert [f["type"] for f in frames] == ["delta", "error"]
    assert not any(f["type"] == "retry" for f in frames)
    assert "dropped the connection" in frames[1]["detail"]
    assert db.scalars(select(QueryLog)).all() == []


def test_zero_retries_keep_the_pre_phase_wire_shape(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The ``BOR_LLM_RETRIES=0`` kill switch: one attempt, the existing
    terminal error frame, no ``retry`` frames — the pre-phase-67
    byte-identical wire shape."""
    broken = FakeRagLLM(embed_error=EmbeddingError("embeddings endpoint down"))
    live = get_settings()
    monkeypatch.setattr(
        chat_api, "get_settings", lambda: _retry_settings(live, llm_retries=0)
    )
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: broken
    try:
        _, _, frames = _stream_chat(client, QUESTION)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert len(frames) == 1
    assert frames[0]["type"] == "error"
    assert "embedding" in frames[0]["detail"]
    assert not any(f["type"] == "retry" for f in frames)


# ---------- phase 71: the deterministic scaffolding guardrail (deflected path) ----------


def _scaffold_span() -> str:
    """The raw span from the 2026-09-03 incident (the E2E mock's trigger,
    task 05) — a complete span the filter strips in full."""
    return "<|tool_call_start|>[read(path='/homelab/backup-notes.md')]<|tool_call_end|>"


def test_deflected_scaffolding_only_reply_recovers_once(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """(a) A deflected reply that is pure scaffolding streams ZERO delta
    frames (no raw tokens on the wire); the one bounded recovery —
    ``tools=None``, the correction folded into the single system prompt,
    a fresh filter, the same retry budget — streams the clean answer, the
    turn settles with ``done`` + a query_log row, and the log line counts
    the stripped chars (the recovery does not bump ``retries=N``)."""
    span = _scaffold_span()
    clean = "I don't have that on hand — try one of the chips below?"
    flaky = FakeRagLLM(answer_sequence=[span, clean])
    live = get_settings()
    monkeypatch.setattr(chat_api, "get_settings", lambda: _retry_settings(live))
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _, _, frames = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    # The raw tokens never reach the wire; the deltas reassemble to the
    # clean recovery answer.
    assert span not in json.dumps(frames)
    deltas = [f for f in frames if f["type"] == "delta"]
    assert "".join(d["text"] for d in deltas) == clean
    assert not any(f["type"] == "error" for f in frames)
    done = frames[-1]
    assert done["type"] == "done" and done["deflected"] is True

    # Exactly two requests: the stripped round + the one recovery, both
    # without a tools key…
    assert len(flaky.seen_messages) == 2
    assert flaky.seen_tools == [None, None]
    # …and the recovery's system prompt is the ORIGINAL deflected prompt
    # with the correction folded in (a single system message — the user
    # message stays last).
    recovered = flaky.seen_messages[1]
    assert len(recovered) == 2
    assert recovered[1] == {"role": "user", "content": OFF_TOPIC}
    first_system = flaky.seen_messages[0][0]["content"]
    assert "DEFLECT_MODE" in first_system
    assert recovered[0] == {
        "role": "system",
        "content": first_system + "\n" + agent.CORRECTION_INSTRUCTION,
    }

    # The turn settled normally: one query_log row…
    (row,) = db.scalars(select(QueryLog)).all()
    assert row.deflected is True
    # …and the log line carries the summed stripped count (the clean
    # recovery stripped nothing) with retries untouched.
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and f"scaffold_stripped={len(span)}" in lines[-1]
    assert "retries=0" in lines[-1]  # the recovery is not an endpoint-retry


def test_deflected_scaffolding_twice_settles_malformed(
    client, db, seeded_kb: FakeRagLLM, monkeypatch: pytest.MonkeyPatch
) -> None:
    """(b) The recovery answer is scaffolding again — a second empty
    reply is terminal: the DEDICATED error frame (the exact copy), no
    ``done``, no query_log row — byte-for-byte today's ``LLMError``
    terminal shape — and no third request (at most one recovery per
    turn)."""
    span = _scaffold_span()
    dead = FakeRagLLM(answer_sequence=[span, span])
    live = get_settings()
    monkeypatch.setattr(chat_api, "get_settings", lambda: _retry_settings(live))
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: dead
    try:
        _, _, frames = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert [f["type"] for f in frames] == ["error"]
    assert frames[0]["detail"] == (
        "The model returned a malformed reply — please try again."
    )
    assert set(frames[0].keys()) == {"type", "detail"}  # the contract shape
    assert span not in json.dumps(frames)
    assert not any(f["type"] == "done" for f in frames)
    assert db.scalars(select(QueryLog)).all() == []
    assert len(dead.seen_messages) == 2  # round + one recovery — no more
    assert dead.seen_tools == [None, None]


def test_deflected_mixed_scaffolding_and_content_needs_no_recovery(
    client,
    db,
    seeded_kb: FakeRagLLM,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """(c) Real visible content plus scaffolding: the clean remainder
    streams (no raw tokens on the wire), NO recovery runs, and the log
    line counts the stripped span (``scaffold_stripped>0``)."""
    span = _scaffold_span()
    mixed = f"I don't have that. {span} Try the chips below?"
    flaky = FakeRagLLM(answer=mixed)
    live = get_settings()
    monkeypatch.setattr(chat_api, "get_settings", lambda: _retry_settings(live))
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: flaky
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _, _, frames = _stream_chat(client, OFF_TOPIC)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert span not in json.dumps(frames)
    deltas = [f for f in frames if f["type"] == "delta"]
    assert "".join(d["text"] for d in deltas) == "I don't have that.  Try the chips below?"
    assert not any(f["type"] == "error" for f in frames)
    assert frames[-1]["type"] == "done"
    assert len(flaky.seen_messages) == 1  # the clean content stands — no recovery
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and f"scaffold_stripped={len(span)}" in lines[-1]


# ---------- phase 74: client-provided history (with prior thinking) ----------

#: The client's prior turns (oldest first — the ``bor.chat.v1`` record
#: minus the current question): two user turns, two brain turns, the
#: FIRST brain turn carrying a prior thinking block (A4) and the second
#: not (the ``reasoning_content`` gate has both shapes on one request).
HISTORY: list[dict[str, Any]] = [
    {"who": "user", "text": "What port does Tailscale run on?"},
    {
        "who": "brain",
        "text": "Tailscale runs on 41641/udp.",
        "thinking": "The Tailscale wire protocol uses 41641/udp.",
    },
    {"who": "user", "text": "And the subnet router?"},
    {"who": "brain", "text": "The subnet router shares the same port."},
]

#: What :func:`app.rag.prompts.history_to_messages` must produce for
#: :data:`HISTORY` — chronological, ``reasoning_content`` ONLY on the
#: turn that had thinking.
HISTORY_MESSAGES: list[dict[str, Any]] = [
    {"role": "user", "content": "What port does Tailscale run on?"},
    {
        "role": "assistant",
        "content": "Tailscale runs on 41641/udp.",
        "reasoning_content": "The Tailscale wire protocol uses 41641/udp.",
    },
    {"role": "user", "content": "And the subnet router?"},
    {"role": "assistant", "content": "The subnet router shares the same port."},
]


def _stream_chat_with_history(
    client: TestClient, message: str, history: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    """Phase 74 variant of :func:`_stream_chat`: sends ``history`` (the
    client's prior turns, oldest first) in the request body."""
    with client.stream(
        "POST", "/api/chat", json={"message": message, "history": history}
    ) as r:
        assert r.status_code == 200
        assert r.headers["content-type"].startswith("text/event-stream")
        buf = ""
        frames: list[dict[str, Any]] = []
        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() == "", "stream must end on a frame boundary"
    return frames


def test_deflected_turn_forwards_history_with_prior_thinking(
    client,
    db,
    seeded_kb: FakeRagLLM,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A DEFLECTED turn sends the prior turns — chronological, with the
    prior brain turn's thinking as ``reasoning_content`` — between the
    LOW system prompt and the current question (A2/A3/A4); the per-turn
    log line carries ``history_msgs=4``."""
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        frames = _stream_chat_with_history(client, OFF_TOPIC, HISTORY)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[-1]["type"] == "done"
    assert frames[-1]["deflected"] is True
    assert len(seeded_kb.seen_messages) == 1
    (messages,) = seeded_kb.seen_messages
    assert messages[0]["role"] == "system"
    assert "DEFLECT_MODE" in messages[0]["content"]  # the LOW prompt
    assert messages[1:-1] == HISTORY_MESSAGES  # the prior turns, chronological
    assert messages[-1] == {"role": "user", "content": OFF_TOPIC}
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "history_msgs=4" in lines[-1]


def test_grounded_turn_forwards_history_through_the_agent(
    client,
    db,
    seeded_kb: FakeRagLLM,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The GROUNDED agent branch receives the same block: its first
    request is ``[HIGH system, *history, current question]`` (the tool
    rounds then append to that same list); ``history_msgs=4``."""
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        frames = _stream_chat_with_history(client, QUESTION, HISTORY)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[-1]["type"] == "done"
    assert frames[-1]["deflected"] is False
    assert len(seeded_kb.seen_messages) == 1  # the canned answer ends the loop
    (messages,) = seeded_kb.seen_messages
    assert messages[0]["role"] == "system"
    assert "<tools>" in messages[0]["content"]  # the HIGH prompt
    assert messages[1:-1] == HISTORY_MESSAGES
    assert messages[-1] == {"role": "user", "content": QUESTION}
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "history_msgs=4" in lines[-1]


def test_endpoint_two_turn_history_reaches_the_llm(
    client,
    db,
    seeded_kb: FakeRagLLM,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Phase 108 layer 2 (TODO L4 — the owner's follow-up, server wire):
    the owner's exact 2-turn history (Q1 "What is my name?" / R1 "Your
    name is Reese.") plus the follow-up "What did I just ask you?" must
    reach the LLM as ``[system, user Q1, assistant R1, user Q2]`` — the
    two prior turns NOT dropped (the reported missing-first-turn symptom
    would be their absence from this captured request). The pin is
    branch-agnostic BY DESIGN: phase 74 pinned the history splice on
    BOTH branches, so whichever the seeded KB retrieves to (LOW/deflected
    or HIGH/grounded), the captured request must carry the full prior
    exchange. The seeded-KB fixture is used because it is the file's
    standard idiom (and keeps the cosine gate exercised like production);
    the canned fake answer ends the turn on exactly one LLM request on
    either branch."""
    history = [
        {"who": "user", "text": "What is my name?"},
        {"who": "brain", "text": "Your name is Reese."},
    ]
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        frames = _stream_chat_with_history(client, "What did I just ask you?", history)
    finally:
        fastapi_app.dependency_overrides.clear()

    assert frames[-1]["type"] == "done"  # the turn completes (either branch)
    assert len(seeded_kb.seen_messages) == 1  # the canned answer ends the turn
    (messages,) = seeded_kb.seen_messages
    assert messages[0]["role"] == "system"
    assert messages[1:-1] == [
        {"role": "user", "content": "What is my name?"},
        {"role": "assistant", "content": "Your name is Reese."},
    ]  # the FULL prior exchange — chronological, nothing dropped
    assert messages[-1] == {"role": "user", "content": "What did I just ask you?"}
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert lines and "history_msgs=2" in lines[-1]


def test_request_without_history_sends_exactly_system_and_user(
    client,
    db,
    seeded_kb: FakeRagLLM,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Byte-identical pin (A2): a request WITHOUT ``history`` sends
    exactly the two-message ``[system, user]`` request on BOTH branches
    (deflected + grounded), and the per-turn log line carries
    ``history_msgs=0``."""
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        caplog.set_level(logging.INFO, logger="app.chat")
        _stream_chat(client, OFF_TOPIC)  # deflected branch
        _stream_chat(client, QUESTION)  # grounded branch
    finally:
        fastapi_app.dependency_overrides.clear()

    assert len(seeded_kb.seen_messages) == 2
    for messages in seeded_kb.seen_messages:
        assert [m["role"] for m in messages] == ["system", "user"]
    assert seeded_kb.seen_messages[0][1] == {"role": "user", "content": OFF_TOPIC}
    assert seeded_kb.seen_messages[1][1] == {"role": "user", "content": QUESTION}
    lines = [r.getMessage() for r in caplog.records if "question=" in r.getMessage()]
    assert len(lines) == 2
    assert all("history_msgs=0" in line for line in lines)


def test_history_rejects_unknown_who(client, db) -> None:
    """Schema pin: ``who`` is a ``Literal["user", "brain"]`` — anything
    else is a 422 at the boundary (the same trust model as the saved-
    chat ``ChatMessage``)."""
    r = client.post(
        "/api/chat",
        json={"message": "hi", "history": [{"who": "alien", "text": "x"}]},
    )
    assert r.status_code == 422


def test_history_rejects_more_than_100_entries(client, db) -> None:
    """Schema pin: the DoS sanity ceiling is 100 turns — 101 is a 422
    (the config budgets do the real trimming; this only keeps a
    pathological body from wasting the mapper's work)."""
    r = client.post(
        "/api/chat",
        json={
            "message": "hi",
            "history": [{"who": "user", "text": f"q{i}"} for i in range(101)],
        },
    )
    assert r.status_code == 422


def test_done_event_serializes_column_maximum_source_refs() -> None:
    """Phase 83 A3 pin: ``SourceRef`` is SHARED by the SSE ``done``
    event and the saved-chat surface — the boundary caps added there
    (``source`` ≤ 120, ``path`` ≤ 1000, ``title`` ≤ 500) mirror the
    ``documents`` column lengths EXACTLY, so a server-built event from
    a full-length row (values at the column maxima) still constructs
    and serializes byte-identical: the SSE contract is provably
    unaffected. The one-over caps raise — only client-saved refs can
    ever trip a cap, never a server-built ref."""
    event = ChatDoneEvent(
        deflected=False,
        sources=[SourceRef(source="s" * 120, path="p" * 1000, title="t" * 500)],
        suggestions=[],
    )
    assert event.model_dump() == {
        "type": "done",
        "deflected": False,
        "sources": [{"source": "s" * 120, "path": "p" * 1000, "title": "t" * 500}],
        # Phase 113: the additive related tier defaults to [] (the
        # key is always present on new frames; old clients ignore it).
        "related": [],
        "suggestions": [],
    }
    # The caps sit exactly ON the column maxima: one over any of them
    # is rejected (a row could never hold such a value in the first
    # place — the string columns enforce the same lengths).
    with pytest.raises(ValidationError):
        SourceRef(source="s" * 121, path="p" * 1000, title="t" * 500)
    with pytest.raises(ValidationError):
        SourceRef(source="s" * 120, path="p" * 1001, title="t" * 500)
    with pytest.raises(ValidationError):
        SourceRef(source="s" * 120, path="p" * 1000, title="t" * 501)


def test_done_event_related_defaults_empty_and_old_payload_parses() -> None:
    """Phase 113 back-compat pin: ``related`` defaults to ``[]`` — a
    pre-phase-113 done frame (no ``related`` key) still parses, and a
    frame with the field round-trips it (PLAN §4: old clients ignore
    unknown fields, so the field is additive in both directions)."""
    old_payload = {
        "type": "done",
        "deflected": True,
        "sources": [],
        "suggestions": ["Maybe try X?"],
    }
    event = ChatDoneEvent(**old_payload)
    assert event.related == []
    assert event.model_dump() == {**old_payload, "related": []}

    new_payload = {
        "deflected": False,
        "sources": [SourceRef(source="docs", path="a.md", title="A")],
        "related": [SourceRef(source="docs", path="b.md", title="B")],
        "suggestions": [],
    }
    dumped = ChatDoneEvent(**new_payload).model_dump()
    assert [r["path"] for r in dumped["related"]] == ["b.md"]
    assert [r["path"] for r in dumped["sources"]] == ["a.md"]