"""Unit: the grounded-turn agent loop (phase 37, ``app.rag.agent``; the
harness-aligned ``ls``/``read``/``grep`` surface, phase 70).

A scripted fake LLM (canned stream sequences) + monkeypatched
``ls_top`` / ``ls_folder`` / ``list_source_names`` / ``find_document`` /
``all_documents`` — no database, no network. Covers the loop mechanics:
the ls → read (combined ``source/path``) → answer happy path (event
order, holder state, the tools staying offered on every request —
phase 45 removed the per-tool budgets, the assistant/tool message
history), the phase-94 drill-down ``ls`` (no-arg top level = the
registered sources with counts + stored summaries in the pinned
``{N} sources:`` template, a source scope = its root folder —
subfolders + capped file lines in the pinned folder template —, a
``source/folder`` scope = one level deeper, a registered source with 0
documents → the ``… — 0 documents, 0 folders:`` header counted, an
unknown-source refusal that counts nothing, the NOT-A-FOLDER teaching
with the parent's subfolders, the 50-file cap + grep-pointer note),
``read`` on the canonical combined form (split at the
FIRST slash, full content, the bare-source-name refusal, the
already-in-context dedupe, missing-args refusals), the phase-68 ``grep``
contract under its new name (the locked A5 pins: fixed substring,
case-insensitive, 20-cap in catalog order, 200-char lines, locator-only
— ``read_docs`` untouched, no-match lines counted), the round cap
forcing a final no-tools answer, the kill switch
(``agent_max_rounds=0`` single-call path), and the phase-67 per-round
retries (a dead-then-recovered round restarts before its first piece
with a ``RetryPiece``; a mid-stream drop stays terminal — locked A2;
the forced final no-tools call retries too; ``llm_retries=0`` is one
plain attempt; retries are invisible to the round cap; consumer abandon
mid-retry-sleep leaks nothing).
"""
from __future__ import annotations

import asyncio
import json
import logging
import uuid
from collections.abc import AsyncGenerator, AsyncIterator, Callable, Sequence
from copy import deepcopy
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import MagicMock

import pytest
from sqlalchemy.orm import Session

from app.config import Settings
from app.models import Document, GitSource
from app.rag import agent
from app.rag.agent import (
    AGENT_TOOLS,
    READ_TRUNCATION_NOTICE,
    AgentHolder,
    MalformedReplyError,
    run_agent,
)
from app.rag.llm import (
    LLMClient,
    LLMError,
    RetryPiece,
    StreamPiece,
    ToolCallPiece,
    ToolResultPiece,
)
from app.rag.prompts import TOOLS_SECTION, _base, build_deflect_prompt, build_high_prompt
from app.rag.retriever import TRUNCATION_MARKER

if TYPE_CHECKING:
    from app.rag.scaffolding import ScaffoldingFilter


def _settings(**kwargs: Any) -> Settings:
    kwargs.setdefault("_env_file", None)
    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]


#: The fixture documents' fixed creation date (phase 106, D5): the
#: agent formats ``doc.created_at`` on the ``read`` result's second
#: line — the detached fixture rows carry it exactly as the NOT NULL
#: DB column guarantees it for real rows.
_FIXTURE_CREATED_AT = datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC)


def _doc(source: str, path: str, title: str = "Title", content: str = "CONTENT") -> Document:
    return Document(
        id=uuid.uuid4(),
        source=source,
        path=path,
        full_path=f"/tmp/{path}",
        title=title,
        content=content,
        content_hash="0" * 64,
        created_at=_FIXTURE_CREATED_AT,
    )


class ScriptedLLM:
    """Canned stream sequences; records every ``chat_stream`` request so
    the tests can assert on the messages and the ``tools`` passthrough.
    Phase 71: when the caller passes a ``ScaffoldingFilter``, the canned
    content pieces are fed through it exactly like
    ``LLMClient.chat_stream`` (an empty clean result yields nothing; the
    held tail is flushed on normal completion) — so a scaffolding-only
    canned round streams no content pieces and leaves ``stripped_chars``
    behind for the recovery policy to key on."""

    def __init__(self, *streams: list[StreamPiece | ToolCallPiece]) -> None:
        self.streams: list[list[StreamPiece | ToolCallPiece]] = list(streams)
        self.requests: list[tuple[list[dict[str, Any]], list[dict[str, Any]] | None]] = []

    async def chat_stream(
        self,
        messages: list[dict[str, str]],
        tools: list[dict[str, Any]] | None = None,
        scaffolding: ScaffoldingFilter | None = None,
    ) -> AsyncIterator[StreamPiece | ToolCallPiece]:
        self.requests.append((deepcopy(messages), tools))
        if not self.streams:
            raise AssertionError("ScriptedLLM ran out of canned streams")
        pieces = self.streams.pop(0)
        if scaffolding is None:
            for piece in pieces:
                yield piece
            return
        for piece in pieces:
            if isinstance(piece, StreamPiece) and piece.kind == "content":
                cleaned = scaffolding.feed(piece.text)
                if cleaned:
                    yield StreamPiece("content", cleaned)
            else:
                yield piece
        tail = scaffolding.flush()
        if tail:
            yield StreamPiece("content", tail)


def _mock_session() -> Session:
    """A minimal mock Session for unit tests (DB accessors are monkeypatched).

    The mock works as a context manager: ``__enter__`` returns itself so
    ``with db_factory() as tool_db:`` binds *tool_db* to the same mock.
    """
    mock = cast("Session", MagicMock())
    mock.__enter__ = MagicMock(return_value=mock)
    mock.__exit__ = MagicMock(return_value=False)
    mock.scalar = MagicMock(return_value=None)
    mock.execute = MagicMock(return_value=MagicMock(scalars=MagicMock(return_value=[])))
    return mock


async def _run(
    llm: ScriptedLLM | FailingLLM,
    holder: AgentHolder,
    settings: Settings,
    seed_docs: list[Document] | None = None,
    history: Sequence[dict[str, Any]] = (),
    db_factory: Callable[[], Session] | None = None,
) -> list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece]:
    """Consume one ``run_agent`` turn; *history* (phase 74) is the
    client's prior turns spliced between system and user (default
    ``()`` — the pre-phase-74 two-message request). Phase 95: the loop
    may also yield a ``ToolResultPiece`` (a truncated ``read``)."""
    out: list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece] = []
    factory = db_factory or (lambda: _mock_session())
    async for piece in run_agent(
        cast("LLMClient", llm),
        factory,
        system_prompt="SYSTEM_PROMPT",
        user_message="QUESTION",
        seed_docs=seed_docs or [],
        settings=settings,
        holder=holder,
        history=history,
    ):
        out.append(piece)
    return out


# ---------- AGENT_TOOLS shape (phase 70: ls / read / grep) ----------


def test_agent_tools_names_and_parameters() -> None:
    by_name = {t["function"]["name"]: t for t in AGENT_TOOLS}
    assert len(AGENT_TOOLS) == 3  # ls / read / grep (phase 70)
    assert set(by_name) == {"ls", "read", "grep"}
    # The phase-37/68 names exist nowhere in the tool surface.
    assert not set(by_name) & {"list_documents", "read_document", "search_documents"}
    assert all(t["type"] == "function" for t in AGENT_TOOLS)
    ls = by_name["ls"]["function"]
    # Task 05 (live gate iteration 2): the one-call-at-a-time discipline
    # clause (the harness prior batches calls; the loop executes one
    # per round — the extras count as unexecuted in the gate).
    # Phase 94 (task 03): the description is the drill-down tree
    # contract (pinned copy — the tool-surface revision, owner
    # permission 2026-09-10, ``TODO.md`` L4): one level per call,
    # sources at the top, folders + files below, the file-line format
    # and the combined-identity handoff to read/grep intact.
    assert ls["description"] == (
        "List the knowledge base as a tree, one level at a time. "
        "With no path: the synced sources — each with its document "
        "count and a summary of its contents. With a source name (no "
        "'/'): that source's top-level folders and files. With a "
        "`source/folder` path: that folder's subfolders and files. "
        "Folder lines carry a summary of what the folder contains. "
        "File lines are `source: X | path: Y | title: Z` — use the "
        "combined `source/path` with `read` and `grep`. Call one tool "
        "at a time — wait for this result before your next call."
    )
    ls_params = ls["parameters"]
    assert ls_params["type"] == "object"
    assert ls_params["required"] == []  # path is optional
    assert set(ls_params["properties"]) == {"path"}
    assert ls_params["properties"]["path"]["type"] == "string"
    # Phase 94 (task 03): the 'path' argument teaches the drill-down
    # semantics — a source name lists that source's top level, a
    # `source/folder` path drills one level deeper, omitted lists
    # every source (pinned copy).
    assert ls_params["properties"]["path"]["description"] == (
        "Optional — a source name (e.g. 'homelab') to list its top "
        "level, or a `source/folder` path to drill down (e.g. "
        "'homelab/active'). Omit it to list every source."
    )
    read = by_name["read"]["function"]
    # Tool-calling fast loop (2026-09-04, controlled fixture gate):
    # the do-not-read rule is FRONT-LOADED — the controlled gate's
    # telemetry showed the `lite` model obeying the user's "open it /
    # read it" and reading seed-context documents the <documents>
    # section already carries (every refusal of a 12-call run was
    # ALREADY_IN_CONTEXT); the rule now leads the description instead
    # of sitting mid-paragraph, and the tool is framed as "only for
    # documents NOT already in <documents>". Phase 95 (task 01): the
    # read-truncation sentence is inserted before the one-call-at-a-
    # time discipline clause (the discipline rule stays last, as in the
    # other two tools) — a capped read carries the TRUNCATED notice and
    # the `grep` follow-up (the pinned copy).
    assert read["description"] == (
        "Do not call this tool for a document already shown in "
        "the <documents> section, even when the user asks you to "
        "open or read it — its full text is already in your "
        "prompt; answer directly from it. Use it only to add a "
        "document NOT already in <documents> to your context, "
        "by its combined `source/path` string. Very large "
        "documents are truncated: you receive the first part "
        "plus a TRUNCATED notice naming how many more characters "
        "exist — the notice is authoritative, the document did "
        "NOT end where it stopped. Follow it and use `grep` "
        "(pattern) to locate the rest — it searches the whole "
        "document. Call one tool at a time — wait for this "
        "result before your next call."
    )
    read_params = read["parameters"]
    assert read_params["type"] == "object"
    assert read_params["required"] == ["path"]
    assert set(read_params["properties"]) == {"path"}
    assert read_params["properties"]["path"]["type"] == "string"
    # The combined source/path string is the canonical document identity
    # (phase 70) — the description pins it with a worked example. Phase
    # 72 (task 02): the bare-path contract is stated up front; task 05
    # (live gate iteration 1): the do-not-re-read clause (the dedupe
    # refusal's prevention at the prompt).
    assert read_params["properties"]["path"]["description"] == (
        "The document to add to your context, as the combined "
        "`source/path` string exactly as shown in the `ls` output (e.g. "
        "'homelab/active/container_caddy/caddy.md'). A bare document "
        "path (without the source name) will not resolve. Only pass a "
        "document NOT already shown in the <documents> section — it is "
        "already in your context; do not re-read it."
    )
    grep = by_name["grep"]["function"]
    # Task 05 (live gate iterations 2-6, refined in the 2026-09-03
    # re-run): the pattern-only-is-the-knowledge-base-search clause
    # ("pass ONLY `pattern`") + the source-name-is-not-a-document
    # clause (the model kept scoping grep with an ls-style source name
    # — the 2026-09-03 incident loop shape, but on grep) plus the
    # one-call-at-a-time discipline clause. The 2026-09-05 incident
    # (the "Qwen 3.8" sample question — the harness prior is that grep
    # takes a REGEX; this grep is a fixed substring, owner-locked A5):
    # the plain-substring-never-a-regex clause states the contract up
    # front, so the regex-shaped first grep that does fire gets the
    # teaching no-match line instead of a trusted miss.
    assert grep["description"] == (
        "Search the indexed documents for an exact string "
        "(case-insensitive) and return up to 20 matching lines "
        "as `source/path:line: text` — a locator, not a "
        "context-adder: read the winner with `read`. The "
        "pattern is a plain substring, NEVER a regex — if a "
        "pattern with regex syntax (like '.*' or '\\.') comes "
        "back with no matches, retry with the plain text you "
        "expect to see. For a normal search pass ONLY `pattern` "
        "— it searches every document and that is how you "
        "search the knowledge base; never pass a source name "
        "as `path` (a source name is not a document). Call one "
        "tool at a time — wait for this result before your "
        "next call."
    )
    grep_params = grep["parameters"]
    assert grep_params["type"] == "object"
    assert grep_params["required"] == ["pattern"]
    assert set(grep_params["properties"]) == {"pattern", "path"}
    assert all(p["type"] == "string" for p in grep_params["properties"].values())
    assert grep_params["properties"]["pattern"]["description"] == (
        "The exact text to search for (a plain substring, "
        "not a regex — no '.*', no '\\.', no character classes)"
    )
    # Phase 72 (task 02): the bare-path contract is stated up front;
    # task 05 (live gate iterations 1-8): the one-known-document clause
    # with a worked combined-identity example and the source-name ban
    # (the model kept scoping grep with an ls-style source name — the
    # incident loop shape, but on grep).
    # Iteration 8 drops the standalone 'homelab' from this negative
    # example — the gate's live telemetry showed the model emitting
    # exactly that value, and naming it beside the parameter risks
    # priming it (the negative-example effect). The 2026-09-03 re-run
    # makes the rarity explicit ("Rarely needed") and re-states the
    # pattern-only normal search.
    assert grep_params["properties"]["path"]["description"] == (
        "Rarely needed — only for re-searching one "
        "document you already know: that document's "
        "combined `source/path` identity (e.g. "
        "'homelab/ansible/inventory.yaml'). Never a "
        "source name. A bare document path (without "
        "the source name) will not resolve. Omit it "
        "for a normal search (pass only `pattern`)."
    )


