"""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 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

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"
    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 ----------


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
        db,
        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,
    )
    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"]
        assert body.startswith("Document docs/big.md:\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,
    )
    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, byte-identical to
        # the pre-phase-95 read result (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:\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,
    )
    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"]
    assert body.startswith(f"Document docs/big-read.md:\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:\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_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}],
        "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)