"""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
from collections.abc import Iterator
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import func, select, text

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, QueryLog
from app.rag import agent
from app.rag.agent import AGENT_TOOLS
from app.rag.importer import import_sources
from app.rag.llm import EmbeddingError, LLMError, StreamPiece, ToolCallPiece

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,
    ) -> 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 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] = []
        self.seen_messages: list[list[dict[str, str]]] = []
        #: 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)

    async def chat_stream(
        self,
        messages: list[dict[str, str]],
        tools: list[dict[str, Any]] | 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."""
        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")
        if tools is not None and self.tool_script:
            for piece in self.tool_script.pop(0):
                yield piece
            return
        if self.fail_mid_stream:
            yield StreamPiece("content", "partial ")
            raise LLMError("mid-stream dropout")
        for i in range(0, len(self.thinking), 12):
            yield StreamPiece("thinking", self.thinking[i : i + 12])
        for i in range(0, len(self.answer), 12):
            yield StreamPiece("content", self.answer[i : i + 12])


@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()


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"
    assert done["sources"], "weak hits are still reported as the closest 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


def test_keyword_question_grounded_by_lexical_hit_despite_weak_cosine(
    client, db, seeded_kb: FakeRagLLM
) -> 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 — LOW only fires at weak cosine AND zero hits."""
    fastapi_app.dependency_overrides[chat_api.get_llm] = lambda: seeded_kb
    try:
        _, _, frames = _stream_chat(client, "How does kafkabridge work?")
    finally:
        fastapi_app.dependency_overrides.clear()

    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


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 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) -> None:
    from app.db import SessionLocal

    def broken_db():
        real = SessionLocal()
        try:
            yield _BrokenCommitSession(real)
        finally:
            real.close()

    fastapi_app.dependency_overrides[chat_api.get_db] = broken_db
    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 ----------


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="list_documents", arguments={}),
            ],
            [
                ToolCallPiece(
                    id="call_2",
                    name="read_document",
                    arguments={"source": "docs", "path": "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"] == "list_documents"
    assert list_frame["argument"] is None  # the tool takes no parameters
    assert set(read_frame) == {"type", "name", "argument"}
    assert read_frame["name"] == "read_document"
    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]


def test_grounded_turn_streams_search_tool_frames(
    client, db, seeded_kb: FakeRagLLM
) -> None:
    """Phase 68: a scripted ``search_documents`` call streams as
    ``{type: "tool", name: "search_documents", 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 search adds no
    source: ``done.sources`` stays the retrieval docs (locked A5)."""
    scripted = FakeRagLLM(
        tool_script=[
            [
                ToolCallPiece(
                    id="call_1",
                    name="search_documents",
                    arguments={"pattern": "Cilium"},
                ),
            ],
            [
                ToolCallPiece(
                    id="call_2",
                    name="search_documents",
                    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 searches 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"] == "search_documents"
    assert first["argument"] == "Cilium"  # the raw pattern
    assert set(second) == {"type", "name", "argument"}
    assert second["name"] == "search_documents"
    assert second["argument"] is None  # the non-string pattern → null

    # The searches 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 search adds no source


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="list_documents", arguments={})],
            [
                ToolCallPiece(
                    id="call_2",
                    name="read_document",
                    arguments={"source": "docs", "path": "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="list_documents", arguments={})],
            [
                ToolCallPiece(
                    id="call_2",
                    name="read_document",
                    arguments={"source": "docs", "path": "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]


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="list_documents", arguments={})]]
    )

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

    monkeypatch.setattr(agent, "list_catalog", 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"] == "list_documents"
    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]


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]


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)