def test_agent_tools_order_is_ls_read_grep() -> None:
    """The listing → context → locator order the prompt teaches (the API
    layer and the mock key off the names)."""
    assert [t["function"]["name"] for t in AGENT_TOOLS] == ["ls", "read", "grep"]


def test_refusal_constants_are_harness_aligned() -> None:
    """The updated module-level refusal lines (the names moved to the
    harness surface). The ALREADY_IN_CONTEXT line is a phase-72,
    task 05 gate-iteration teaching (live telemetry: the model
    repeated the terse phase-37 line) — same refusal behavior, the
    copy names the correct action."""
    assert agent.ALREADY_IN_CONTEXT == (
        "Already in your context — the full text is already in your "
        "prompt. Do not call read on it again; answer from that text."
    )
    assert agent.UNKNOWN_TOOL == "Unknown tool."
    assert agent.MISSING_READ_ARGS == "read requires a string argument 'path'."
    assert agent.MISSING_SEARCH_ARGS == "grep requires a string argument 'pattern'."
    # Phase 94 (task 03): the phase-72 document-path teaching refusal is
    # DELETED (a ``/`` now names a folder — the drill-down contract);
    # the no-source refusal stays byte-identical (the task's "existing
    # refusal, teaching parenthetical intact" pin).
    assert not hasattr(agent, "LS_PATH_NOT_A_SOURCE")
    assert agent.NO_SOURCE_NOT_A_DIRECTORY.startswith(
        "No source named '{scope}' — check the ls output."
    )
    assert agent.NO_SOURCE_NOT_A_DIRECTORY == (
        "No source named '{scope}' — check the ls output. (The 'path' "
        "argument is a source name, not a directory — omit it to list "
        "every document.)"
    )
    # Phase 94 (task 03): the NOT-A-FOLDER drill-down teaching template,
    # pinned byte-for-byte (argument echoed, parent's subfolders
    # listed), and the pinned file-line cap constant.
    assert agent.NOT_A_FOLDER == "'{arg}' is not a folder — {parent} has: {subfolders}"
    assert agent.LS_MAX_FILE_LINES == 50
    # Phase 72 (task 02): the read/grep "did you mean …?" suggestion
    # templates, pinned byte-for-byte, and the suggestion cap.
    assert agent.NO_DOCUMENT_DID_YOU_MEAN == (
        "No document at '{arg}' — did you mean '{source}/{path}'?"
    )
    assert agent.NO_DOCUMENT_DID_YOU_MEAN_MANY == (
        "No document at '{arg}' — did you mean one of: {candidates}?"
    )
    assert agent.SUGGESTION_LIMIT == 3


# ---------- list_source_names (the scoped ls registry join) ----------


def test_list_source_names_resolves_registry_rows(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Names resolve exactly as the import pipeline indexes them (the
    phase-69 ``resolve_source_name`` expressions — reuse, not
    re-derivation), deduped (two rows resolving to the same name share
    documents), in registry order."""
    rows = [
        GitSource(url="https://github.com/reese/homelab.git", kind="git"),
        GitSource(
            url="/srv/reese/deployments", kind="local", path="/srv/reese/deployments"
        ),
        # The phase-69 sibling case: a second row, same resolved name.
        GitSource(url="https://github.com/reese/homelab", kind="git"),
    ]
    monkeypatch.setattr(agent, "effective_sources", lambda db: (rows, "db"))
    assert agent.list_source_names(cast("Session", object())) == [
        "homelab",
        "deployments",
    ]


def test_list_source_names_empty_registry(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(agent, "effective_sources", lambda db: ([], "env"))
    assert agent.list_source_names(cast("Session", object())) == []


# ---------- happy path: ls → read (combined path) → answer ----------


def test_ls_then_read_then_answer(monkeypatch: pytest.MonkeyPatch) -> None:
    # Phase 94 (task 03): the no-arg ``ls`` is the drill-down TOP level
    # (the registered sources, registry order) — monkeypatched the way
    # the phase-70 full-catalog listing used to be.
    monkeypatch.setattr(
        agent, "ls_top", lambda db: [("Deployments", 1, None), ("Homelab", 1, None)]
    )
    target = _doc("Homelab", "aws-route53.md", "AWS Route53 Records", "R53-CONTENT")
    calls: list[tuple[str, str]] = []

    def _find(db: Any, source: str, path: str) -> Document | None:
        calls.append((source, path))
        return target if (source, path) == ("Homelab", "aws-route53.md") else None

    monkeypatch.setattr(agent, "find_document", _find)
    seed = [_doc("Homelab", "kubernetes.md", "Kubernetes", "K8S-CONTENT")]
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [
            ToolCallPiece(
                id="call_2",
                name="read",
                arguments={"path": "Homelab/aws-route53.md"},
            )
        ],
        [StreamPiece("thinking", "hmm "), StreamPiece("content", "Done! ")],
    )

    pieces = asyncio.run(_run(llm, holder, _settings(), seed_docs=seed))

    # Event order: tool pieces before the answer content/thinking.
    assert [type(p) for p in pieces] == [
        ToolCallPiece,
        ToolCallPiece,
        StreamPiece,
        StreamPiece,
    ]
    assert pieces[0] == ToolCallPiece(id="call_1", name="ls", arguments={})
    assert pieces[1] == ToolCallPiece(
        id="call_2", name="read", arguments={"path": "Homelab/aws-route53.md"}
    )
    assert pieces[3] == StreamPiece("content", "Done! ")
    # The read document is recorded for done.sources / query_log (task 04).
    assert holder.read_docs == [target]
    assert holder.tool_calls == 2

    # Phase 45: no per-tool budgets — the tools stay offered on every
    # request (the round cap, not spent budgets, bounds the loop), so
    # the answer request still carries them (2 rounds < default cap 10).
    assert llm.requests[0][1] == AGENT_TOOLS
    assert llm.requests[1][1] == AGENT_TOOLS
    assert llm.requests[2][1] == AGENT_TOOLS
    assert len(llm.requests) == 3

    # read splits the combined form at the FIRST slash — one exact
    # lookup, no self-correction candidates (phase 70).
    assert calls == [("Homelab", "aws-route53.md")]

    # The follow-up request carries the assistant tool-call + tool
    # result (phase 106, D5: the ``date:`` second line rides every
    # read — the first line is byte-identical).
    msgs = llm.requests[1][0]
    assert msgs[0] == {"role": "system", "content": "SYSTEM_PROMPT"}
    assert msgs[1] == {"role": "user", "content": "QUESTION"}
    assert msgs[2] == {
        "role": "assistant",
        "content": None,
        "tool_calls": [
            {
                "id": "call_1",
                "type": "function",
                "function": {"name": "ls", "arguments": "{}"},
            }
        ],
    }
    assert msgs[3] == {
        "role": "tool",
        "tool_call_id": "call_1",
        "content": "2 sources:\n\nDeployments — 1 documents\nHomelab — 1 documents",
    }
    # The second follow-up request carries the read call + the FULL text.
    msgs = llm.requests[2][0]
    assert msgs[4]["role"] == "assistant"
    assert msgs[4]["tool_calls"][0]["id"] == "call_2"
    assert json.loads(msgs[4]["tool_calls"][0]["function"]["arguments"]) == {
        "path": "Homelab/aws-route53.md"
    }
    assert msgs[5] == {
        "role": "tool",
        "tool_call_id": "call_2",
        "content": (
            "Document Homelab/aws-route53.md:\n"
            "date: 2024-06-15\n"
            "R53-CONTENT"
        ),  # full text, no cap (date: the D5 second line)
    }


def test_content_and_tool_call_in_one_stream_keeps_both(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Rare stream with content AND a tool call: the content stays (it was
    already emitted) and the tool still runs."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            StreamPiece("content", "Let me check "),
            ToolCallPiece(id="call_1", name="ls", arguments={}),
        ],
        [StreamPiece("content", "the answer")],
    )
    pieces = asyncio.run(_run(llm, holder, _settings()))
    assert [type(p) for p in pieces] == [StreamPiece, ToolCallPiece, StreamPiece]
    assert holder.tool_calls == 1  # the tool ran despite the content
    assert llm.requests[1][0][3]["content"] == "0 sources:"


# ---------- phase 74: client history between system and user ----------


def test_run_agent_default_history_keeps_two_message_request() -> None:
    """No *history* (the default ``()``) → the model sees exactly the
    pre-phase-74 two-message request ``[system, user]`` — byte-identical
    behavior (owner-locked A2)."""
    llm = ScriptedLLM([StreamPiece("content", "the answer")])
    asyncio.run(_run(llm, AgentHolder(), _settings()))
    (messages, _tools) = llm.requests[0]
    assert messages == [
        {"role": "system", "content": "SYSTEM_PROMPT"},
        {"role": "user", "content": "QUESTION"},
    ]


def test_run_agent_places_history_between_system_and_user() -> None:
    """A non-empty *history* (the client's prior turns, already mapped by
    ``history_to_messages``) is spliced between the system prompt and the
    CURRENT user message — oldest-first, with the assistant turn's prior
    thinking riding on ``reasoning_content`` (A4). The current question
    stays LAST."""
    history = [
        {"role": "user", "content": "old question"},
        {
            "role": "assistant",
            "content": "old answer",
            "reasoning_content": "old thinking",
        },
    ]
    llm = ScriptedLLM([StreamPiece("content", "the answer")])
    pieces = asyncio.run(
        _run(llm, AgentHolder(), _settings(), history=history)
    )
    assert [p for p in pieces if isinstance(p, StreamPiece)] == [
        StreamPiece("content", "the answer")
    ]
    (messages, _tools) = llm.requests[0]
    assert messages == [
        {"role": "system", "content": "SYSTEM_PROMPT"},
        {"role": "user", "content": "old question"},
        {
            "role": "assistant",
            "content": "old answer",
            "reasoning_content": "old thinking",
        },
        {"role": "user", "content": "QUESTION"},
    ]


def test_run_agent_history_survives_a_tool_round(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The tool rounds append assistant/tool messages to the SAME
    ``messages`` list — the prior history stays in place between the
    system prompt and the current question on the SECOND request too."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [])
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [StreamPiece("content", "the answer")],
    )
    history = [{"role": "assistant", "content": "old answer"}]
    asyncio.run(_run(llm, AgentHolder(), _settings(), history=history))
    _first, second = llm.requests
    assert second[0][:3] == [
        {"role": "system", "content": "SYSTEM_PROMPT"},
        {"role": "assistant", "content": "old answer"},
        {"role": "user", "content": "QUESTION"},
    ]


# ---------- ls: the drill-down tree (phase 94, task 03) ----------


def test_ls_top_level_lists_sources_with_summaries(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """No argument: the TOP level — the registered sources, registry
    order, each ``{source} — {n} documents`` + the indented summary line
    only when stored — the pinned template, counted; the registry IS
    consulted (unlike the phase-70 full catalog, the top level is the
    registry itself)."""
    monkeypatch.setattr(
        agent,
        "ls_top",
        lambda db: [
            ("Deployments", 0, None),
            ("Homelab", 1, "The homelab notes."),
        ],
    )
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "2 sources:\n"
        "\n"
        "Deployments — 0 documents\n"
        "Homelab — 1 documents\n"
        "  The homelab notes."
    )
    assert holder.tool_calls == 1


def test_ls_empty_registry_says_zero_sources(monkeypatch: pytest.MonkeyPatch) -> None:
    """No registered sources: the top level is the header line alone —
    ``0 sources:`` (the old ``0 documents:`` behavior preserved in
    spirit), still a counted result."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "0 sources:"
    assert holder.tool_calls == 1


@pytest.mark.parametrize(
    ("arguments", "label"),
    [
        ({"path": "   "}, "blank path"),
        ({"path": 7}, "non-string path"),
    ],
)
def test_ls_blank_path_lists_top_level(
    monkeypatch: pytest.MonkeyPatch, arguments: dict[str, Any], label: str
) -> None:
    """A blank (or non-string) ``path`` is treated as omitted — the top
    level (the sources), counted (no refusal for an empty scope)."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["S"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments=arguments)],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "1 sources:\n\nS — 1 documents"
    assert holder.tool_calls == 1


def test_ls_source_scope_lists_root_folder(monkeypatch: pytest.MonkeyPatch) -> None:
    """A registered source name (no ``/``): the source's ROOT folder —
    subfolders (2-space-indented, path order, ``: {summary}`` only when
    stored) + the root's own file lines in EXACTLY the
    ``source: X | path: Y | title: Z | date: YYYY-MM-DD`` format (the
    phase-106 D5 appended date field) — the pinned template, counted."""
    monkeypatch.setattr(
        agent,
        "ls_folder",
        lambda db, source, folder: (
            [("backups", 2, "Backup notes."), ("networking", 1, None)],
            [("Homelab", "readme.md", "Readme", "2024-06-15")],
            1,
        ),
    )
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["Deployments", "Homelab"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={"path": "Homelab"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "Homelab — 1 documents, 2 folders:\n"
        "\n"
        "  backups/ — 2 documents: Backup notes.\n"
        "  networking/ — 1 documents\n"
        "\n"
        "source: Homelab | path: readme.md | title: Readme | date: 2024-06-15"
    )
    assert holder.tool_calls == 1


def test_ls_nested_folder_scope_lists_one_level_deeper(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A ``source/folder`` path: that folder's subfolders + own file
    lines, identity = ``source/folder`` (the same template as the
    root), counted; the fetchers are the source-scoped ones."""

    def _rows(db: Any, source: str) -> list[tuple[str, str, str]]:
        assert source == "Homelab"  # db is a mock session (SEC-14-04)
        return [
            ("networking/lan.md", "LAN", "2024-06-15"),
            ("networking/vpn.md", "VPN", "2024-06-15"),
        ]

    monkeypatch.setattr(agent, "_source_document_rows", _rows)
    monkeypatch.setattr(
        agent, "_source_folder_summaries", lambda db, source: {"networking": "Network notes."}
    )
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["Homelab"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1", name="ls", arguments={"path": "Homelab/networking"}
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "Homelab/networking — 2 documents, 0 folders:\n"
        "\n"
        "source: Homelab | path: networking/lan.md | title: LAN | date: 2024-06-15\n"
        "source: Homelab | path: networking/vpn.md | title: VPN | date: 2024-06-15"
    )
    assert holder.tool_calls == 1


def test_ls_scoped_known_source_with_zero_docs_counts(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A registered source with no indexed documents is KNOWN (the
    registry is the source of truth, not the catalog): it lists its
    header line alone (``… — 0 documents, 0 folders:`` — the old
    ``0 documents:`` behavior preserved in spirit) — a valid, counted
    result, not a refusal."""
    monkeypatch.setattr(
        agent, "ls_folder", lambda db, source, folder: ([], [], 0)
    )
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["Homelab", "Other"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={"path": "Homelab"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "Homelab — 0 documents, 0 folders:"
    assert holder.tool_calls == 1  # an executed ls, not a refusal
    assert llm.requests[1][1] == AGENT_TOOLS


def test_ls_scoped_unknown_source_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    """A ``path`` without ``/`` matching no source name is a refusal —
    the extended line with the teaching parenthetical (phase 72), not
    counted, the round cap bounds its repetition."""
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["S"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={"path": "Ghost"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 0  # a refusal counts in nothing
    assert (
        llm.requests[1][0][3]["content"]
        == agent.NO_SOURCE_NOT_A_DIRECTORY.format(scope="Ghost")
    )
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_ls_path_like_scope_unknown_source_gets_no_source_refusal(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 94: a ``/`` now names a folder, so the phase-72
    document-path teaching is DELETED — a ``source/…`` argument whose
    FIRST segment names no registered source gets the no-source refusal
    (the segment echoed), counted in nothing, tools stay offered."""
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["S"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="ls",
                arguments={"path": "app/rag/importer.py"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 0  # a refusal counts in nothing
    assert (
        llm.requests[1][0][3]["content"]
        == agent.NO_SOURCE_NOT_A_DIRECTORY.format(scope="app")
    )
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_ls_dot_scope_gets_not_a_directory_teaching_refusal(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 72: ``ls(path='.')`` (the incident's second round — no
    ``/``, no matching source) gets the extended no-source refusal with
    the teaching parenthetical, ``'.'`` echoed — not counted, tools stay
    offered (unchanged by phase 94)."""
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["S"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={"path": "."})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 0  # a refusal counts in nothing
    assert (
        llm.requests[1][0][3]["content"]
        == agent.NO_SOURCE_NOT_A_DIRECTORY.format(scope=".")
    )
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_ls_unknown_top_level_folder_gets_not_a_folder_teaching(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 94: a ``source/…`` argument whose TOP-LEVEL folder segment
    matches no indexed prefix gets the NOT-A-FOLDER teaching — the
    argument echoed, the source named, its direct subfolders listed so
    the model self-corrects in the next round; not counted, tools stay
    offered."""
    monkeypatch.setattr(
        agent,
        "_source_document_rows",
        lambda db, source: [
            ("backups/cron.md", "Cron", "2024-06-15"),
            ("containers/caddy.md", "Caddy", "2024-06-15"),
            ("networking/lan.md", "LAN", "2024-06-15"),
        ],
    )
    monkeypatch.setattr(agent, "_source_folder_summaries", lambda db, source: {})
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["Homelab"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1", name="ls", arguments={"path": "Homelab/netwoking"}
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 0  # a refusal counts in nothing
    assert llm.requests[1][0][3]["content"] == (
        "'Homelab/netwoking' is not a folder — Homelab has: "
        "backups/ containers/ networking/"
    )
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_ls_unknown_nested_folder_gets_not_a_folder_with_nested_parent(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 94: a nested miss names the DEEPEST existing ancestor —
    ``source/folder`` — and lists ITS direct subfolders (bounded: the
    parent's own listing, no new flood path)."""
    monkeypatch.setattr(
        agent,
        "_source_document_rows",
        lambda db, source: [
            ("networking/lan/a.md", "A", "2024-06-15"),
            ("networking/vpn/b.md", "B", "2024-06-15"),
            ("readme.md", "Readme", "2024-06-15"),
        ],
    )
    monkeypatch.setattr(agent, "_source_folder_summaries", lambda db, source: {})
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["Homelab"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1", name="ls", arguments={"path": "Homelab/networking/lan/x"}
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == (
        "'Homelab/networking/lan/x' is not a folder — "
        "Homelab/networking/lan has: none"
    )


def test_ls_file_path_scope_gets_not_a_folder(monkeypatch: pytest.MonkeyPatch) -> None:
    """Phase 94: a document's OWN path is never a folder (nothing starts
    with ``path + '/'``) — ``ls`` of a file path refuses with the
    NOT-A-FOLDER teaching (the parent's subfolders listed)."""
    monkeypatch.setattr(
        agent,
        "_source_document_rows",
        lambda db, source: [
            ("notes.md", "Notes", "2024-06-15"),
            ("a/b.md", "B", "2024-06-15"),
        ],
    )
    monkeypatch.setattr(agent, "_source_folder_summaries", lambda db, source: {})
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["S"])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(id="call_1", name="ls", arguments={"path": "S/notes.md"})
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == (
        "'S/notes.md' is not a folder — S has: a/"
    )


# ---------- ls_top / ls_folder: the drill-down accessors (pure + composed) ----------


def test_ls_top_registry_order_zero_docs_and_summaries(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``ls_top``: registry order (not catalog order), a 0-document
    source still lists, the summary is the stored ``(source, "")`` row
    or ``None`` when absent; an empty registry → ``[]``."""
    monkeypatch.setattr(agent, "list_source_names", lambda db: ["Zeta", "Alpha"])
    monkeypatch.setattr(
        agent, "_source_document_counts", lambda db: [("Zeta", 3), ("Beta", 1)]
    )
    monkeypatch.setattr(
        agent, "_source_root_summaries", lambda db: [("Zeta", "Zeta stuff.")]
    )
    assert agent.ls_top(cast("Session", object())) == [
        ("Zeta", 3, "Zeta stuff."),
        ("Alpha", 0, None),  # 0 docs (no count row) + no stored summary
    ]


def test_ls_top_empty_registry_is_empty(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(agent, "list_source_names", lambda db: [])

    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError("no fetches for an empty registry")

    monkeypatch.setattr(agent, "_source_document_counts", _boom)
    monkeypatch.setattr(agent, "_source_root_summaries", _boom)
    assert agent.ls_top(cast("Session", object())) == []


def test_ls_folder_composes_the_fetchers(monkeypatch: pytest.MonkeyPatch) -> None:
    """``ls_folder`` = the source's document rows + stored summaries
    through the pure :func:`group_folder_listing` (the fetchers are the
    monkeypatch surface)."""
    seen: list[tuple[str, str, str]] = []

    def _rows(db: Any, source: str) -> list[tuple[str, str, str]]:
        seen.append(("rows", source, ""))
        return [("a/b.md", "B", "2024-06-15"), ("a.md", "A", "2024-06-15")]

    def _summaries(db: Any, source: str) -> dict[str, str]:
        seen.append(("summaries", source, ""))
        return {"a": "A stuff."}

    monkeypatch.setattr(agent, "_source_document_rows", _rows)
    monkeypatch.setattr(agent, "_source_folder_summaries", _summaries)
    assert agent.ls_folder(cast("Session", object()), "S", "") == (
        [("a", 1, "A stuff.")],
        [("S", "a.md", "A", "2024-06-15")],
        1,
    )
    assert seen == [("rows", "S", ""), ("summaries", "S", "")]


def test_group_folder_listing_subfolder_recursion_and_counts() -> None:
    """The recursive count per subfolder — every path equal to the
    folder or starting with ``folder + '/'`` (a doc under ``a/b/``
    counts for BOTH ``a`` and ``a/b``), path order, the stored summary
    attached or ``None``."""
    rows = [
        ("a/b/c.md", "C", "2024-06-15"),
        ("a/b/d.md", "D", "2024-06-15"),
        ("a/e.md", "E", "2024-06-15"),
        ("f.md", "F", "2024-06-15"),
    ]
    sub, files, total = agent.group_folder_listing(
        "S", "", rows, {"a": "A subtree.", "a/b": "B subtree."}
    )
    # ROOT level: the direct subfolders of "" are the TOP-LEVEL folders
    # only (a/b is nested under a, not direct) — a's count is its whole
    # recursive subtree (a/e.md + a/b/c.md + a/b/d.md), the stored
    # summary attached.
    assert sub == [("a", 3, "A subtree.")]
    assert files == [("S", "f.md", "F", "2024-06-15")]
    assert total == 1
    # One level down: a/b is a's direct subfolder with its own count.
    sub2, _files2, _total2 = agent.group_folder_listing("S", "a", rows, {"a/b": "B subtree."})
    assert sub2 == [("a/b", 2, "B subtree.")]


def test_group_folder_listing_nested_level_counts_and_membership() -> None:
    """One level down: ``a``'s direct subfolder is ``a/b`` (count 2),
    its own direct file is ``a/e.md`` (``a/b/c.md`` is NOT a direct
    file of ``a``) — membership is the folder_of rule, order is path
    order."""
    rows = [
        ("a/b/c.md", "C", "2024-06-15"),
        ("a/b/d.md", "D", "2024-06-15"),
        ("a/e.md", "E", "2024-06-15"),
    ]
    sub, files, total = agent.group_folder_listing("S", "a", rows, {})
    assert sub == [("a/b", 2, None)]
    assert files == [("S", "a/e.md", "E", "2024-06-15")]
    assert total == 1


def test_group_folder_listing_file_path_is_not_a_folder() -> None:
    """A document whose path is a prefix of NO other path is a file,
    never a folder: ``ls`` of it must not list a subfolder (and the
    ``path == folder`` count arm only fires for TRUE folders — a doc
    sharing a real folder's name counts for that folder, the existence
    rule intact)."""
    rows = [
        ("a.md", "A", "2024-06-15"),  # a file at the root, and a folder name? NO —
        ("b/x.md", "X", "2024-06-15"),  # nothing starts with "a.md/"
    ]
    sub, files, total = agent.group_folder_listing("S", "", rows, {})
    assert sub == [("b", 1, None)]  # "a.md" is NOT a subfolder
    assert files == [("S", "a.md", "A", "2024-06-15")]  # b/x.md is NOT a direct root file
    assert total == 1
    # The path == folder arm: a doc named "a" under a real folder "a/".
    rows2 = [("a", "FileA", "2024-06-15"), ("a/c.md", "C", "2024-06-15")]
    sub2, files2, total2 = agent.group_folder_listing("S", "", rows2, {})
    assert sub2 == [("a", 2, None)]  # the file "a" counts for folder "a"
    assert files2 == [("S", "a", "FileA", "2024-06-15")]  # …and is a direct ROOT file
    assert total2 == 1


def test_group_folder_listing_caps_files_at_fifty_keeps_the_total() -> None:
    """The cap: 51 direct files → 50 file lines + the PRE-cap total (51)
    for the renderer's note; 50 files → 50 lines, no note material.
    A 500-file folder costs 50 lines, never 500."""
    rows51 = [(f"big/f{i:03d}.md", f"T{i}", "2024-06-15") for i in range(51)]
    sub, files, total = agent.group_folder_listing("S", "big", rows51, {})
    assert sub == []
    assert total == 51
    assert len(files) == 50
    assert files[0] == ("S", "big/f000.md", "T0", "2024-06-15")
    assert files[-1] == ("S", "big/f049.md", "T49", "2024-06-15")
    rows50 = [(f"big/f{i:03d}.md", f"T{i}", "2024-06-15") for i in range(50)]
    _sub, files50, total50 = agent.group_folder_listing("S", "big", rows50, {})
    assert total50 == 50 and len(files50) == 50


# ---------- the pinned drill-down templates (byte-for-byte) ----------


def test_render_ls_top_template() -> None:
    assert (
        agent.render_ls_top(
            [("Deployments", 3, None), ("Homelab", 5, "Home lab notes.")]
        )
        == "2 sources:\n\n"
        "Deployments — 3 documents\n"
        "Homelab — 5 documents\n"
        "  Home lab notes."
    )
    assert agent.render_ls_top([]) == "0 sources:"


def test_render_folder_listing_root_template() -> None:
    assert (
        agent.render_folder_listing(
            "Homelab",
            [("backups", 2, "Backup notes."), ("networking", 1, None)],
            [("Homelab", "readme.md", "Readme", "2024-06-15")],
            1,
        )
        == "Homelab — 1 documents, 2 folders:\n"
        "\n"
        "  backups/ — 2 documents: Backup notes.\n"
        "  networking/ — 1 documents\n"
        "\n"
        "source: Homelab | path: readme.md | title: Readme | date: 2024-06-15"
    )


def test_render_folder_listing_empty_level_is_header_alone() -> None:
    """A registered source with no documents: the header line alone —
    the old ``0 documents:`` behavior preserved in spirit."""
    assert agent.render_folder_listing("Homelab", [], [], 0) == (
        "Homelab — 0 documents, 0 folders:"
    )


def test_render_folder_listing_subfolders_only_no_blank_trailer() -> None:
    """Subfolders but no own files: header + blank + subfolder lines —
    no trailing blank line, no file section."""
    assert (
        agent.render_folder_listing("S", [("a", 1, None)], [], 0)
        == "S — 0 documents, 1 folders:\n\n  a/ — 1 documents"
    )


def test_render_folder_listing_cap_note_only_past_fifty() -> None:
    """The note appears ONLY when the folder's own files outnumber the
    cap: 51 → 50 lines + the deterministic grep-pointer note (the
    ``…and 1 more…`` shape — unpluralized, the house pin); 50 → no
    note."""
    files51 = [("S", f"f{i:03d}.md", f"T{i}", "2024-06-15") for i in range(51)]
    capped = files51[:50]
    rendered = agent.render_folder_listing("S/big", [], capped, 51)
    lines = rendered.splitlines()
    assert lines[0] == "S/big — 51 documents, 0 folders:"
    assert len(lines) == 1 + 1 + 50 + 1  # header, blank, 50 lines, note
    assert lines[-1] == (
        "…and 1 more documents in this folder — use grep (pattern) to "
        "find a specific one."
    )
    files50 = [("S", f"f{i:03d}.md", f"T{i}", "2024-06-15") for i in range(50)]
    rendered50 = agent.render_folder_listing("S/big", [], files50, 50)
    assert rendered50.splitlines()[-1] == (
        "source: S | path: f049.md | title: T49 | date: 2024-06-15"
    )
    assert "more documents" not in rendered50


# ---------- read: the canonical combined source/path form ----------


def test_read_combined_path_resolves_and_returns_full_content(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The combined ``source/path`` form (the model's trained shape) is
    the canonical identity: split at the FIRST '/', one exact lookup,
    the full content returned (A7-revised: never truncated) — even when
    the path itself carries further slashes."""
    doc = _doc("Homelab", "active/container_caddy/caddy.md", "Caddy", "CADDY-CONTENT")
    calls: list[tuple[str, str]] = []

    def _find(db: Any, source: str, path: str) -> Document | None:
        calls.append((source, path))
        return (
            doc
            if (source, path) == ("Homelab", "active/container_caddy/caddy.md")
            else None
        )

    monkeypatch.setattr(agent, "find_document", _find)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="read",
                arguments={"path": "Homelab/active/container_caddy/caddy.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    # First-slash split — exactly one lookup, the canonical pair.
    assert calls == [("Homelab", "active/container_caddy/caddy.md")]
    assert holder.read_docs == [doc]
    assert holder.tool_calls == 1
    assert llm.requests[1][0][3]["content"] == (
        "Document Homelab/active/container_caddy/caddy.md:\n"
        "date: 2024-06-15\n"
        "CADDY-CONTENT"
    )


def test_read_bare_source_name_refused_without_db(monkeypatch: pytest.MonkeyPatch) -> None:
    """A bare source name (no '/') can never be a document — the
    no-document refusal (the argument echoed as passed), no DB lookup
    (NOT even the phase-72 candidate lookup — ``all_documents`` must
    not run either), nothing counted."""

    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError(
            "no DB lookup (find_document or all_documents) for a bare "
            "source name"
        )

    monkeypatch.setattr(agent, "find_document", _boom)
    monkeypatch.setattr(agent, "all_documents", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "Homelab"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'Homelab' — check the ls output."
    )
    assert llm.requests[1][1] == AGENT_TOOLS


def test_read_unknown_path_refused_echoing_argument(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """An unknown combined identity that matches NO indexed document's
    ``path`` (zero candidates — the phase-72 lookup runs, finds nothing)
    → today's refusal echoing the argument as passed, byte-identical —
    the old split-teaching refusal is gone (phase 70)."""
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: [])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/ghost.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'S/ghost.md' — check the ls output."
    )
    assert llm.requests[1][1] == AGENT_TOOLS  # tools stay offered (cap bounds)


# ---------- read/grep: the "did you mean …?" suggestions (phase 72, task 02) ----------


def test_find_path_candidates_exact_suffix_catalog_order(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The pure catalog lookup (monkeypatched ``all_documents`` — one
    bulk query per call): ``path`` == arg (exact) or a ``/arg`` suffix —
    catalog order, case-sensitive, as ``(source, path, title)`` triples;
    a plain substring is NOT a suffix; the result is uncapped (the
    :data:`~app.rag.agent.SUGGESTION_LIMIT` cap lives in the refusal).
    """
    docs = [
        _doc("A", "x.md", "Ax", "A"),
        _doc("A", "shared/x.md", "As", "AS"),
        _doc("B", "shared/x.md", "Bs", "BS"),
        _doc("C", "deep/shared/x.md", "Cs", "CS"),
        _doc("D", "X.md", "Dx", "D"),  # case-sensitive: not 'x.md'
        _doc("E", "nosuffixx.md", "Ex", "E"),  # substring, not a /suffix
    ]
    calls: list[int] = []

    def _all(db: Any) -> list[Document]:
        calls.append(1)
        return docs

    monkeypatch.setattr(agent, "all_documents", _all)
    db = cast("Session", object())

    # Exact bare path ('shared/x.md') plus the deeper suffix
    # ('deep/shared/x.md' ends with '/shared/x.md') — catalog order.
    assert agent.find_path_candidates(db, "shared/x.md") == [
        ("A", "shared/x.md", "As"),
        ("B", "shared/x.md", "Bs"),
        ("C", "deep/shared/x.md", "Cs"),
    ]
    # 'x.md' equals A's path exactly AND suffix-matches the rest — all
    # four, catalog order (uncapped: the cap is the refusal's).
    assert agent.find_path_candidates(db, "x.md") == [
        ("A", "x.md", "Ax"),
        ("A", "shared/x.md", "As"),
        ("B", "shared/x.md", "Bs"),
        ("C", "deep/shared/x.md", "Cs"),
    ]
    # Case-sensitive file paths: 'X.md' matches ONLY D's identically-
    # cased path (never the lowercase 'x.md' ones), and the plain
    # substring inside 'nosuffixx.md' is not a suffix.
    assert agent.find_path_candidates(db, "X.md") == [("D", "X.md", "Dx")]
    # One bulk query per call (at most one).
    assert len(calls) == 3


def test_read_bare_path_exact_match_gets_did_you_mean(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The incident shape: an unresolved ``read`` argument containing
    ``/`` that EXACTLY matches one indexed document's ``path`` (the bare
    path missing the source prefix — the harness prior) gets the
    ``NO_DOCUMENT_DID_YOU_MEAN`` line naming the combined identity —
    still a refusal: ``read_docs`` empty, nothing counted, tools stay
    offered."""
    doc = _doc("Homelab", "active/container_caddy/caddy.md", "Caddy", "CADDY-CONTENT")

    def _find(db: Any, source: str, path: str) -> Document | None:
        return (
            doc
            if (source, path) == ("Homelab", "active/container_caddy/caddy.md")
            else None
        )

    monkeypatch.setattr(agent, "find_document", _find)
    monkeypatch.setattr(agent, "all_documents", lambda db: [doc])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="read",
                arguments={"path": "active/container_caddy/caddy.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0  # a refusal
    assert llm.requests[1][0][3]["content"] == (
        agent.NO_DOCUMENT_DID_YOU_MEAN.format(
            arg="active/container_caddy/caddy.md",
            source="Homelab",
            path="active/container_caddy/caddy.md",
        )
    )
    # The rendered line, pinned byte-for-byte.
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'active/container_caddy/caddy.md' — "
        "did you mean 'Homelab/active/container_caddy/caddy.md'?"
    )
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_read_bare_path_suffix_match_gets_did_you_mean(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The suffix form of the same teaching: a path-like argument that
    matches a deeper indexed path (``active/container_caddy/caddy.md``
    ends with ``/container_caddy/caddy.md``) names the same combined
    identity."""
    doc = _doc("Homelab", "active/container_caddy/caddy.md", "Caddy", "CADDY-CONTENT")

    def _find(db: Any, source: str, path: str) -> Document | None:
        return (
            doc
            if (source, path) == ("Homelab", "active/container_caddy/caddy.md")
            else None
        )

    monkeypatch.setattr(agent, "find_document", _find)
    monkeypatch.setattr(agent, "all_documents", lambda db: [doc])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="read",
                arguments={"path": "container_caddy/caddy.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0  # a refusal
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'container_caddy/caddy.md' — "
        "did you mean 'Homelab/active/container_caddy/caddy.md'?"
    )


def test_read_bare_path_two_sources_gets_one_of_suggestion(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The same bare path under two sources: the ``one of`` line — up to
    ``SUGGESTION_LIMIT`` combined identities, each single-quoted, joined
    with ``, `` in catalog order (A before B)."""
    a = _doc("A", "shared/x.md", "Ax", "A")
    b = _doc("B", "shared/x.md", "Bx", "B")
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: [a, b])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(id="call_1", name="read", arguments={"path": "shared/x.md"})
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0  # a refusal
    assert llm.requests[1][0][3]["content"] == (
        agent.NO_DOCUMENT_DID_YOU_MEAN_MANY.format(
            arg="shared/x.md", candidates="'A/shared/x.md', 'B/shared/x.md'"
        )
    )
    # The rendered line, pinned byte-for-byte.
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'shared/x.md' — did you mean one of: "
        "'A/shared/x.md', 'B/shared/x.md'?"
    )


def test_read_bare_path_four_sources_capped_at_three_suggestions(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Four sources sharing the same path: exactly ``SUGGESTION_LIMIT``
    (3) identities are suggested — catalog order, the fourth dropped."""
    docs = [_doc(s, "shared/x.md", f"{s}x", s) for s in ("A", "B", "C", "D")]
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: list(docs))
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(id="call_1", name="read", arguments={"path": "shared/x.md"})
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'shared/x.md' — did you mean one of: "
        "'A/shared/x.md', 'B/shared/x.md', 'C/shared/x.md'?"
    )
    assert "'D/shared/x.md'" not in llm.requests[1][0][3]["content"]
    assert holder.read_docs == [] and holder.tool_calls == 0  # a refusal


def test_read_bare_filename_without_slash_keeps_no_db_refusal(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The gate is the ``/`` in the argument: a bare FILENAME (no ``/``
    — e.g. ``caddy.md``) is a bare name for the lookup — today's
    refusal byte-identical, and NO ``find_document`` / ``all_documents``
    call (the same no-DB-lookup invariant as a bare source name)."""

    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError("no DB lookup for a bare (no '/') argument")

    monkeypatch.setattr(agent, "find_document", _boom)
    monkeypatch.setattr(agent, "all_documents", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "caddy.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'caddy.md' — check the ls output."
    )
    assert llm.requests[1][1] == AGENT_TOOLS


def test_read_bare_path_of_seed_doc_gets_suggestion_then_dedupe(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dedupe precedence: the in-context dedupe fires on the SPLIT pair
    of the argument — the bare path of an in-context document
    (``read('app/rag/importer.py')`` with ``sample/app/rag/importer.py``
    seeded) is NOT that pair, so it is not a dedupe: it gets the
    suggestion line naming the combined identity, and the model's next,
    correctly-formed call is then deduped as ALREADY_IN_CONTEXT."""
    seed = [_doc("sample", "app/rag/importer.py", "Importer", "IMPORTER")]

    def _find(db: Any, source: str, path: str) -> Document | None:
        return seed[0] if (source, path) == ("sample", "app/rag/importer.py") else None

    monkeypatch.setattr(agent, "find_document", _find)
    monkeypatch.setattr(agent, "all_documents", lambda db: list(seed))
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="read",
                arguments={"path": "app/rag/importer.py"},
            )
        ],
        [
            # Round 2: the corrected call (the suggested combined
            # identity) — the seed document is already in context, so it
            # dedupes.
            ToolCallPiece(
                id="call_2",
                name="read",
                arguments={"path": "sample/app/rag/importer.py"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings(), seed_docs=seed))
    assert holder.read_docs == [] and holder.tool_calls == 0  # both refused
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'app/rag/importer.py' — "
        "did you mean 'sample/app/rag/importer.py'?"
    )
    assert llm.requests[2][0][5]["content"] == agent.ALREADY_IN_CONTEXT


@pytest.mark.parametrize(
    ("arguments", "label"),
    [
        ({}, "no arguments"),
        ({"path": ""}, "empty path"),
        ({"path": "   "}, "blank path"),
        ({"path": 7}, "non-string path"),
        ({"path": None}, "null path"),
    ],
)
def test_read_missing_arguments_refused(
    monkeypatch: pytest.MonkeyPatch, arguments: dict[str, Any], label: str
) -> None:
    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError(f"find_document must not be called ({label})")

    monkeypatch.setattr(agent, "find_document", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments=arguments)],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == agent.MISSING_READ_ARGS
    assert llm.requests[1][1] == AGENT_TOOLS


def test_reading_a_seed_doc_is_already_in_context(monkeypatch: pytest.MonkeyPatch) -> None:
    """The combined identity of a seeded document: its split pair is in
    the known set → ALREADY_IN_CONTEXT with no DB lookup (the dedupe
    check precedes the resolve)."""
    seed = [_doc("Homelab", "kubernetes.md", "Kubernetes", "K8S-CONTENT")]

    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError("find_document must not be called for a seeded doc")

    monkeypatch.setattr(agent, "find_document", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="read",
                arguments={"path": "Homelab/kubernetes.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings(), seed_docs=seed))
    assert holder.read_docs == [] and holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == agent.ALREADY_IN_CONTEXT
    # Rejected → the tools are still offered on the next request (the
    # round cap is the only bound).
    assert llm.requests[1][1] == AGENT_TOOLS


def test_reading_an_already_read_doc_is_deduped(monkeypatch: pytest.MonkeyPatch) -> None:
    """The second read of the same document (holder.read_docs) →
    ALREADY_IN_CONTEXT — appended once, counted once."""
    doc = _doc("S", "a.md", "A", "A-CONTENT")
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: doc)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/a.md"})],
        [ToolCallPiece(id="call_2", name="read", arguments={"path": "S/a.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [doc]  # appended exactly once
    assert holder.tool_calls == 1  # the re-read counts nothing
    assert llm.requests[1][0][3]["content"] == (
        "Document S/a.md:\ndate: 2024-06-15\nA-CONTENT"
    )
    assert llm.requests[2][0][5]["content"] == agent.ALREADY_IN_CONTEXT
    # Rejected → the tools are still offered on the next request…
    assert llm.requests[2][1] == AGENT_TOOLS


# ---------- phase 95: the read cap (bounded reads, honest truncation) ----------


def test_read_exactly_at_cap_is_byte_identical_and_untruncated(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95 boundary: a document whose content length EQUALS the cap
    fits — read whole, byte-identical to the pre-phase-95 result (no
    marker, no notice, no holder entry, no ``ToolResultPiece``)."""
    cap = 20
    content = "x" * cap
    doc = _doc("S", "big.md", "Big", content)
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: doc)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/big.md"})],
        [StreamPiece("content", "ans")],
    )
    out = asyncio.run(_run(llm, holder, _settings(read_max_chars=cap)))
    # The read result plus the phase-106 D5 date line (no marker, no
    # notice).
    assert llm.requests[1][0][3]["content"] == (
        "Document S/big.md:\ndate: 2024-06-15\n" + content
    )
    assert TRUNCATION_MARKER not in llm.requests[1][0][3]["content"]
    # No truncation recorded, none surfaced to the loop.
    assert holder.read_truncations == []
    assert not any(isinstance(p, ToolResultPiece) for p in out)
    # Still a successful read.
    assert holder.read_docs == [doc]
    assert holder.tool_calls == 1


def test_read_at_cap_plus_one_truncates_with_marker_and_notice(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95 boundary: ONE char over the cap truncates — the first
    ``cap`` chars + the shared :data:`TRUNCATION_MARKER` + the pinned
    grep-pointer notice (``{total}`` = the true length, ``{shown}`` = the
    cap), and the truncation is recorded on the holder. A truncated read
    is still a successful call (``tool_calls`` / ``read_docs`` as today)."""
    cap = 20
    content = "x" * (cap + 1)
    doc = _doc("S", "big.md", "Big", content)
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: doc)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/big.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings(read_max_chars=cap)))
    expected = (
        "Document S/big.md:\n"
        "date: 2024-06-15\n"
        + content[:cap]
        + "\n"
        + TRUNCATION_MARKER
        + "\n"
        + READ_TRUNCATION_NOTICE.format(shown=cap, total=cap + 1)
    )
    assert llm.requests[1][0][3]["content"] == expected
    # (argument, chars_shown, chars_total) — the raw argument the tool
    # frame carries, the cap kept, the true length.
    assert holder.read_truncations == [("S/big.md", cap, cap + 1)]
    assert holder.read_docs == [doc]  # still added to the context
    assert holder.tool_calls == 1  # still a counted, successful call


def test_read_truncation_notice_is_pinned() -> None:
    """Phase 95: the notice copy is pinned — it names the true length
    (``{total}``), the cap kept (``{shown}``), states the rest is NOT
    shown (the document did not end where it stopped), and points at
    ``grep`` (which searches the whole document)."""
    assert READ_TRUNCATION_NOTICE.format(shown=100, total=250) == (
        "TRUNCATED — this document is 250 characters; only the first "
        "100 are in your context. The rest is NOT shown. Use grep "
        "(pattern) to locate what you need — grep searches the whole "
        "document."
    )


def test_run_agent_yields_tool_result_after_tool_frame_before_next_round(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95: ``run_agent`` yields exactly ONE ``ToolResultPiece`` per
    truncated read — AFTER the round's ``tool`` frame (the matching
    ``ToolCallPiece``) and BEFORE the next model round (the answer
    pieces). It carries (argument, shown, total); ``argument`` is the
    same value the matching ``tool`` frame carries (the raw
    ``source/path`` the model passed)."""
    cap = 20
    content = "y" * (cap + 5)
    doc = _doc("S", "big.md", "Big", content)
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: doc)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/big.md"})],
        [StreamPiece("content", "ans")],
    )
    out = asyncio.run(_run(llm, holder, _settings(read_max_chars=cap)))
    # The piece order: the read's ToolCallPiece, then the ToolResultPiece,
    # then the next round's content (the answer).
    assert isinstance(out[0], ToolCallPiece) and out[0].name == "read"
    piece = out[1]
    assert isinstance(piece, ToolResultPiece)
    assert isinstance(out[2], StreamPiece)
    assert piece.name == "read"
    assert piece.argument == "S/big.md"
    assert piece.truncated is True
    assert piece.chars_shown == cap
    assert piece.chars_total == cap + 5
    # Exactly one ToolResultPiece for the one truncated read.
    assert [p for p in out if isinstance(p, ToolResultPiece)] == [piece]


def test_run_agent_short_read_yields_no_tool_result_piece(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95: a read at or under the cap yields NO ``ToolResultPiece``
    — the non-truncated stream is byte-identical to the pre-phase-95
    one (just the ``ToolCallPiece`` + the answer)."""
    cap = 20
    content = "z" * cap  # exactly at the cap → not truncated
    doc = _doc("S", "small.md", "Small", content)
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: doc)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/small.md"})],
        [StreamPiece("content", "ans")],
    )
    out = asyncio.run(_run(llm, holder, _settings(read_max_chars=cap)))
    assert not any(isinstance(p, ToolResultPiece) for p in out)
    assert holder.read_truncations == []
    # Order: the ToolCallPiece then the answer content (no piece between).
    assert isinstance(out[0], ToolCallPiece) and out[0].name == "read"
    assert isinstance(out[1], StreamPiece)


def test_read_truncation_does_not_touch_refusal_paths(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Phase 95: the read refusal paths are untouched by the cap — a SEED
    document that is over the cap is still refused with
    ``ALREADY_IN_CONTEXT`` (not truncated, nothing recorded, nothing
    counted), and an unknown path is still the no-document refusal (no
    content is read, so no truncation either)."""
    big = "B" * 5000  # far over the tiny cap below
    seed = _doc("S", "seed.md", "Seed", big)
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: [])
    # (a) Reading the (over-cap) seed doc → ALREADY_IN_CONTEXT (refusal).
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/seed.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings(read_max_chars=100), seed_docs=[seed]))
    assert llm.requests[1][0][3]["content"] == agent.ALREADY_IN_CONTEXT
    assert holder.read_truncations == []
    assert holder.tool_calls == 0 and holder.read_docs == []
    # (b) An unknown path → the no-document refusal (argument echoed),
    # even though a big doc could have truncated — no content is read.
    holder2 = AgentHolder()
    llm2 = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/missing.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm2, holder2, _settings(read_max_chars=100), seed_docs=[seed]))
    assert llm2.requests[1][0][3]["content"] == (
        "No document at 'S/missing.md' — check the ls output."
    )
    assert holder2.read_truncations == []
    assert holder2.tool_calls == 0 and holder2.read_docs == []


def test_unknown_tool_name_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="delete_universe", arguments={"x": 1})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [] and holder.tool_calls == 0
    assert llm.requests[1][0][3]["content"] == agent.UNKNOWN_TOOL
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


# ---------- grep (the phase-68 A5 contract under the new name) ----------


def test_grep_document_case_insensitive_line_numbers() -> None:
    """Case-insensitive fixed substring, 1-based line numbers, file order,
    repeated matches within a line collapse to one match (grep semantics)."""
    content = "The NEEDLE is here\nno hit\nneedle again\nNEEDLE NEEDLE\n"
    assert agent.grep_document(content, "NEEDLE") == [
        (1, "The NEEDLE is here"),
        (3, "needle again"),
        (4, "NEEDLE NEEDLE"),
    ]


def test_grep_document_rstrips_lines_and_empty_content() -> None:
    assert agent.grep_document("hello \t\nworld ", "WORLD") == [(2, "world")]
    assert agent.grep_document("", "x") == []
    assert agent.grep_document("no newlines", "NO") == [(1, "no newlines")]
    assert agent.grep_document("a\nb\n", "MISSING") == []


def test_grep_whole_kb_grep_style_output(monkeypatch: pytest.MonkeyPatch) -> None:
    """Whole-KB grep: catalog order, `source/path:line: text` lines,
    case-insensitive; the call counts in ``tool_calls`` and never touches
    ``read_docs``; the tools stay offered on the answer request."""
    d1 = _doc("Alpha", "a/one.md", "One", "first\nNEEDLE in one\nlast")
    d2 = _doc("Beta", "b/two.md", "Two", "no hit\nneedle in two\n")
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1, d2])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "needle"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "Alpha/a/one.md:2: NEEDLE in one\n"
        "Beta/b/two.md:2: needle in two"
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []  # locked A5: a grep adds no context
    assert llm.requests[1][1] == AGENT_TOOLS  # tools stay offered


def test_grep_capped_at_20_matches_in_catalog_order(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The 20-match cap is GLOBAL across documents in catalog order, and
    the scan stops once it is hit (a 35-match corpus yields exactly 20)."""
    d1 = _doc("S", "a.md", "A", "\n".join(f"hit-{i}" for i in range(15)))
    d2 = _doc("S", "b.md", "B", "\n".join(f"hit-{i}" for i in range(20)))
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1, d2])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "hit-"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    lines = llm.requests[1][0][3]["content"].split("\n")
    assert len(lines) == agent.SEARCH_MAX_MATCHES
    assert lines[0] == "S/a.md:1: hit-0"
    assert lines[14] == "S/a.md:15: hit-14"  # all of a.md
    assert lines[15] == "S/b.md:1: hit-0"  # then b.md, in order
    assert lines[19] == "S/b.md:5: hit-4"  # cut at the global cap
    assert holder.tool_calls == 1


def test_grep_truncates_match_lines_at_200_chars(monkeypatch: pytest.MonkeyPatch) -> None:
    """A 300-char match line yields exactly 200 chars of it (no crash)."""
    d1 = _doc("S", "a.md", "A", "top\n" + "x" * 300 + " NEEDLE tail")
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "needle"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == f"S/a.md:2: {'x' * agent.SEARCH_LINE_LIMIT}"
    assert holder.tool_calls == 1


# ---------- grep no-match teaching: the regex-shaped pattern
# (the 2026-09-05 "Qwen 3.8" incident — the harness prior is that
# grep takes a REGEX; this grep is a fixed substring, owner-locked
# A5, and the contract does not change) ----------


def test_plain_form_reduces_regex_to_literal_text() -> None:
    """The plain-form hint: the pattern reduced to literal text — the
    incident's exact recovery (``qwen.*3\\.8`` → ``qwen3.8``) plus the
    edge cases (raw ``.*`` runs dropped before unescape, so an escaped
    dot survives; first alternative only; classes/quantifiers/parens/
    anchors gone; whitespace preserved; pure metacharacters → ``""``).
    """
    assert agent.plain_form(r"qwen.*3\.8") == "qwen3.8"  # the incident
    assert agent.plain_form(r"qwen 3\.8") == "qwen 3.8"
    assert agent.plain_form(r"Qwen 3\.8") == "Qwen 3.8"  # case kept
    assert agent.plain_form(r"qwen3\.8") == "qwen3.8"
    assert agent.plain_form(r"llama\.cpp") == "llama.cpp"  # escaped dot kept
    assert agent.plain_form(r"qwen[0-9]+") == "qwen"  # class + quantifier
    assert agent.plain_form("a|b") == "a"  # first alternative only
    assert agent.plain_form(r"\d+") == ""  # no literal text — no hint
    assert agent.plain_form(r".*") == ""  # pure wildcard — no hint
    assert agent.plain_form(r"(qwen)3\.8") == "qwen3.8"  # group contents kept
    assert agent.plain_form("a{2,3}b") == "ab"
    assert agent.plain_form(r"^qwen$") == "qwen"  # anchors dropped
    assert agent.plain_form(r"a\.b") == "a.b"  # escaped dot is a literal
    assert agent.plain_form("plain") == "plain"  # identity for plain text


def test_looks_like_regex_detection() -> None:
    """One metacharacter anywhere marks the pattern regex-shaped; a
    plain substring (even with a space) does not."""
    for p in (
        r"qwen.*3\.8", r"qwen 3\.8", "qwen+", "a?b", "x|y", "(a)", "[a-z]", "a^b", "b$c", "a{2}"
    ):
        assert agent.looks_like_regex(p) is True, p
    for p in ("qwen 3.8", "qwen3.8", "plain substring", ""):
        assert agent.looks_like_regex(p) is False, p


def test_grep_no_match_regex_pattern_gets_teaching_line(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The incident's shape: a regex-shaped pattern that (necessarily)
    misses gets the TEACHING no-match line — the plain-substring
    contract stated, the plain-form retry hint handed over. Still a
    counted result; the context is untouched (locked A5)."""
    d1 = _doc("Alpha", "a/one.md", "One", "qwen3.8-27b juggernaut\nno regex text")
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1", name="grep", arguments={"pattern": r"qwen.*3\.8"}
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == agent.NO_MATCHES_REGEX.format(
        pattern=r"qwen.*3\.8", plain="qwen3.8"
    )
    assert llm.requests[1][0][3]["content"] == (
        "No matches for 'qwen.*3\\.8'. grep matches a plain substring "
        "(case-insensitive), not a regex — '.*', '\\.' and the like are "
        "literal text here, so that pattern can never match. Retry with "
        "the plain text you expect to see (e.g. 'qwen3.8')."
    )
    assert holder.tool_calls == 1  # a no-match with teaching is still a result
    assert holder.read_docs == []  # locked A5: a grep adds no context


def test_grep_no_match_regex_scoped_gets_teaching_line(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The scoped teaching variant: the resolved identity is echoed, the
    hint handed over."""
    d1 = _doc("Alpha", "a/one.md", "One", "nothing regex-shaped here")

    def _find(db: Any, source: str, path: str) -> Document | None:
        return d1 if (source, path) == ("Alpha", "a/one.md") else None

    monkeypatch.setattr(agent, "find_document", _find)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={"pattern": r"qwen 3\.8", "path": "Alpha/a/one.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No matches for 'qwen 3\\.8' in Alpha/a/one.md. grep matches a "
        "plain substring (case-insensitive), not a regex — retry with "
        "the plain text you expect to see (e.g. 'qwen 3.8')."
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []


def test_grep_no_match_plain_pattern_keeps_ordinary_line(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A no-match for a PLAIN pattern (no metacharacters — "qwen 3.8" with
    the space included) keeps the ordinary line byte-identical: the
    teaching never fires for a well-formed pattern (the retrieval side —
    the name-hit lexical signal — is what covers that case)."""
    d1 = _doc("Alpha", "a/one.md", "One", "qwen3.8-27b juggernaut")
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "qwen 3.8"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No matches for 'qwen 3.8' in the knowledge base."
    )
    assert holder.tool_calls == 1


def test_grep_matched_regex_pattern_returns_matches_not_teaching(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A pattern with metacharacters that MATCHES literally gets the
    ordinary match output — the teaching can never suppress a real hit
    (the detection keys on a NO-MATCH only)."""
    d1 = _doc("S", "a.md", "A", "the C++ compiler is here")
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "C++"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "S/a.md:1: the C++ compiler is here"
    assert holder.tool_calls == 1


def test_grep_no_match_regex_reducing_to_empty_falls_back(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A regex-shaped pattern with no literal text left after the
    reduction (``.*``) gets the ORDINARY line — no empty hint."""
    d1 = _doc("Alpha", "a/one.md", "One", "any text at all")
    monkeypatch.setattr(agent, "all_documents", lambda db: [d1])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": r".*"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No matches for '.*' in the knowledge base."
    )
    assert holder.tool_calls == 1


def test_no_matches_regex_templates_pin() -> None:
    """The teaching templates are verbatim pins (the model-facing copy —
    the mock E2E keys off the plain-substring clause)."""
    assert agent.NO_MATCHES_REGEX == (
        "No matches for '{pattern}'. grep matches a plain substring "
        "(case-insensitive), not a regex — '.*', '\\.' and the like are "
        "literal text here, so that pattern can never match. Retry with "
        "the plain text you expect to see (e.g. '{plain}')."
    )
    assert agent.NO_MATCHES_REGEX_SCOPED == (
        "No matches for '{pattern}' in {source}/{path}. grep matches a "
        "plain substring (case-insensitive), not a regex — retry with "
        "the plain text you expect to see (e.g. '{plain}')."
    )


def test_grep_scoped_to_one_document(monkeypatch: pytest.MonkeyPatch) -> None:
    """Scoped grep: only the named document is loaded (find_document on
    the first-slash split), ``all_documents`` never runs, and the match
    line carries its path."""
    d1 = _doc("S", "a.md", "A", "needle here")

    def _find(db: Any, source: str, path: str) -> Document | None:
        if (source, path) == ("S", "a.md"):
            return d1
        raise AssertionError(
            f"find_document({source}, {path}) — the scoped "
            "grep must not load any other document"
        )

    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError("all_documents must not run for a scoped grep")

    monkeypatch.setattr(agent, "find_document", _find)
    monkeypatch.setattr(agent, "all_documents", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={"pattern": "needle", "path": "S/a.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "S/a.md:1: needle here"
    assert holder.tool_calls == 1
    assert holder.read_docs == []  # grepped doc did not enter the context


def test_grep_scoped_combined_path_with_nested_path(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A combined target whose path itself contains slashes: the split is
    at the FIRST slash — the scoped grep runs on the right document."""
    d1 = _doc("S", "deep/nested/a.md", "A", "needle here")

    def _find(db: Any, source: str, path: str) -> Document | None:
        if (source, path) == ("S", "deep/nested/a.md"):
            return d1
        raise AssertionError(f"find_document({source}, {path}) — wrong first-slash split")

    monkeypatch.setattr(agent, "find_document", _find)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={"pattern": "needle", "path": "S/deep/nested/a.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "S/deep/nested/a.md:1: needle here"
    assert holder.tool_calls == 1


def test_grep_scoped_missing_document_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    """A scoped ``grep`` miss that matches NO indexed document's ``path``
    (zero candidates — the phase-72 lookup runs, finds nothing) keeps
    today's line byte-identical: a refusal (not counted), tools stay
    offered."""
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: [])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={"pattern": "x", "path": "S/ghost.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'S/ghost.md' — check the ls output."
    )
    assert holder.tool_calls == 0 and holder.read_docs == []  # a refusal
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_grep_scoped_missing_path_like_doc_gets_did_you_mean(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The same teaching on the scoped ``grep`` miss: an unresolved
    path-like scope that matches an indexed document's path gets the
    ``NO_DOCUMENT_DID_YOU_MEAN`` suggestion line (a refusal — not
    counted, no context added, tools stay offered); the whole-KB grep is
    untouched (no ``path`` argument → no scoped resolution at all).
    """
    doc = _doc("Homelab", "active/container_caddy/caddy.md", "Caddy", "CADDY-CONTENT")
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: [doc])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={
                    "pattern": "needle",
                    "path": "active/container_caddy/caddy.md",
                },
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        agent.NO_DOCUMENT_DID_YOU_MEAN.format(
            arg="active/container_caddy/caddy.md",
            source="Homelab",
            path="active/container_caddy/caddy.md",
        )
    )
    assert holder.tool_calls == 0 and holder.read_docs == []  # a refusal
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_grep_scoped_bare_source_name_refused_without_db(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A bare source name as the grep target can never resolve to exactly
    one document — the no-document refusal, no DB lookup, not counted."""
    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError("find_document must not run for a bare source name")

    monkeypatch.setattr(agent, "find_document", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={"pattern": "x", "path": "Homelab"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'Homelab' — check the ls output."
    )
    assert holder.tool_calls == 0 and holder.read_docs == []


@pytest.mark.parametrize(
    ("arguments", "label"),
    [
        ({}, "no arguments"),
        ({"pattern": ""}, "empty pattern"),
        ({"pattern": "   "}, "whitespace pattern"),
        ({"pattern": 42}, "non-string pattern"),
        ({"pattern": None}, "null pattern"),
    ],
)
def test_grep_missing_arguments_refused(
    monkeypatch: pytest.MonkeyPatch, arguments: dict[str, Any], label: str
) -> None:
    """A missing/blank/non-string pattern → the missing-args refusal, with
    no DB access at all."""

    def _boom(*_a: Any, **_k: Any) -> None:
        raise AssertionError(f"no DB access for a refused grep ({label})")

    monkeypatch.setattr(agent, "all_documents", _boom)
    monkeypatch.setattr(agent, "find_document", _boom)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments=arguments)],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == agent.MISSING_SEARCH_ARGS
    assert holder.tool_calls == 0 and holder.read_docs == []
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_grep_no_matches_whole_kb(monkeypatch: pytest.MonkeyPatch) -> None:
    """Zero hits across the KB → the no-match line (pattern quoted); the
    grep still executed, so it counts — and never adds context."""
    monkeypatch.setattr(
        agent, "all_documents", lambda db: [_doc("S", "a.md", "A", "nothing here")]
    )
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "zebra"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        "No matches for 'zebra' in the knowledge base."
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []


def test_grep_no_matches_scoped(monkeypatch: pytest.MonkeyPatch) -> None:
    """The scoped no-match line is keyed on the resolved source/path."""
    doc = _doc("S", "a.md", "A", "nothing here")

    def _find(db: Any, source: str, path: str) -> Document | None:
        return doc if (source, path) == ("S", "a.md") else None

    monkeypatch.setattr(agent, "find_document", _find)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(
                id="call_1",
                name="grep",
                arguments={"pattern": "zebra", "path": "S/a.md"},
            )
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == "No matches for 'zebra' in S/a.md."
    assert holder.tool_calls == 1
    assert holder.read_docs == []


def test_grep_no_match_truncates_long_pattern(monkeypatch: pytest.MonkeyPatch) -> None:
    """A pattern longer than 100 chars is truncated in the no-match line
    (kept short); the grep itself still runs on the full pattern."""
    monkeypatch.setattr(agent, "all_documents", lambda db: [])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "p" * 150})
        ],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert llm.requests[1][0][3]["content"] == (
        f"No matches for '{'p' * 100}' in the knowledge base."
    )
    assert holder.tool_calls == 1


def test_grep_counts_but_never_adds_context(monkeypatch: pytest.MonkeyPatch) -> None:
    """The locate-then-read workflow: a grep finds the document but does
    NOT add it — the subsequent read does (and is not rejected as
    already-in-context, because the grep touched nothing)."""
    doc = _doc("S", "a.md", "A", "needle here")
    monkeypatch.setattr(agent, "all_documents", lambda db: [doc])
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: doc)
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="grep", arguments={"pattern": "needle"})],
        [ToolCallPiece(id="call_2", name="read", arguments={"path": "S/a.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 2  # grep + read, both executed
    assert holder.read_docs == [doc]  # only the read added context (A5)
    assert llm.requests[1][0][3]["content"] == "S/a.md:1: needle here"
    assert llm.requests[2][0][5]["content"] == (
        "Document S/a.md:\ndate: 2024-06-15\nneedle here"
    )


# ---------- unlimited calls: re-lists and multi-reads (phase 45) ----------


def test_relist_executes_and_counts(monkeypatch: pytest.MonkeyPatch) -> None:
    """Re-lists execute — a second ``ls`` in one turn returns the top
    level again and counts in ``tool_calls`` (no budget to exhaust)."""
    monkeypatch.setattr(
        agent, "ls_top", lambda db: [("Deployments", 1, None), ("Homelab", 1, None)]
    )
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [ToolCallPiece(id="call_2", name="ls", arguments={})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.tool_calls == 2  # both re-lists executed and counted
    listing = "2 sources:\n\nDeployments — 1 documents\nHomelab — 1 documents"
    # The answer request carries the listing a second time as a tool result.
    assert llm.requests[2][0][3]["content"] == listing  # first listing
    assert llm.requests[2][0][5]["content"] == listing  # the re-list
    assert llm.requests[2][1] == AGENT_TOOLS  # still offered (no budgets)


def test_multi_read_executes_without_budgets(monkeypatch: pytest.MonkeyPatch) -> None:
    """Reads are no longer budgeted either — two different documents can
    be read in one turn (re-reading the same one is still deduped via
    ALREADY_IN_CONTEXT — see the rejection tests)."""
    a = _doc("S", "a.md", "A", "A-CONTENT")
    b = _doc("S", "b.md", "B", "B-CONTENT")
    monkeypatch.setattr(
        agent, "find_document", lambda db, source, path: {"a.md": a, "b.md": b}[path]
    )
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/a.md"})],
        [ToolCallPiece(id="call_2", name="read", arguments={"path": "S/b.md"})],
        [StreamPiece("content", "ans")],
    )
    asyncio.run(_run(llm, holder, _settings()))
    assert holder.read_docs == [a, b]  # both reads appended, in order
    assert holder.tool_calls == 2
    assert llm.requests[1][0][3]["content"] == (
        "Document S/a.md:\ndate: 2024-06-15\nA-CONTENT"
    )
    assert llm.requests[2][0][5]["content"] == (
        "Document S/b.md:\ndate: 2024-06-15\nB-CONTENT"
    )
    assert llm.requests[2][1] == AGENT_TOOLS  # the second read was still offered


# ---------- round cap (phase 45: replaces the per-tool budgets) ----------


def test_always_ls_bounded_by_round_cap(monkeypatch: pytest.MonkeyPatch) -> None:
    """A model that keeps calling ``ls`` gets exactly
    ``agent_max_rounds`` tool rounds, then one forced ``tools=None``
    request streams the answer — the cap is the only forced exit."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    listing = "1 sources:\n\nS — 1 documents"
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [ToolCallPiece(id="call_2", name="ls", arguments={})],
        [ToolCallPiece(id="call_3", name="ls", arguments={})],
        [StreamPiece("content", "forced answer")],
    )
    pieces = asyncio.run(_run(llm, holder, _settings(agent_max_rounds=3)))
    assert [type(p) for p in pieces] == [
        ToolCallPiece,
        ToolCallPiece,
        ToolCallPiece,
        StreamPiece,
    ]
    assert len(llm.requests) == 4  # 3 tool rounds + the forced answer
    # The three tool rounds were offered the tools…
    assert llm.requests[0][1] == AGENT_TOOLS
    assert llm.requests[1][1] == AGENT_TOOLS
    assert llm.requests[2][1] == AGENT_TOOLS
    # …and the forced final request carries no tools, whatever is left.
    assert llm.requests[3][1] is None
    # Every re-list executed and counted.
    assert holder.tool_calls == 3
    # The final request carries all three executed listings as history.
    final_msgs = llm.requests[3][0]
    assert len(final_msgs) == 8  # 2 + 3 rounds × (assistant + tool)
    assert final_msgs[3]["content"] == listing
    assert final_msgs[5]["content"] == listing
    assert final_msgs[7]["content"] == listing


def test_zero_max_rounds_is_one_request_without_tools() -> None:
    """``agent_max_rounds=0`` — the kill switch: exactly one request,
    ``tools=None``, no tool lines, no history growth (byte-identical to
    the pre-phase-37 path)."""
    holder = AgentHolder()
    llm = ScriptedLLM([StreamPiece("thinking", "t "), StreamPiece("content", "direct answer")])
    pieces = asyncio.run(_run(llm, holder, _settings(agent_max_rounds=0)))
    assert [type(p) for p in pieces] == [StreamPiece, StreamPiece]
    assert len(llm.requests) == 1
    assert llm.requests[0][1] is None
    assert llm.requests[0][0] == [
        {"role": "system", "content": "SYSTEM_PROMPT"},
        {"role": "user", "content": "QUESTION"},
    ]
    assert holder.read_docs == [] and holder.tool_calls == 0


def test_rejected_read_spam_runs_to_round_cap(monkeypatch: pytest.MonkeyPatch) -> None:
    """Every call rejected (unknown document — "No document at …"):
    rejections no longer end the loop early via budgets — the round cap
    bounds them and forces the final no-tools answer. Zero candidates
    (empty catalog) → the pre-phase-72 line, byte-identical."""
    monkeypatch.setattr(agent, "find_document", lambda db, source, path: None)
    monkeypatch.setattr(agent, "all_documents", lambda db: [])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="read", arguments={"path": "S/ghost.md"})],
        [ToolCallPiece(id="call_2", name="read", arguments={"path": "S/ghost.md"})],
        [ToolCallPiece(id="call_3", name="read", arguments={"path": "S/ghost.md"})],
        [StreamPiece("content", "forced answer")],
    )
    asyncio.run(_run(llm, holder, _settings(agent_max_rounds=3)))
    assert len(llm.requests) == 4  # 3 rejected rounds + the forced answer
    assert llm.requests[0][1] == AGENT_TOOLS
    assert llm.requests[1][1] == AGENT_TOOLS
    assert llm.requests[2][1] == AGENT_TOOLS
    assert llm.requests[3][1] is None  # the forced final request: no tools
    assert holder.read_docs == [] and holder.tool_calls == 0  # nothing executed
    refusal = "No document at 'S/ghost.md' — check the ls output."
    assert llm.requests[1][0][3]["content"] == refusal
    assert llm.requests[2][0][5]["content"] == refusal
    assert llm.requests[3][0][7]["content"] == refusal


# ---------- retries inside the agent loop (phase 67, locked A2) ----------


class FailingLLM:
    """A scripted fake whose Nth ``chat_stream`` call yields pieces and
    then raises (phase 67): ``attempts`` is a list of ``(pieces, error)``
    — an error after zero pieces = "the endpoint died before the first
    token"; after some pieces = a mid-stream drop. Records every
    request's messages/tools and the indices of the attempts whose stream
    teardown ran (``closed``)."""

    def __init__(
        self,
        attempts: list[tuple[list[StreamPiece | ToolCallPiece], Exception | None]],
    ) -> None:
        self.attempts = list(attempts)
        self.requests: list[tuple[list[dict[str, Any]], list[dict[str, Any]] | None]] = []
        #: Indices of attempts whose stream teardown has run.
        self.closed: list[int] = []

    def chat_stream(
        self,
        messages: list[dict[str, str]],
        tools: list[dict[str, Any]] | None = None,
        scaffolding: ScaffoldingFilter | None = None,  # phase 71: fed like the real client
    ) -> AsyncIterator[StreamPiece | ToolCallPiece]:
        index = len(self.requests)
        pieces, error = (
            self.attempts[index]
            if index < len(self.attempts)
            else ([], LLMError("script exhausted"))
        )
        self.requests.append(
            (deepcopy(messages), deepcopy(tools) if tools is not None else None)
        )
        return self._attempt(index, pieces, error, scaffolding)

    async def _attempt(
        self,
        index: int,
        pieces: list[StreamPiece | ToolCallPiece],
        error: Exception | None,
        scaffolding: ScaffoldingFilter | None,
    ) -> AsyncIterator[StreamPiece | ToolCallPiece]:
        try:
            for piece in pieces:
                if (
                    scaffolding is not None
                    and isinstance(piece, StreamPiece)
                    and piece.kind == "content"
                ):
                    cleaned = scaffolding.feed(piece.text)
                    if cleaned:
                        yield StreamPiece("content", cleaned)
                else:
                    yield piece
            if error is not None:
                # The tail is NOT flushed on a failed attempt — the real
                # client only flushes a cleanly completed stream.
                raise error
            if scaffolding is not None:
                tail = scaffolding.flush()
                if tail:
                    yield StreamPiece("content", tail)
        finally:
            self.closed.append(index)


def _record_sleeps(monkeypatch: pytest.MonkeyPatch) -> list[float]:
    """Monkeypatch ``asyncio.sleep`` (what ``chat_stream_retried`` awaits
    for the flat pre-retry delay) and record every awaited delay."""
    sleeps: list[float] = []

    async def fake_sleep(seconds: float) -> None:
        sleeps.append(seconds)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    return sleeps


def test_round_retried_before_first_piece(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A tool round that dies before its first piece is restarted with the
    same messages: the stream carries a RetryPiece BEFORE the tool call,
    the tool executes, the final answer streams, and the per-call log line
    is still emitted exactly once (retries are invisible to the loop)."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    holder = AgentHolder()
    llm = FailingLLM(
        [
            ([], LLMError("connection refused")),
            ([ToolCallPiece(id="call_1", name="ls", arguments={})], None),
            ([StreamPiece("content", "Done!")], None),
        ]
    )
    sleeps = _record_sleeps(monkeypatch)
    with caplog.at_level(logging.INFO, logger="app.agent"):
        pieces = asyncio.run(
            _run(llm, holder, _settings(agent_max_rounds=2, llm_retry_delay=2.5))
        )
    assert pieces == [
        RetryPiece(2, 4),  # default llm_retries=3 → 4 attempts
        ToolCallPiece(id="call_1", name="ls", arguments={}),
        StreamPiece("content", "Done!"),
    ]
    assert holder.tool_calls == 1
    assert holder.read_docs == []
    # The restart is byte-identical: same messages, same tools offered.
    assert len(llm.requests) == 3
    assert llm.requests[0] == llm.requests[1]
    assert llm.requests[0][1] == AGENT_TOOLS
    assert llm.requests[2][1] == AGENT_TOOLS  # the answer round still offered
    # The flat delay was awaited exactly once, before the retry.
    assert sleeps == [2.5]
    tool_logs = [r for r in caplog.records if r.getMessage().startswith("agent tool=")]
    assert len(tool_logs) == 1  # the retry did not re-run the tool or log
    assert tool_logs[0].getMessage() == "agent tool=ls args={} round=1/2"


def test_round_failure_after_first_piece_is_terminal(monkeypatch: pytest.MonkeyPatch) -> None:
    """Locked A2: a round that already streamed a piece fails the turn —
    the LLMError propagates out of ``run_agent``, no RetryPiece, no
    sleep, no second request, and the holder is untouched (the tool
    never ran)."""
    holder = AgentHolder()
    llm = FailingLLM(
        [([StreamPiece("content", "partial ")], LLMError("mid-stream drop"))]
    )
    sleeps = _record_sleeps(monkeypatch)

    async def drain() -> list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece]:
        out: list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece] = []
        with pytest.raises(LLMError, match="mid-stream drop"):
            async for piece in run_agent(
                cast("LLMClient", llm),
                lambda: _mock_session(),
                system_prompt="SYSTEM_PROMPT",
                user_message="QUESTION",
                seed_docs=[],
                settings=_settings(),
                holder=holder,
            ):
                out.append(piece)
        return out

    out = asyncio.run(drain())
    assert out == [StreamPiece("content", "partial ")]  # no RetryPiece
    assert len(llm.requests) == 1  # no retry
    assert sleeps == []
    assert holder.read_docs == [] and holder.tool_calls == 0


def test_forced_final_no_tools_call_is_retried(monkeypatch: pytest.MonkeyPatch) -> None:
    """The forced final request (round cap reached) goes through the same
    retry rule: a failure before its first piece yields a RetryPiece and
    restarts with ``tools=None``; the answer from the retry streams."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    holder = AgentHolder()
    llm = FailingLLM(
        [
            ([ToolCallPiece(id="call_1", name="ls", arguments={})], None),
            ([ToolCallPiece(id="call_2", name="ls", arguments={})], None),
            ([], LLMError("down at the cap")),
            ([StreamPiece("content", "forced answer")], None),
        ]
    )
    pieces = asyncio.run(_run(llm, holder, _settings(agent_max_rounds=2)))
    assert [type(p) for p in pieces] == [
        ToolCallPiece,
        ToolCallPiece,
        RetryPiece,
        StreamPiece,
    ]
    assert pieces[2] == RetryPiece(2, 4)
    assert pieces[3] == StreamPiece("content", "forced answer")
    assert len(llm.requests) == 4  # 2 tool rounds + the final + its retry
    # The forced final (and its retry) carry no tools, whatever is left.
    assert llm.requests[2][1] is None
    assert llm.requests[3][1] is None
    # …and the restart is byte-identical.
    assert llm.requests[2][0] == llm.requests[3][0]
    assert holder.tool_calls == 2


def test_zero_retries_is_one_plain_attempt(monkeypatch: pytest.MonkeyPatch) -> None:
    """The kill-switch path (``llm_retries=0``): a dead round raises
    immediately — one request, no RetryPiece, no sleep (pre-phase-67
    behavior)."""
    holder = AgentHolder()
    llm = FailingLLM([([], LLMError("connection refused"))])
    sleeps = _record_sleeps(monkeypatch)

    async def drain() -> list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece]:
        out: list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece] = []
        with pytest.raises(LLMError, match="connection refused"):
            async for piece in run_agent(
                cast("LLMClient", llm),
                lambda: _mock_session(),
                system_prompt="SYSTEM_PROMPT",
                user_message="QUESTION",
                seed_docs=[],
                settings=_settings(llm_retries=0),
                holder=holder,
            ):
                out.append(piece)
        return out

    out = asyncio.run(drain())
    assert out == []  # nothing streamed, no RetryPiece
    assert len(llm.requests) == 1
    assert sleeps == []
    assert holder.read_docs == [] and holder.tool_calls == 0


def test_abandon_mid_retry_sleep_leaks_nothing(monkeypatch: pytest.MonkeyPatch) -> None:
    """Consumer abandon while a retried round is parked in the pre-retry
    sleep (client disconnect): the driving task is cancelled cleanly, the
    production teardown ``aclose()`` on ``run_agent`` does not raise, the
    inner attempt's stream was torn down, and the retry never starts."""
    entered = asyncio.Event()

    async def parking_sleep(seconds: float) -> None:
        entered.set()
        await asyncio.Event().wait()  # park until the abandon arrives

    monkeypatch.setattr(asyncio, "sleep", parking_sleep)
    holder = AgentHolder()
    llm = FailingLLM(
        [([], LLMError("endpoint down")), ([StreamPiece("content", "never")], None)]
    )

    async def run() -> None:
        gen = run_agent(
            cast("LLMClient", llm),
            lambda: _mock_session(),
            system_prompt="SYSTEM_PROMPT",
            user_message="QUESTION",
            seed_docs=[],
            settings=_settings(),
            holder=holder,
        )

        async def consumer() -> list[
            StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece
        ]:
            return [p async for p in gen]

        task = asyncio.ensure_future(consumer())
        await entered.wait()  # the round's retry is parked in the sleep
        assert not task.done()
        task.cancel()  # client disconnect: the driving task is cancelled
        with pytest.raises(asyncio.CancelledError):
            await task
        # Production teardown (phase 48 pattern): must not raise. ``run_agent``
        # is an async generator despite its AsyncIterator annotation.
        await cast(
            "AsyncGenerator[StreamPiece | ToolCallPiece | RetryPiece, None]", gen
        ).aclose()

    asyncio.run(run())
    assert len(llm.requests) == 1  # the retry never started
    assert llm.closed == [0]  # attempt 1's inner stream was torn down
    assert holder.read_docs == [] and holder.tool_calls == 0


def test_retries_are_invisible_to_the_round_cap(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A failing-then-succeeding round consumes ONE round: with a cap of
    2, the retried first round and the second tool round fill the cap —
    the forced final follows the SECOND call, and the log lines read
    round=1/2 and round=2/2."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    holder = AgentHolder()
    llm = FailingLLM(
        [
            ([], LLMError("down")),
            ([ToolCallPiece(id="call_1", name="ls", arguments={})], None),
            ([ToolCallPiece(id="call_2", name="ls", arguments={})], None),
            ([StreamPiece("content", "forced answer")], None),
        ]
    )
    with caplog.at_level(logging.INFO, logger="app.agent"):
        pieces = asyncio.run(_run(llm, holder, _settings(agent_max_rounds=2)))
    assert [type(p) for p in pieces] == [
        RetryPiece,
        ToolCallPiece,
        ToolCallPiece,
        StreamPiece,
    ]
    assert len(llm.requests) == 4  # 2 (round 1 + its retry) + 1 + the forced final
    assert llm.requests[3][1] is None  # the forced final, after round 2
    assert holder.tool_calls == 2
    msgs = [r.getMessage() for r in caplog.records]
    assert "agent tool=ls args={} round=1/2" in msgs
    assert "agent tool=ls args={} round=2/2" in msgs
    assert any("round cap reached (rounds=2)" in m for m in msgs)


# ---------- prompts: <tools> section (HIGH only) ----------
# NOTE (phase 70, task 02): these pins cover the phase-70 TOOLS_SECTION
# copy — the harness-aligned ls/read/grep names (the old phase-37/68
# names and the phase-37 per-tool budget line are gone; the round cap
# is the bound, not re-stated in the prompt, phase 45). The
# LOW/deflection path is untouched by this phase.


def test_high_prompt_carries_tools_section_after_documents() -> None:
    prompt = build_high_prompt([_doc("S", "a.md", "A", "A-CONTENT")])
    assert TOOLS_SECTION in prompt
    # Phase 70: the harness-aligned ls/read/grep copy.
    assert "`ls`" in prompt
    assert "`grep`" in prompt
    assert "`read`" in prompt
    assert "source: X | path: Y | title: Z" in prompt
    assert "locator, not a context-adder" in prompt
    assert "combined `source/path`" in prompt
    assert "Answer as soon as you have what you need" in prompt
    # The old names and the per-tool budget restatement are gone.
    for old in ("list_documents", "read_document", "search_documents"):
        assert old not in prompt
    assert "more than one extra document" not in prompt
    # After the mode body: <tools> follows </documents>.
    assert prompt.index("</documents>") < prompt.index("<tools>")
    assert prompt.rstrip().endswith("</tools>")


def test_high_prompt_tools_section_with_notes_and_kb() -> None:
    prompt = build_high_prompt(
        [_doc("S", "a.md", "A", "A-CONTENT")], notes=["be concise"], kb_overview="- KB"
    )
    assert prompt.index("<knowledge_base>") < prompt.index("<tuning>")
    assert prompt.index("<tuning>") < prompt.index("<documents>")
    assert prompt.index("<documents>") < prompt.index("<tools>")


def test_low_prompt_is_byte_identical_and_tool_free() -> None:
    # Phase 71: the LOW prompt carries the owner-permitted plain-text
    # line after the DEFLECT_MODE sentence (the marker-keying contract
    # is unchanged; the line must not leak into the HIGH prompt —
    # pinned in tests/unit/test_prompts.py).
    expected = (
        _base("LOW")
        + "\nDEFLECT_MODE: retrieval was weak — the titles below are the closest "
        "your notes come to the question. They are titles only; do not pretend "
        "they answer it. Use them to propose 2-3 alternative questions.\n"
        "Reply in plain text only — you have no tools in this mode.\n"
        + "- T1\n- T2"
    )
    assert build_deflect_prompt(["T1", "T2"]) == expected
    for prompt in (
        build_deflect_prompt(["T1"]),
        build_deflect_prompt(["T1"], notes=["be concise"]),
        build_deflect_prompt(["T1"], kb_overview="- KB"),
        build_deflect_prompt(["T1"], notes=["be concise"], kb_overview="- KB"),
    ):
        assert "<tools>" not in prompt
        assert TOOLS_SECTION not in prompt


# ---------- phase 71: the scaffolding recovery policy (deterministic only) ----------

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


def test_correction_instruction_is_the_harness_constant() -> None:
    """Verbatim constant: the E2E mock (task 05) keys on a stable
    substring of it, so it must not drift."""
    assert agent.CORRECTION_INSTRUCTION == (
        "Your previous reply contained raw tool-call markup, which is not "
        "interpreted here. Answer the user's question directly in plain "
        "text — no tool syntax."
    )


def test_malformed_reply_error_subclasses_llm_error() -> None:
    assert issubclass(MalformedReplyError, LLMError)
    assert not issubclass(LLMError, MalformedReplyError)


def test_scaffolding_only_round_gets_exactly_one_recovery(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A round whose visible content is pure scaffolding → exactly TWO
    model requests: the normal round, then the ONE recovery —
    ``tools=None`` with :data:`CORRECTION_INSTRUCTION` folded into the
    original single system message (the user message stays last). The
    clean recovery answer ends the turn, the holder is untouched by the
    recovery, and the strip was captured in the warning log."""
    holder = AgentHolder()
    llm = ScriptedLLM(
        [StreamPiece("content", _INCIDENT_SPAN)],
        [StreamPiece("content", "The clean recovery answer.")],
    )
    with caplog.at_level(logging.WARNING, logger="app.agent"):
        pieces = asyncio.run(_run(llm, holder, _settings()))
    # Nothing of the round's scaffolding was yielded — only the recovery.
    assert pieces == [StreamPiece("content", "The clean recovery answer.")]
    assert len(llm.requests) == 2  # the round + the one recovery
    # The round: the original system prompt, the tools offered.
    assert llm.requests[0][1] == AGENT_TOOLS
    assert llm.requests[0][0] == [
        {"role": "system", "content": "SYSTEM_PROMPT"},
        {"role": "user", "content": "QUESTION"},
    ]
    # The recovery: no tools, a SINGLE system message carrying the folded
    # correction, the user message last.
    assert llm.requests[1][1] is None
    assert llm.requests[1][0] == [
        {
            "role": "system",
            "content": "SYSTEM_PROMPT\n" + agent.CORRECTION_INSTRUCTION,
        },
        {"role": "user", "content": "QUESTION"},
    ]
    # The recovery is a fixed policy, not a conversation: the holder is
    # untouched by it.
    assert holder.read_docs == [] and holder.tool_calls == 0
    # The turn total feeds the API layer's ``scaffold_stripped=N`` field.
    assert holder.scaffold_stripped == len(_INCIDENT_SPAN)
    # One strip warning per stripped span, the span truncated to 200 chars.
    strip_logs = [
        r
        for r in caplog.records
        if r.levelno == logging.WARNING and r.getMessage().startswith("agent: stripped")
    ]
    assert len(strip_logs) == 1
    message = strip_logs[0].getMessage()
    assert message.startswith(
        f"agent: stripped {len(_INCIDENT_SPAN)} chars of tool-scaffolding in round 1:"
    )
    assert _INCIDENT_SPAN[:200] in message


def test_scaffolding_twice_settles_with_malformed_reply(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The recovery answer is scaffolding again (a second empty reply) —
    terminal: :class:`MalformedReplyError` (an :class:`LLMError` subclass)
    after exactly two requests — no third request, no recovery of a
    recovery."""
    holder = AgentHolder()
    llm = ScriptedLLM(
        [StreamPiece("content", _INCIDENT_SPAN)],
        [StreamPiece("content", _INCIDENT_SPAN)],
    )
    with pytest.raises(MalformedReplyError) as excinfo:
        asyncio.run(_run(llm, holder, _settings()))
    assert isinstance(excinfo.value, LLMError)
    assert len(llm.requests) == 2  # exactly one recovery per turn
    assert llm.requests[1][1] is None
    assert agent.CORRECTION_INSTRUCTION in llm.requests[1][0][0]["content"]
    assert holder.scaffold_stripped == 2 * len(_INCIDENT_SPAN)


def test_scaffolding_with_real_content_needs_no_recovery(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A round with real visible content PLUS scaffolding: the clean
    content stands — one request only, the clean remainder yielded (no
    raw tokens), no correction in any system prompt."""
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            StreamPiece("content", "Here it is: "),
            StreamPiece("content", _INCIDENT_SPAN),
            StreamPiece("content", " hope that helps."),
        ],
    )
    pieces = asyncio.run(_run(llm, holder, _settings()))
    assert [p for p in pieces if isinstance(p, StreamPiece)] == [
        StreamPiece("content", "Here it is: "),
        StreamPiece("content", " hope that helps."),
    ]
    assert len(llm.requests) == 1  # no recovery
    for messages, _tools in llm.requests:
        assert all(agent.CORRECTION_INSTRUCTION not in m["content"] for m in messages)
    assert holder.scaffold_stripped == len(_INCIDENT_SPAN)


def test_clean_turn_carries_no_correction_and_no_strip(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A clean turn: one request, no correction in any system prompt,
    zero stripped (the log field stays 0 — uniform)."""
    holder = AgentHolder()
    llm = ScriptedLLM(
        [StreamPiece("thinking", "hmm "), StreamPiece("content", "a clean answer")]
    )
    pieces = asyncio.run(_run(llm, holder, _settings()))
    assert pieces == [
        StreamPiece("thinking", "hmm "),
        StreamPiece("content", "a clean answer"),
    ]
    assert len(llm.requests) == 1
    for messages, _tools in llm.requests:
        assert all(agent.CORRECTION_INSTRUCTION not in m["content"] for m in messages)
    assert holder.scaffold_stripped == 0


def test_empty_round_without_a_strip_keeps_today_behavior(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Round content 0 with NOTHING stripped (an empty/thinking-only
    answer) → return as today — no recovery, no error."""
    holder = AgentHolder()
    llm = ScriptedLLM([StreamPiece("thinking", "nothing to say, honestly")])
    pieces = asyncio.run(_run(llm, holder, _settings()))
    assert pieces == [StreamPiece("thinking", "nothing to say, honestly")]
    assert len(llm.requests) == 1
    assert holder.scaffold_stripped == 0


def test_scaffolding_round_with_tool_calls_needs_no_recovery(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A scaffolding-only round that ALSO carried tool calls: the tool
    ran, and the policy keys on the no-calls exit only — no recovery (the
    next round is a normal tools-offered round carrying the tool
    history)."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [
            StreamPiece("content", _INCIDENT_SPAN),
            ToolCallPiece(id="call_1", name="ls", arguments={}),
        ],
        [StreamPiece("content", "the answer after the tool")],
    )
    pieces = asyncio.run(_run(llm, holder, _settings()))
    # The round's scaffolding was stripped (no raw delta), the tool frame
    # and the next round's answer flowed on.
    assert pieces == [
        ToolCallPiece(id="call_1", name="ls", arguments={}),
        StreamPiece("content", "the answer after the tool"),
    ]
    assert len(llm.requests) == 2
    assert llm.requests[1][1] == AGENT_TOOLS  # a normal round, not a recovery
    for messages, _tools in llm.requests:
        assert all(
            m["content"] is None or agent.CORRECTION_INSTRUCTION not in m["content"]
            for m in messages
        )
    assert holder.tool_calls == 1
    assert holder.scaffold_stripped == len(_INCIDENT_SPAN)


def test_recovery_after_tool_rounds_keeps_the_history(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A scaffolding-only answer round after a tool round: the recovery
    keeps the SINGLE (folded) system message at the front and the tool
    history intact behind it — no second system message, no duplicated
    correction."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [StreamPiece("content", _INCIDENT_SPAN)],
        [StreamPiece("content", "recovered after a tool round")],
    )
    pieces = asyncio.run(_run(llm, holder, _settings()))
    assert pieces == [
        ToolCallPiece(id="call_1", name="ls", arguments={}),
        StreamPiece("content", "recovered after a tool round"),
    ]
    assert len(llm.requests) == 3  # tool round + stripped round + recovery
    assert llm.requests[2][1] is None
    recovered = llm.requests[2][0]
    assert recovered[0] == {
        "role": "system",
        "content": "SYSTEM_PROMPT\n" + agent.CORRECTION_INSTRUCTION,
    }
    assert recovered[1] == {"role": "user", "content": "QUESTION"}
    assert len(recovered) == 4
    assert recovered[2]["role"] == "assistant"
    assert recovered[3] == {
        "role": "tool",
        "tool_call_id": "call_1",
        "content": "1 sources:\n\nS — 1 documents",
    }
    assert sum(1 for m in recovered if m["role"] == "system") == 1
    assert holder.scaffold_stripped == len(_INCIDENT_SPAN)


def test_forced_final_scaffolding_only_settles_malformed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The round-cap forced final (``tools=None``) is filtered too: a
    scaffolding-only forced answer never reaches the user raw — the turn
    settles with :class:`MalformedReplyError` (the same terminal
    semantics; this turn used no recovery, so nothing is doubled up)."""
    monkeypatch.setattr(agent, "ls_top", lambda db: [("S", 1, None)])
    holder = AgentHolder()
    llm = ScriptedLLM(
        [ToolCallPiece(id="call_1", name="ls", arguments={})],
        [StreamPiece("content", _INCIDENT_SPAN)],
    )
    with pytest.raises(MalformedReplyError):
        asyncio.run(_run(llm, holder, _settings(agent_max_rounds=1)))
    assert len(llm.requests) == 2
    assert llm.requests[1][1] is None  # the forced final — no recovery after it
    assert holder.scaffold_stripped == len(_INCIDENT_SPAN)