brain-of-reese/app/rag/agent.py

"""Agent loop: the grounded-turn document tools (phase 37, task 03; the
harness-aligned ``ls``/``read``/``grep`` surface, phase 70; the
drill-down tree ``ls`` + sync-time folder summaries, phase 94).

Probe verdict (task 01 — ``uv run python -m scripts.llm_probe --tools``
run live against aipi): **``probe: turbo tool_calls=supported 2026-08-26``**
— ``turbo`` answers OpenAI ``tools`` requests with
``finish_reason="tool_calls"`` and streams the calls as indexed
``delta.tool_calls`` partials (id + name on the first partial, arguments
in fragments). This module therefore uses the **native tool-calling
path**: tool calls arrive as :class:`app.rag.llm.ToolCallPiece` values
from ``chat_stream(messages, tools=AGENT_TOOLS)``. The prompt-based
JSON-block fallback (documented in the task file) is *not* implemented —
it exists only for a "not supported"/"intermittent" verdict, and the
probe came back "supported".

Real-model gate (phase 72, task 05 — live vs the configured chat
model; re-run 2026-09-04 on the controlled fixture KB — see
``TOOL_CALLING_TESTING.md``): the bare-path teaching (did-you-mean
refusals) makes every trap self-correct in exactly one round — zero
cap hits, zero repeat loops. Locked derived battery (phase 72,
executed ≥ 90 % bar): ``gate: lite FAIL turns=10 answered=10 caps=0
tool-turns=10 calls 5/15 executed (33%) contract 12/15 (80%) 2026-09-04
(wall 47.7s)`` — the bar is blocked by :data:`ALREADY_IN_CONTEXT` dedupe
refusals on the corrected re-reads, a copy-invariant model behavior
(five copy variants, 2026-09-03 → 04) and an app-semantics decision
(TOOL_CALLING_TESTING.md §7). Controlled fixture battery (the 2026-09-04
methodology — contract accuracy ≥ 90 %): PASS on three consecutive runs,
``contract 11/11 (100%)`` / ``12/13 (92%)`` / ``11/11 (100%)``.

Loop contract (one grounded chat turn; the API layer wires this in,
task 04):

1. The model is offered the three OpenAI functions in :data:`AGENT_TOOLS`
   for the whole turn — phase 45 removed the phase-37 per-tool budgets
   (owner permission 2026-08-27, ``TODO.md`` L8: "allow the LLM to make
   as many tool calls as it wants"): ``ls``, ``read`` and ``grep`` can
   each be called as many times as the model needs, re-lists and
   re-greps included. With ``settings.agent_max_rounds``
   (``BOR_AGENT_MAX_ROUNDS``, default 10) at 0 the loop makes exactly
   one request with ``tools=None`` — byte-identical to the
   pre-phase-37 chat path (the kill switch). Phase 70 (owner permission
   2026-09-03: "match existing harnesses as much as possible") renamed
   and reshaped the tools to the harness-trained surface —
   ``ls(path?)`` / ``read(path)`` / ``grep(pattern, path?)``, the
   pi.dev tool shapes the model was trained on: the combined
   ``source/path`` string is the canonical document identity in every
   tool argument, refusal, and result header, and the old two-argument
   split (with its self-correction and "teach the split" refusals) is
   gone — the model's combined form is now simply correct. The
   phase-68 A5 match/output contract rides along under the new name.
   Phase 94 revision (owner permission 2026-09-10, ``TODO.md`` L4 — the
   tool-surface revision recorded in the phase 94 overview ``00_phase.md``;
   PLAN.md is being redone by the owner): the ``ls`` RESULT format and
   ``path`` semantics changed — ``ls`` is now a filesystem-style
   drill-down tree (no path: the synced sources with counts + stored
   source-root summaries; a source name: its top-level folders + files;
   a ``source/folder`` path: that folder's subfolders + files — the
   model drills one level per call instead of flooding the whole
   catalog into one result), while the tool NAME and the
   ``read``/``grep`` contract (combined ``source/path``) are untouched.
2. Each tool call the model emits is executed server-side against
   Postgres only (no LLM, no network): ``ls`` lists ONE level of the
   drill-down tree (phase 94, task 03) — no ``path``: every registered
   source (registry order; a 0-document source still lists) as
   ``{source} — {n} documents`` plus the indented stored source-root
   summary line when one is in ``folder_summaries``; a ``path`` that is
   a registered source name (no ``/``): that source's root folder —
   each direct subfolder ``  {sub}/ — {m} documents[: {summary}]``
   (the count is the subfolder's recursive subtree — every document
   whose path equals the folder or starts with ``folder + "/"``, the
   same set the sync-time folder summary describes — and the file
   lines ``source: X | path: Y | title: Z | date: YYYY-MM-DD`` (the
   canonical ``read``/``grep`` identity — the phase-63 labeled format —
   plus the phase-106 D5 ``date`` field APPENDED after ``title``; only
   FILE lines carry a date — source/folder lines are not documents)
   in path order (``GET /api/docs`` order), capped at
   :data:`LS_MAX_FILE_LINES` lines + one deterministic grep-pointer
   note for the rest (a 500-file folder costs 50 lines, never 500);
   a ``source/folder`` ``path``: that folder's subfolders + own file
   lines (the same template, ``identity = source + "/" + folder``);
   a registered source with no documents lists its header line alone
   (``… — 0 documents, 0 folders:`` — the old ``0 documents:``
   behavior preserved in spirit) — every successful listing (top/root/
   folder) counts; a ``source/…`` argument whose first segment names no
   registered source is the no-source refusal (below, the segment
   echoed), and a folder segment matching no indexed prefix is the
   :data:`NOT_A_FOLDER` teaching (below) —
   ``read`` takes the combined ``source/path`` string, splits it at the
   FIRST ``'/'`` (source names are directory basenames — they can never
   contain ``'/'``), and returns the document's content — WHOLE at or
   under ``settings.read_max_chars`` (``BOR_READ_MAX_CHARS``, default
   128 000 chars ≈ 32k tokens), and CAPPED above it (phase 95, owner
   permission 2026-09-10, ``TODO.md`` L5): the first ``read_max_chars``
   chars plus the shared :data:`TRUNCATION_MARKER` and the pinned
   :data:`READ_TRUNCATION_NOTICE` (the rest is NOT in the model's
   context; the notice teaches the phase-131 workflow — ``grep``
   reports line numbers, then ``read`` that part with ``offset`` /
   ``limit``), with the truncation recorded on the holder so the loop
   yields a :class:`app.rag.llm.ToolResultPiece` (task 02 → the SSE
   ``tool_result`` frame + UI marker). Phase 131 (task 02): ``read``
   gains the OPTIONAL line range — ``offset`` (1-based first line) +
   ``limit`` (line count; ``offset`` alone reads to the end, ``limit``
   alone counts from line 1 — :func:`_read_range_args` validates
   whole ints ≥ 1, a bad value is the :data:`BAD_RANGE_ARGS` refusal
   with the values echoed): a range read returns EXACTLY the named
   lines under a third ``lines: {offset}-{end} of {total_lines}``
   header line (the true extent — an overshooting ``limit`` clamps to
   the last line), the cap applies to the SLICE (the holder tuple's
   ``chars_total`` is then the slice's length, not the document's),
   and an ``offset`` past the end is the :data:`OFFSET_PAST_END`
   teaching. The dedupe is COVERAGE-based (phase 141, task 02 —
   :data:`AgentHolder.read_coverage` records the line span each
   successful read actually put in the model's context, and the merged
   spans decide): a range read PROCEEDS unless the requested lines are
   FULLY covered by earlier reads (the owner's stuck-scenario fix —
   two DIFFERENT ranges of a document both reach the model, and the
   phase-131 truncated-then-range row is covered by the state — a
   truncated first read covers only its cut span); a re-read of
   FULLY-covered lines (a read-in-full document, or an earlier span)
   gets :data:`ALREADY_IN_CONTEXT` — the claim is then TRUE (the named
   lines ARE in the prompt); a PLAIN re-read of a PARTIALLY-covered
   document gets the :data:`REREAD_PARTIAL` teaching (the covered
   span(s) named in line form, read the rest with ``offset``/
   ``limit``) instead of the (wrong) full-text claim. **A6
   re-revised contract
   (phase 118, owner directive 2026-09-15):** the retrieval
   ``<documents>`` path seeds SUMMARIES only — a suggested document's
   full text never enters the prompt on the retrieval path; full text
   enters the context ONLY through this ``read`` TOOL path, which is
   the only capped read (the phase-95 cap unchanged). And ``grep``
   greps the
   indexed documents (or the one document a combined ``source/path``
   names) for a case-insensitive fixed substring and returns up to 20
   ``source/path:line: text`` match lines (owner-locked A5, phase 68),
   each line truncated to 200 chars. A grep is a **locator**, not a
   context-adder: it never appends to the answer context (only
   ``read`` does — ``holder.read_docs`` is untouched by a grep).
   The 2026-09-05 incident teaching (the harness prior is that grep
   takes a REGEX; this grep is a fixed substring and the contract does
   not change): when a grep RAN but found nothing and the pattern is
   regex-shaped (:func:`looks_like_regex`) with a non-empty
   :func:`plain_form`, the no-match result is the TEACHING line —
   :data:`NO_MATCHES_REGEX` / :data:`NO_MATCHES_REGEX_SCOPED` — which
   states the plain-substring contract and hands over the plain-form
   retry hint; a grep that matched, or a no-match for a plain pattern
   (or one that reduces to nothing), is byte-identical to the ordinary
   line. Still a (counted) result, never a refusal.
3. Rejected calls get a one-line refusal and count in nothing
   (``holder.tool_calls`` tracks executed calls only): unknown tool name
   → ``"Unknown tool."``; a ``read`` without a usable ``path`` (missing,
   blank or non-string) → ``"read requires a string argument
   'path'."``; a ``grep`` without a usable ``pattern`` (missing, blank
   or non-string) → ``"grep requires a string argument
   'pattern'."``; a scoped ``ls`` whose ``path``'s FIRST segment (split
   at the first ``/``) names no registered source — a bare unknown
   name (the 2026-09-03 incident's ``ls(path='.')``) or the source
   segment of a ``source/…`` argument (phase 94: a ``/`` now names a
   folder, so the phase-72 document-path teaching refusal is deleted)
   → :data:`NO_SOURCE_NOT_A_DIRECTORY`, the
   no-source refusal with the teaching parenthetical appended, the
   segment echoed; a ``source/…`` argument whose folder segment
   matches no indexed prefix (the phase-94 existence rule — some
   indexed path of the source starts with ``folder + "/"``) →
   :data:`NOT_A_FOLDER`, the drill-down teaching with the argument
   echoed and the deepest existing ancestor's direct subfolders
   listed, so the model self-corrects in the next round; a document
   READ into FULL-text context — a re-read whose lines are FULLY
   covered by earlier reads (phase 141: the COVERAGE state, not the
   pair set, decides — the raw-arg fast path is retired, every dedupe
   row resolves the document first; phase 118: the suggested seeds
   are summary blocks in the prompt, not full text, so a first read
   of a suggested document still proceeds) →
   :data:`ALREADY_IN_CONTEXT` (phase 72, task 05 gate iteration:
   the line names the correct action — answer from the text already
   in the prompt, do not call read again — so a fired refusal ends
   the loop instead of inviting a repeat; phase 141: the claim is
   then true — the named lines ARE in the prompt); a ``read`` whose
   optional ``offset``/``limit`` is not a whole line number ≥ 1
   (phase 131) → :data:`BAD_RANGE_ARGS` (the bad values echoed; fires
   FIRST — before any dedupe or resolve); a range ``read`` whose
   ``offset`` is past the document's end (phase 131) →
   :data:`OFFSET_PAST_END` (the true line count named; fires after
   the resolve, before the coverage matrix — an offset past the end
   is refused regardless of coverage); a PLAIN re-read of a document
   only PARTIALLY covered by earlier reads (a truncated read, or a
   range read — phase 141) → :data:`REREAD_PARTIAL` (the covered
   span(s) named, the grep → read(offset/limit) teaching — the full
   text is NOT in the prompt, unlike the line above); an
   unknown document (a ``read`` or scoped ``grep`` whose
   combined ``source/path`` matches nothing — a bare source name, which
   can never be a document, included) → ``"No document at '…' — check
   the ls output."`` with the argument echoed as passed (the model sees
   its own form) — EXCEPT the phase-72 "did you mean …?" teaching
   (task 02): when the argument is a path (contains ``/``) that matches
   an indexed document's ``path`` (exact or as a ``/arg`` suffix,
   case-sensitive, catalog order — :func:`find_path_candidates`, a pure
   catalog lookup, one bulk query, called only from this refusal path),
   the refusal names the combined identity instead — exactly one match
   → :data:`NO_DOCUMENT_DID_YOU_MEAN` (``did you mean
   'source/path'?``), two or more → :data:`NO_DOCUMENT_DID_YOU_MEAN_MANY`
   (up to :data:`SUGGESTION_LIMIT` identities), so the harness-prior
   misuse (the bare document path missing the source prefix,
   ``read('app/rag/importer.py')``) self-corrects in one round; it is
   still a refusal (counts in nothing, consumes a round — no silent
   argument normalization), and a bare argument (no ``/``) or a
   zero-candidate path keeps the line above byte-identical (the bare
   form never hits the DB). A grep that ran but found nothing is NOT a rejection
   — its ``"No matches for …"`` line is a (counted) result. A rejected
   call still consumes a *round* in the loop, so a pathological stream
   that keeps emitting rejected calls is bounded by the cap (point 4).
4. Every call the model emits is appended back to the message history as
   the assistant tool-call message + the tool result (refusals included),
   consumes one round, and the model is called again. At the round cap —
   ``max_rounds = settings.agent_max_rounds`` (``BOR_AGENT_MAX_ROUNDS``,
   default 10) — the loop forces one final retried no-tools request
   (``chat_stream_retried`` with ``tools=None``) and returns: the cap is
   the **only** forced exit (besides "the stream carried no calls"), and
   it bounds pathological rejected-call streams.
5. A rare stream that carries both content and a tool call keeps the
   content (it was already emitted) **and** still runs the tool.
6. *holder* (an :class:`AgentHolder`) records the read documents and the
   number of executed tool calls (re-lists included); the API layer
   (task 04) reads it after the stream to extend ``done.sources`` /
   ``query_log.sources`` and the per-turn log line (``tool_calls=N``).
7. Retries (phase 67, owner-locked A2): every model request — each tool
   round and the forced final ``tools=None`` call — goes through
   ``chat_stream_retried``: a round that dies before its first piece is
   restarted with the SAME messages (up to ``settings.llm_retries``
   restarts, a flat ``settings.llm_retry_delay`` between attempts, each
   preceded by a :class:`app.rag.llm.RetryPiece` the API layer turns into
   an SSE ``retry`` frame); a round that already streamed a piece fails
   the turn as before (no partial answer is ever redone). Retries are
   invisible to the round cap: a round that needed a retry still consumes
   exactly one round. With ``settings.llm_retries=0`` every request is a
   single plain attempt (the pre-phase-67 path).

Scaffolding guardrail (phase 71, deterministic only — owner permission
2026-09-03: "deterministic guardrails only right now, forget using a
model for that"): every model request (each round, the forced final,
and any recovery) runs its ``delta.content`` through a fresh caller-
owned :class:`app.rag.scaffolding.ScaffoldingFilter`, so raw
``<|tool_call_start|>…<|tool_call_end|>`` tokens can never reach the
user as answer text. A round that ends with NO visible content AND a
non-empty strip (the scaffolding was the whole "answer") gets exactly
ONE bounded recovery: one extra request with ``tools=None``, the same
messages with :data:`CORRECTION_INSTRUCTION` folded into the original
single system message, a fresh filter, and the same phase-67 retry
budget. A recovery that also comes back empty — or a round with no
strip and no content (today's empty/thinking-only answer) — settles as
before; a second empty reply raises :class:`MalformedReplyError` (the
API layer turns it into the dedicated error frame). A round with real
visible content plus scaffolding needs no recovery (the clean content
stands), and a scaffolding-only round that also carried tool calls
needs none either (the tool ran) — the policy keys on the no-calls
exit only. No model participates in detection or repair: the
registry + the fixed retry policy are the whole guardrail.

The DB accessors — the drill-down ``ls`` (:func:`ls_top`,
:func:`ls_folder`; the pure grouping :func:`group_folder_listing` and
the pure renderers :func:`render_ls_top` /
:func:`render_folder_listing` sit next to them),
:func:`list_source_names`, :func:`find_document`,
:func:`all_documents` — and the :func:`grep_document` line matcher are
module-level functions so unit tests can monkeypatch them without a
database.
"""
from __future__ import annotations

import json
import logging
import re
from collections.abc import AsyncIterator, Callable, Mapping, Sequence
from dataclasses import dataclass, field
from typing import Any, cast

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.config import Settings
from app.models import Document, FolderSummary
from app.rag.folder_summaries import folder_of
from app.rag.git_sources import effective_sources
from app.rag.llm import (
    LLMClient,
    LLMError,
    RetryPiece,
    StreamPiece,
    ToolCallPiece,
    ToolResultPiece,
    chat_stream_retried,
)
from app.rag.retriever import TRUNCATION_MARKER
from app.rag.scaffolding import ScaffoldingFilter
from app.rag.source_removal import resolve_source_name

logger = logging.getLogger("app.agent")

#: The three agent tools (phase 70: the harness-aligned surface —
#: ``ls`` / ``read`` / ``grep``, the pi.dev tool shapes the model was
#: trained on, replacing the phase-37 list/read and phase-68 search
#: names): OpenAI function
#: definitions passed as ``tools=AGENT_TOOLS`` to ``chat_stream`` for
#: the whole grounded turn — phase 45 removed the per-tool budgets; the
#: round cap (``BOR_AGENT_MAX_ROUNDS``) is the only bound. The combined
#: ``source/path`` string is the canonical document identity in every
#: argument (phase 70, owner permission 2026-09-03).
AGENT_TOOLS: list[dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "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."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "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."
                        ),
                    }
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read",
            "description": (
                "The <documents> section shows the SUMMARIES of the "
                "top-ranked documents — their full texts are NOT in "
                "your prompt yet. Use this tool to add one of them (or "
                "any other document) to your context, by its combined "
                "`source/path` string, exactly as shown in the `ls` "
                "output or the <documents> blocks. Do not re-read a "
                "document you have already read — its full text is "
                "already in your prompt. Very large documents are "
                "truncated: you receive the first part plus a TRUNCATED "
                "notice naming how many more characters exist. To read "
                "the rest: use `grep` (pattern) to find where your "
                "text is — it reports line numbers — then call `read` "
                "again on the same document with `offset` (1-based "
                "first line) and `limit` (number of lines) to read "
                "exactly that part. Do not re-read a document you have "
                "already read in full — its full text is already in "
                "your prompt. Call one tool at a time — wait for this "
                "result before your next call."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "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. Do not re-read a "
                            "document you have already read — its full "
                            "text is already in your prompt."
                        ),
                    },
                    "offset": {
                        "type": "integer",
                        "description": (
                            "The first line to read, 1-based. Omit it to "
                            "read from the beginning of the document. "
                            "Use it to read a specific part of a large "
                            "document — grep reports the line numbers "
                            "of what you found."
                        ),
                    },
                    "limit": {
                        "type": "integer",
                        "description": (
                            "The maximum number of lines to read, "
                            "starting at offset. Omit it to read as "
                            "many lines as fit under the character "
                            "cap. Keep it small — you are reading a "
                            "part, not the document."
                        ),
                    },
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "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` — for a "
                "large document, use the reported line number as "
                "`offset`. 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."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": (
                            "The exact text to search for (a plain "
                            "substring, not a regex — no '.*', no "
                            "'\\.', no character classes)"
                        ),
                    },
                    "path": {
                        "type": "string",
                        "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`)."
                        ),
                    },
                },
                "required": ["pattern"],
            },
        },
    },
]

#: The 4th agent tool (phase 124, task 02; LOCKED A3/A5) — offered
#: ONLY when ``settings.answer_images`` (``BOR_ANSWER_IMAGES``) is true:
#: :func:`agent_tools` composes the offer list (the three base tools
#: stay :data:`AGENT_TOOLS`, byte-identical). The description IS the
#: relevance gate (A5 — "we don't want to just show every image that
#: seems relevant"): call it only when the image itself contains the
#: answer, at most ONE image per answer, the text answer is always
#: given in full (the image supplements, never replaces). Same OpenAI
#: function shape as the other three (one required string arg ``path``,
#: the combined ``source/path`` identity — the bare-path-will-not-
#: resolve teaching mirrored from ``read``).
SHOW_IMAGE_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "show_image",
        "description": (
            "Show a standalone image document to the user as part of "
            "your answer. Call it only when the image itself contains "
            "the answer — a diagram, screenshot, or photo of the thing "
            "the user asked about — never for a merely related image. "
            "Show at most one image per answer. The image supplements "
            "your text answer — answer fully in text either way. Its "
            "`path` is the combined `source/path` string, exactly as "
            "shown in the `ls` output or the `<documents>` blocks — a "
            "bare document path will not resolve. Call one tool at a "
            "time — wait for this result before your next call."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": (
                        "The image document to show, as the combined "
                        "`source/path` string exactly as shown in the "
                        "`ls` output (e.g. "
                        "'homelab/docs/architecture.png'). A bare "
                        "document path (without the source name) will "
                        "not resolve. Show at most one image per "
                        "answer — a second call is refused."
                    ),
                }
            },
            "required": ["path"],
        },
    },
}


def agent_tools(settings: Settings) -> list[dict[str, Any]]:
    """The grounded turn's tool offer list (phase 124, task 02).

    The three base tools (:data:`AGENT_TOOLS`, byte-identical) plus
    :data:`SHOW_IMAGE_TOOL` when ``settings.answer_images``
    (``BOR_ANSWER_IMAGES``) is true — otherwise exactly the three base
    tools (the pre-phase-124 offer, byte-identical: no 4th tool in the
    wire ``tools`` array). A FRESH list on every call — callers never
    mutate the result or the constants (``run_agent`` passes it straight
    to the LLM request).
    """
    tools: list[dict[str, Any]] = list(AGENT_TOOLS)
    if settings.answer_images:
        tools.append(SHOW_IMAGE_TOOL)
    return tools

#: Tool refusal texts (phase 37): rejected calls count in nothing
#: (``holder.tool_calls`` tracks executed calls); the round cap bounds
#: their pathological repetition (phase 45). The in-context line is a
#: phase-72, task 05 gate-iteration teaching (live telemetry: the
#: ``lite`` model obeyed the user's "open it / read it" and re-read
#: seed-context documents, then repeated the call against the terse
#: phase-37 line — the refusal itself carried no correct action): same
#: behavior (a refusal: counts in nothing, consumes a round, changes
#: no context), the copy now names the action, so even a fired
#: refusal ends the loop instead of inviting a repeat.
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."
)
UNKNOWN_TOOL = "Unknown tool."
MISSING_READ_ARGS = "read requires a string argument 'path'."
MISSING_SEARCH_ARGS = "grep requires a string argument 'pattern'."
MISSING_SHOW_ARGS = "show_image requires a string argument 'path'."

#: The ``show_image`` refusals (phase 124, task 02): the same contract
#: as every other refusal — counts in nothing, consumes a round,
#: changes no context. ``NOT_AN_IMAGE_DOC``: one ``{arg}`` field, the
#: argument echoed as passed — a text doc is refused, never silently
#: converted to a read (LOCKED A5). ``ALREADY_SHOWING``: the LOCKED A3
#: one-image-per-answer gate, enforced SERVER-side (the model is taught
#: it in the tool description + the prompt clause; the handler is the
#: actual gate).
NOT_AN_IMAGE_DOC = (
    "'{arg}' is not an image document — only standalone image "
    "documents can be shown as answers."
)
ALREADY_SHOWING = (
    "Already showing an image this answer — do not show another one. "
    "Answer from what you have."
)

#: The read-truncation notice (phase 95, task 01; REWRITTEN phase 131,
#: task 01) — appended, after the shared :data:`TRUNCATION_MARKER`, to
#: a ``read`` result whose document is LONGER than
#: ``settings.read_max_chars`` (``BOR_READ_MAX_CHARS``, default
#: 128 000 chars ≈ 32k tokens). Two format fields: ``{total}`` (the
#: document's true character count) and ``{shown}`` (the cap — how many
#: characters actually reached the model). The copy is pinned: it tells
#: the model the read was truncated and that the rest is NOT in its
#: context (the document did not end where it stopped), then teaches
#: the phase-131 workflow — the old dead-end ("use grep to locate the
#: rest", with no way to act on the line number) is gone: ``grep``
#: reports LINE NUMBERS, and the model reads the part it needs with
#: ``read``'s ``offset`` (1-based first line) + ``limit`` (number of
#: lines). A read at or under the cap appends nothing (byte-identical
#: to the pre-phase-95 result).
READ_TRUNCATION_NOTICE = (
    "TRUNCATED — this document is {total} characters; only the first "
    "{shown} are in your context. The rest is NOT shown. Use grep "
    "(pattern) to find what you need — it reports line numbers — "
    "then read that part with `offset` (1-based first line) and "
    "`limit` (number of lines)."
)

#: The ``read`` range-argument validation refusal (phase 131, task
#: 02): ``offset`` / ``limit``, when present, must each be a whole
#: line number ≥ 1 — an ``int`` (bools excluded — the house
#: ``isinstance(x, int) and not isinstance(x, bool)`` guard) and not
#: ``< 1``. The bad values are echoed (``{offset!r}`` /
#: ``{limit!r}`` — an absent value prints as ``None``), so the model
#: sees exactly what it sent. Absent/``None`` for BOTH is fine (a
#: plain read); ``offset`` without ``limit`` reads to the END of the
#: document; ``limit`` without ``offset`` is treated as ``offset=1``
#: (a line count from the start) — the two one-sided semantics
#: :func:`_read_range_args` validates and the read branch applies.
#: Still a refusal: counts in nothing, consumes a round (the phase-72
#: style — no silent argument repair), and it fires BEFORE the dedupe
#: and the DB resolve (a bad range argument is refused regardless of
#: what is in context).
BAD_RANGE_ARGS = (
    "`offset` and `limit` are whole line numbers — pass integers ≥ 1 "
    "(you passed offset={offset!r}, limit={limit!r})."
)

#: The plain re-read of a PARTIALLY-covered document (phase 141,
#: task 02 — the dedupe matrix's middle row, generalizing and
#: RETIRING the phase-131 ``REREAD_TRUNCATED`` line, whose char-based
#: ``{shown}`` copy is replaced by the line-based truth): earlier
#: reads put SOME — not all — of the document's lines in the
#: model's context (a truncated read covers its cut span, a range
#: read covers its own span — :data:`AgentHolder.read_coverage` is
#: the state this row decides on), so the rest is NOT in its
#: context. The line names the covered span(s) in line form
#: (``{covered}`` — the merged spans rendered by
#: :func:`format_covered_spans`, comma-joined ``a–b`` pairs, a single
#: line bare ``a``) and points at the working workflow — ``grep``
#: reports line numbers, then ``read`` that part with ``offset`` /
#: ``limit`` — instead of the dead end (the plain
#: :data:`ALREADY_IN_CONTEXT` line, which claimed the FULL text was
#: in the prompt — false for a partial document). ``{source}`` /
#: ``{path}`` are the RESOLVED identity; ``{total}`` is the
#: document's true line count (``len(content.splitlines())``) — both
#: plain, no ``!r``. A refusal: no re-emit, ``tool_calls`` NOT
#: incremented, consumes a round; fires after the resolve (the true
#: line count only the resolve knows — the raw-arg fast path is
#: retired).
REREAD_PARTIAL = (
    "You have lines {covered} of {source}/{path} in your context "
    "({total} lines total). To read the rest, find where your text "
    "is with grep (it reports line numbers), then call read with "
    "`offset` (1-based first line) and `limit` (number of lines)."
)

#: The range-``read`` offset past the end (phase 131, task 02): the
#: document is ``{total_lines}`` lines long (``len(content.
#: splitlines())``) and the ``offset`` names a line beyond it.
#: ``{source}`` / ``{path}`` are the RESOLVED identity; ``{total_lines}``
#: and ``{offset}`` are plain integers (no ``!r``). The pointer — grep
#: reports the exact line numbers — hands the model back to the
#: locator that got it here. A refusal: counts in nothing, consumes a
#: round; the document is NOT added to the context, so a corrected
#: retry (a smaller offset) still proceeds.
OFFSET_PAST_END = (
    "{source}/{path} is {total_lines} lines long — `offset` {offset} "
    "is past the end. Use a smaller offset (grep reports the exact "
    "line numbers)."
)

#: The image-document marker (phase 122, task 05): the line prefixed to
#: the vision DESCRIPTION a ``read`` of a standalone-image document
#: returns — the model must reason about what it is reading (the text
#: below is a description GENERATED from the image, not the image's
#: own words). It sits on the result's THIRD line: the ``Document …``
#: header and the phase-106 date line stay byte-identical (the E2E
#: mock's ``_READ_RESULT_PREFIX`` header contract), and a NON-image
#: doc's result carries no marker at all (byte-identical to pre-122).
IMAGE_DOC_MARKER = (
    "Image document — the text below is a description generated from the image:"
)

#: The no-source ``ls`` refusal with the teaching parenthetical
#: appended (phase 72): used when a stripped scope has no ``/`` and
#: matches no registered source (the incident's ``ls(path='.')``). The
#: prefix — the pre-phase-72 line — stays byte-identical; one ``{scope}``
#: field, the argument echoed.
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.)"
)

#: Teaching refusal for a ``read`` / scoped ``grep`` argument that
#: resolves to no combined identity but matches ONE indexed document's
#: ``path`` (phase 72, task 02): names the exact combined
#: ``source/path`` identity to use, so the harness-prior misuse — the
#: bare document path missing the source prefix
#: (``read('app/rag/importer.py')``) — self-corrects in one round.
#: One each of the fields ``{arg}`` (the argument echoed as passed),
#: ``{source}`` and ``{path}`` (the one candidate). Still a refusal:
#: it counts in nothing and consumes a round (no silent argument
#: normalization).
NO_DOCUMENT_DID_YOU_MEAN = (
    "No document at '{arg}' — did you mean '{source}/{path}'?"
)

#: The ambiguous form of the same teaching (phase 72, task 02): the
#: argument matches SEVERAL indexed documents' ``path`` (the same path
#: under several sources). ``{candidates}`` holds up to
#: :data:`SUGGESTION_LIMIT` combined ``source/path`` identities, each
#: single-quoted, joined with ``", "`` in catalog order; ``{arg}`` is
#: the argument echoed as passed.
NO_DOCUMENT_DID_YOU_MEAN_MANY = (
    "No document at '{arg}' — did you mean one of: {candidates}?"
)

#: Cap on the suggested combined identities per "did you mean …?"
#: refusal (phase 72, task 02): the same document ``path`` under
#: several sources suggests up to this many (catalog order, the rest
#: dropped).
SUGGESTION_LIMIT = 3

#: The drill-down ``ls`` file-line cap (phase 94, task 03): a folder's
#: own files list at most this many
#: ``source: X | path: Y | title: Z | date: YYYY-MM-DD``
#: lines (path order), then one deterministic grep-pointer note — a
#: 500-file folder costs the model 50 lines + the note, never 500.
#: Pinned module constant (no env var — the phase-94 TODO asks for a
#: shape change, not a knob; the constant lives next to
#: :data:`SEARCH_MAX_MATCHES`).
LS_MAX_FILE_LINES = 50

#: The NOT-A-FOLDER ``ls`` teaching refusal (phase 94, task 03):
#: a ``source/…`` argument whose folder segment matches no indexed
#: prefix (the ``00_phase.md`` existence rule — a folder exists iff
#: some indexed path starts with ``folder + "/"``; a document's own
#: path is never a folder). Phase-72 teaching style: one line, the
#: argument echoed (``{arg}``), the DEEPEST existing ancestor's name
#: (``{parent}`` — the source for a top-level miss, ``source/folder``
#: for a nested one) and its direct subfolders (``{subfolders}`` —
#: space-joined ``name/`` entries in path order, so the model
#: self-corrects in the next round; ``none`` when the ancestor has no
#: subfolders). Still a refusal: it counts in nothing and consumes a
#: round (no silent argument normalization).
NOT_A_FOLDER = "'{arg}' is not a folder — {parent} has: {subfolders}"

#: The harness-owned recovery line (phase 71, task 03) — folded into the
#: ORIGINAL single system message of the one bounded recovery request
#: (``system_prompt + "\n" + CORRECTION_INSTRUCTION``; provider-safe,
#: the user message stays last). Verbatim constant: the E2E mock
#: (task 05) keys on a stable substring of it, so it must not drift.
CORRECTION_INSTRUCTION: str = (
    "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."
)


class MalformedReplyError(LLMError):
    """The model kept replying in raw tool-scaffolding (phase 71).

    Raised ONLY by the recovery policy — :func:`run_agent` (grounded
    path) and ``app.api.chat`` (deflected path) — when the one bounded
    ``tools=None`` recovery still comes back with no visible content.
    It is never raised from inside a stream, so
    :func:`app.rag.llm.chat_stream_retried`'s retry-before-first-piece
    rule never sees it. The API layer catches it BEFORE the generic
    :class:`LLMError` handler and settles the turn with the dedicated
    "malformed reply" error frame (no ``done``, no ``query_log`` row).
    Deterministic only (owner permission 2026-09-03): no model
    participates in detection or repair.
    """

#: Search caps (owner-locked A5, phase 68): a global per-call match cap
#: (across documents, in catalog order) and a per-match-line char limit.
SEARCH_MAX_MATCHES = 20
SEARCH_LINE_LIMIT = 200

#: No-match result lines (templates — the pattern is truncated to 100
#: chars before formatting, to keep a long pattern from bloating the
#: tool result). A no-match line is a *result* of an executed grep,
#: not a refusal (see the module docstring, point 3).
NO_MATCHES = "No matches for '{pattern}' in the knowledge base."
NO_MATCHES_SCOPED = "No matches for '{pattern}' in {source}/{path}."

#: Teaching no-match lines (the 2026-09-05 incident — owner chat:
#: "Qwen 3.8" question failed repeatedly). The harness-trained prior is
#: that ``grep(pattern)`` takes a REGEX (pi.dev's grep, ripgrep, grep
#: itself); this app's grep is a case-insensitive FIXED SUBSTRING
#: (owner-locked A5 — the contract does not change). A regex-shaped
#: pattern (``qwen.*3\.8``, ``qwen 3\.8``) can therefore never match, and
#: the bare no-match line above made the turbo model trust the miss and
#: end the turn with a wrong "I searched the entire knowledge base"
#: refusal. These lines are the deterministic teaching: the same result
#: (a no-match is still a *result* — counted, no context added) with the
#: correct contract stated and a plain-text retry hint (the
#: :func:`plain_form` of the pattern, when non-empty). Deterministic only
#: (owner permission 2026-09-03: "deterministic guardrails only right
#: now"): no model participates in detection or repair.
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}')."
)
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}')."
)

#: One of these metacharacters anywhere in the pattern marks it
#: regex-shaped (the detection for the teaching no-match lines above).
#: Plain substrings that happen to contain one ("C++", "a|b") are
#: affected only on a NO-MATCH — a pattern that matched literally still
#: gets its ordinary result, so the teaching can never suppress a real
#: hit.
_REGEX_META_RE = re.compile(r"[*+?(){}\[\]|\\^$]")


# Backslash + a non-alphanumeric char is an escaped literal (``\\.`` →
# ``.``); backslash + an alphanumeric is a class shorthand (``\\d``,
# ``\\w``, ``\\s``) with no plain-text equivalent (→ dropped).
_ESCAPE_RE = re.compile(r"\\(.)")


def looks_like_regex(pattern: str) -> bool:
    """True when *pattern* carries regex metacharacters (see
    :data:`_REGEX_META_RE`). Pure detection — the grep itself stays a
    fixed substring (owner-locked A5)."""
    return _REGEX_META_RE.search(pattern) is not None


def plain_form(pattern: str) -> str:
    """A deterministic plain-text retry hint for a regex-shaped pattern.

    The hint is the pattern reduced to literal text, in this order:
    drop ``.*`` runs on the RAW pattern first (the wildcard — an escaped
    ``\\.`` + ``\\*`` pair carries no raw ``.*`` run, so a literal dot
    survives), unescape (``\\.`` → ``.``; class shorthands like ``\\d``
    dropped), keep only the FIRST ``|`` alternative, drop character
    classes (``[0-9]``), drop quantifier runs (``*``, ``+``, ``?``,
    ``{2,3}``), drop group parens and anchors (contents kept). Whitespace
    is preserved. ``qwen.*3\\.8`` → ``qwen3.8`` (the incident's exact
    recovery), ``llama\\.cpp`` → ``llama.cpp``, ``qwen[0-9]+`` →
    ``qwen``. A pattern made of pure metacharacters reduces to ``""`` —
    callers then fall back to the ordinary no-match line (no hint).
    """
    p = re.sub(r"\.\*", "", pattern)  # .* wildcard runs (raw form)
    p = _ESCAPE_RE.sub(lambda m: m.group(1) if not m.group(1).isalnum() else "", p)
    p = p.split("|", 1)[0]  # first alternative only — a hint, not an answer
    p = re.sub(r"\[[^\]]*\]", "", p)  # character classes carry no literal text
    p = re.sub(r"\{[^{}]*\}", "", p)  # {n} / {n,} / {n,m} quantifiers
    p = re.sub(r"[*+?]+", "", p)  # stray * + ? quantifiers
    p = p.replace("(", "").replace(")", "")
    p = p.replace("^", "").replace("$", "")
    return p.strip()


def list_source_names(db: Session) -> list[str]:
    """Every registered source name, deduped, in registry order.

    The source registry (the ``git_sources`` rows — the
    ``BOR_GIT_SOURCES`` env fallback while the table is empty) is the
    source of truth for *source* names independent of document count:
    a registered source with no indexed documents still lists (as
    ``0 documents:`` — the scoped ``ls`` must not refuse it as unknown).
    Names resolve exactly as the import pipeline indexes them
    (:func:`app.rag.source_removal.resolve_source_name` — reuse, or a
    scoped ``ls`` would judge the wrong names unknown, the phase-69
    "RAG consistent with the registry" invariant); two rows resolving
    to the same name (the phase-69 sibling case) share documents, so
    the name is listed once. Module-level (not a method) so unit tests
    can monkeypatch it.
    """
    rows, _origin = effective_sources(db)
    names: list[str] = []
    for row in rows:
        name = resolve_source_name(row)
        if name not in names:
            names.append(name)
    return names


#: The drill-down ``ls`` fetchers (phase 94, task 03) — module-level so
#: unit tests can monkeypatch them without a database (the house style:
#: :func:`ls_top` / :func:`ls_folder` compose them; the pure grouping
#: :func:`group_folder_listing` and renderers
#: :func:`render_ls_top` / :func:`render_folder_listing` sit below).


def _source_document_counts(db: Session) -> list[tuple[str, int]]:
    """``(source, document count)`` per indexed source (one grouped
    query — the top-level ``ls`` counts, phase 94 task 03)."""
    return [
        (source, count)
        for source, count in db.execute(
            select(Document.source, func.count()).group_by(Document.source)
        )
    ]


def _source_root_summaries(db: Session) -> list[tuple[str, str]]:
    """The stored source-root summaries ``(source, summary)``
    (``folder_path = ""`` — the top-level ``ls`` indented lines, phase
    94 task 03; absent when never generated)."""
    return [
        (source, summary)
        for source, summary in db.execute(
            select(FolderSummary.source, FolderSummary.summary).where(
                FolderSummary.folder_path == ""
            )
        )
    ]


def _source_document_rows(db: Session, source: str) -> list[tuple[str, str, str]]:
    """``(path, title, created_iso_date)`` of every document under
    *source*, ordered by ``path`` — the one bounded fetch a folder
    drill level lists (phase 94 task 03; one source's paths, not the
    whole KB). The date is the row's ``created_at`` UTC date part
    (``YYYY-MM-DD``, phase 106 D5 — the ``ls`` FILE line's appended
    `` | date: …`` field; only file lines carry a date)."""
    return [
        (path, title, created_at.strftime("%Y-%m-%d"))
        for path, title, created_at in db.execute(
            select(Document.path, Document.title, Document.created_at)
            .where(Document.source == source)
            .order_by(Document.path)
        )
    ]


def _source_folder_summaries(db: Session, source: str) -> dict[str, str]:
    """The stored folder summaries ``{folder_path: summary}`` of one
    source (phase 94 task 03; includes the ``""`` source-root row when
    stored)."""
    return {
        folder_path: summary
        for folder_path, summary in db.execute(
            select(FolderSummary.folder_path, FolderSummary.summary).where(
                FolderSummary.source == source
            )
        )
    }


def ls_top(db: Session) -> list[tuple[str, int, str | None]]:
    """The top level of the drill-down ``ls`` (phase 94, task 03).

    Every registered source in :func:`list_source_names` order (the
    registry is the source of truth — a registered source with 0
    indexed documents still lists, the phase-70/72 invariant) as
    ``(name, recursive_document_count, source_root_summary)``: the
    count is the source's whole subtree (all of its documents — the
    same set its stored summary describes) and the summary is the
    stored ``folder_summaries`` row for ``(source, "")`` (the source
    root, phase 94 task 01) or ``None`` when absent. Module-level so
    unit tests can monkeypatch the fetchers.
    """
    names = list_source_names(db)
    if not names:
        return []
    counts = dict(_source_document_counts(db))
    summaries = dict(_source_root_summaries(db))
    return [(name, counts.get(name, 0), summaries.get(name)) for name in names]


def group_folder_listing(
    source: str,
    folder: str,
    rows: Sequence[tuple[str, str, str]],
    summaries: Mapping[str, str],
) -> tuple[
    list[tuple[str, int, str | None]], list[tuple[str, str, str, str]], int
]:
    """One level of the drill-down tree (phase 94, task 03) — pure.

    Given *rows* — the source's ``(path, title, created_iso_date)``
    triples in catalog (path) order — and *summaries* (the source's
    stored ``folder_summaries`` rows: ``folder_path → summary``), the
    folder level *folder* (source-relative; ``""`` = the source root):

    * **(a) direct subfolders** — the folders whose parent is exactly
      *folder*, in path order, each
      ``(sub_path_relative_to_source, recursive_count,
      summary_or_None)``. A folder is a slash-boundary prefix of at
      least one indexed path (the ``00_phase.md`` existence rule: a
      folder ``F`` exists ⟺ some path starts with ``F + "/"`` — a
      document's OWN path is never a folder); the count is the
      folder's recursive subtree — every path equal to the folder or
      starting with ``folder + "/"`` (the same set the sync-time
      folder summary describes, phase 94 task 01 — one concept end to
      end).
    * **(b) direct file lines** — the documents whose folder (the
      prefix before the last ``/`` —
      :func:`app.rag.folder_summaries.folder_of`, the shared notion) IS
      *folder*, in path order (catalog order — the same order
      ``GET /api/docs`` serves), as ``(source, path, title, date)``
      4-tuples — the canonical ``read``/``grep`` identity plus the
      phase-106 D5 ``date`` field (the row's ``created_at`` UTC date
      part, APPENDED — never inserted before ``title``), capped at
      :data:`LS_MAX_FILE_LINES` (the rest fold into the renderer's
      note; a 500-file folder never costs 500 lines).
    * **(c) the TOTAL direct-file count** — pre-cap, for the note.

    Pure (no I/O) — unit tests drive the grouping without a database;
    :func:`ls_folder` is the DB-composing wrapper.
    """
    # The source's existing folders: every slash-boundary prefix of an
    # indexed path (the existence rule's candidate set — a folder is
    # present iff at least one path starts with ``folder + "/"``).
    folders: set[str] = set()
    for path, _title, _date in rows:
        f = folder_of(path)
        while f:
            folders.add(f)
            f = folder_of(f)
    # The recursive count per folder — the ``path == folder`` arm (a
    # document sharing a folder's name) plus the ``startswith
    # folder + "/"`` arm (the folder's true descendants), one pass per
    # document.
    counts: dict[str, int] = {f: 0 for f in folders}
    for path, _title, _date in rows:
        if path in folders:
            counts[path] += 1
        f = folder_of(path)
        while f:
            counts[f] += 1
            f = folder_of(f)
    subfolders = [
        (g, counts[g], summaries.get(g))
        for g in sorted(g for g in folders if folder_of(g) == folder)
    ]
    files = [
        (source, path, title, date)
        for path, title, date in rows
        if folder_of(path) == folder
    ]
    return subfolders, files[:LS_MAX_FILE_LINES], len(files)


def ls_folder(
    db: Session, source: str, folder: str
) -> tuple[
    list[tuple[str, int, str | None]], list[tuple[str, str, str, str]], int
]:
    """One folder level of the drill-down ``ls`` (phase 94, task 03).

    The source's document rows (:func:`_source_document_rows`) and
    stored folder summaries (:func:`_source_folder_summaries`) through
    :func:`group_folder_listing` — the pure grouping the unit tests
    drive directly. ``folder = ""`` is the source root. Module-level
    so unit tests can monkeypatch the fetchers.
    """
    return group_folder_listing(
        source,
        folder,
        _source_document_rows(db, source),
        _source_folder_summaries(db, source),
    )


def _folder_exists_in(rows: Sequence[tuple[str, str, str]], folder: str) -> bool:
    """The phase-94 folder-existence rule (``00_phase.md``), pure.

    Folder *folder* (source-relative) under a registered source
    exists ⟺ ``folder == ""`` OR some indexed path starts with
    ``folder + "/"`` — a document's OWN path is never a folder
    (nothing starts with ``path + "/"``), so ``ls`` of a file path
    refuses with :data:`NOT_A_FOLDER` rather than listing.
    """
    if not folder:
        return True
    prefix = folder + "/"
    return any(path.startswith(prefix) for path, _title, _date in rows)


def _deepest_existing_ancestor(
    rows: Sequence[tuple[str, str, str]], folder: str
) -> str:
    """The deepest EXISTING folder prefix of a missing *folder* (pure).

    The :data:`NOT_A_FOLDER` refusal's teaching context: the argument's
    segments are walked from the top; the walk stops at the first
    segment that is no folder, so the returned prefix is the deepest
    existing ancestor (``""`` = the source root when the first segment
    is already the miss) and its direct subfolders are the bounded
    self-correction list the refusal prints.
    """
    parent = ""
    for part in folder.split("/"):
        candidate = f"{parent}/{part}" if parent else part
        if not _folder_exists_in(rows, candidate):
            break
        parent = candidate
    return parent


def render_ls_top(entries: Sequence[tuple[str, int, str | None]]) -> str:
    """The top-level ``ls`` result (phase 94, task 03) — the pinned
    template.

    ``{N} sources:`` — the line alone when the registry is empty (the
    old ``0 documents:`` behavior preserved in spirit) — then, when at
    least one source is registered, a blank line and one block per
    source in registry order: ``{source} — {n} documents`` plus the
    indented ``  {summary}`` line ONLY when the source-root summary is
    stored (absent → the count line alone, no placeholder) — with NO
    blank line between blocks.
    """
    lines = [f"{len(entries)} sources:"]
    if entries:
        lines.append("")
        for source, count, summary in entries:
            lines.append(f"{source} — {count} documents")
            if summary:
                lines.append(f"  {summary}")
    return "\n".join(lines)


def render_folder_listing(
    identity: str,
    subfolders: Sequence[tuple[str, int, str | None]],
    files: Sequence[tuple[str, str, str, str]],
    total_files: int,
) -> str:
    """One folder level of the drill-down ``ls`` (phase 94, task 03) —
    the pinned template.

    The header ``{identity} — {n_files} documents, {n_folders}
    folders:`` (``n_files`` = *total_files*, the PRE-cap direct-file
    count; ``identity`` is the source name at the root level and
    ``source/folder`` below it), then — when the level carries anything
    below the header — a blank line, the 2-space-indented subfolder
    lines ``  {sub}/ — {m} documents`` in path order (``: {summary}``
    appended ONLY when the subfolder's summary is stored), a blank
    line, the file lines in EXACTLY the
    ``source: X | path: Y | title: Z | date: YYYY-MM-DD`` format (the
    canonical ``read``/``grep`` identity plus the phase-106 D5
    appended ``date`` field — the only changed part), and the cap note
    ``…and {hidden} more documents in this folder — use grep
    (pattern) to find a specific one.`` ONLY when the folder's own
    files outnumber :data:`LS_MAX_FILE_LINES` (*files* arrives capped;
    *total_files* carries the pre-cap count). A header with no
    subfolders and no files — a registered source with no documents —
    is the header line alone (``… — 0 documents, 0 folders:``).
    """
    header = f"{identity} — {total_files} documents, {len(subfolders)} folders:"
    if not subfolders and not files:
        return header
    body: list[str] = []
    if subfolders:
        body.append("")
        for sub, count, summary in subfolders:
            line = f"  {sub}/ — {count} documents"
            if summary:
                line += f": {summary}"
            body.append(line)
    if files or total_files > len(files):
        body.append("")
        body.extend(
            f"source: {source} | path: {path} | title: {title} | date: {date}"
            for source, path, title, date in files
        )
        hidden = total_files - len(files)
        if hidden > 0:
            body.append(
                f"…and {hidden} more documents in this folder — use grep "
                "(pattern) to find a specific one."
            )
    return "\n".join([header, *body])


def suggested_folder_lines(
    db: Session,
    suggested: Sequence[Document],
    max_lines: int = 5,
    max_entries: int = 8,
) -> list[str]:
    """The HIGH prompt's suggested-folder context lines (phase 119, D3,
    LOCKED A4) — pure composition over the existing ``ls`` machinery.

    One line per DISTINCT parent folder of the *suggested* documents —
    in suggested-doc order, deduped by ``(source, parent prefix)`` (the
    first suggested doc wins the slot), at most *max_lines* lines:

    * the parent prefix is the path up to (excluding) the last ``/``
      (``""`` = the source root);
    * the line is ``<source>/<prefix>/: e1, e2, …`` (the source root
      renders as ``<source>/: …`` — the filesystem-style folder path,
      trailing slash included, + the colon) with the folder's direct
      children in the
      EXISTING ``ls`` folder-level rendering order — the direct
      subfolders first (``name/ (N docs)``, the recursive doc count,
      singular ``(1 doc)``), then the files by relative filename — so
      the line reads the same as the model's own ``ls`` output of that
      folder (the :func:`group_folder_listing` grouping, over
      :func:`_source_document_rows` + :func:`_source_folder_summaries`);
    * the suggested document that OWNS the line is excluded from the
      entries (its identity is already in its ``<document>`` block — the
      line is the folder's OTHER contents, the pre-seed that makes the
      model ``read`` the right file in round 1 instead of walking the
      ``ls`` drill-downs);
    * at most *max_entries* entries, then `` +N more`` (N = the
      remaining count, the true pre-cap folder total — the suggested
      doc leaves the total even when its row sat past
      :data:`LS_MAX_FILE_LINES`); a folder whose only entry was the
      suggested doc renders its header alone (the ``… — 0 documents, 0
      folders:`` empty-level precedent).

    Empty *suggested* → ``[]`` (the caller then builds the
    byte-identical phase-118 prompt). Module-level so unit tests can
    monkeypatch the fetchers without a database (the house style:
    :func:`ls_top` / :func:`ls_folder` compose the same fetchers).
    """
    lines: list[str] = []
    seen: set[tuple[str, str]] = set()
    rows_cache: dict[str, tuple[list[tuple[str, str, str]], dict[str, str]]] = {}
    for doc in suggested:
        if len(lines) >= max_lines:
            break
        prefix = folder_of(doc.path)
        key = (doc.source, prefix)
        if key in seen:
            continue
        seen.add(key)
        if doc.source not in rows_cache:
            rows_cache[doc.source] = (
                _source_document_rows(db, doc.source),
                _source_folder_summaries(db, doc.source),
            )
        rows, summaries = rows_cache[doc.source]
        subfolders, files, total_files = group_folder_listing(
            doc.source, prefix, rows, summaries
        )
        entries = [
            f"{sub}/ ({count} {'doc' if count == 1 else 'docs'})"
            for sub, count, _summary in subfolders
        ]
        # The owning suggested doc is a direct file of this folder —
        # drop it from the entries (its identity is already in its
        # <document> block); its count leaves the total either way
        # (even when its row sat past the LS_MAX_FILE_LINES file cap).
        files = [entry for entry in files if entry[1] != doc.path]
        # The file entries come AFTER the subfolders (the ls folder-level
        # order) and ride by RELATIVE filename — the basename within the
        # folder (the line's ``<source>/<prefix>/:`` header supplies the
        # folder; combined ``source/prefix/name`` is the read identity).
        entries.extend(path.rsplit("/", 1)[-1] for _src, path, _title, _date in files)
        total = len(subfolders) + total_files - 1
        shown = entries[:max_entries]
        # The pinned identity shape: ``<source>/<prefix>/:`` (the source
        # root: ``<source>/:``) — the filesystem-style folder path
        # (trailing slash included) + the colon.
        identity = f"{doc.source}/{prefix}/" if prefix else f"{doc.source}/"
        line = f"{identity}:"
        if shown:
            suffix = ", ".join(shown)
            if total > len(shown):
                suffix += f" +{total - len(shown)} more"
            line += f" {suffix}"
        lines.append(line)
    return lines


def find_document(db: Session, source: str, path: str) -> Document | None:
    """The indexed document at ``(source, path)``, or ``None``.

    Module-level (not a method) so unit tests can monkeypatch it.
    """
    return db.scalar(
        select(Document).where(Document.source == source, Document.path == path)
    )


def _resolve_path(db: Session, combined: str) -> tuple[Document | None, str, str]:
    """The combined ``source/path`` identity → document (phase 70).

    The canonical document identity in every tool argument, refusal and
    result header is the combined string exactly as printed in the
    ``ls`` output, the ``Document …`` result headers, and the grep
    result lines. Source names are directory basenames (``app.rag.importer``:
    ``source = root.name``) and can never contain a ``'/'``, so the
    split at the FIRST slash is exact: the part before is the source
    name, the part after is the path. Returns ``(doc, source, path)``
    with the split pair (so callers can echo the canonical form, e.g.
    the scoped no-match line); no ``'/'`` in the argument →
    ``(None, combined, "")`` — a bare source name is never a document
    (no DB lookup; the refusal echoes the argument as passed).
    """
    if "/" not in combined:
        return None, combined, ""
    source, _, path = combined.partition("/")
    return find_document(db, source, path), source, path


def all_documents(db: Session) -> list[Document]:
    """Every indexed document (full rows), ordered by ``(source, path)``
    — catalog order.

    The whole-KB ``grep`` path loads all contents in this one bulk query
    (catalog order is the locked match order, owner-locked A5).
    Module-level (not a method) so unit tests can monkeypatch it.
    """
    return list(
        db.execute(
            select(Document).order_by(Document.source, Document.path)
        ).scalars()
    )


def find_path_candidates(db: Session, arg: str) -> list[tuple[str, str, str]]:
    """The indexed documents a bare document *arg* names by ``path``.

    Phase 72, task 02: a ``read`` / scoped-``grep`` argument that
    resolves to no combined identity but *is* a document path (contains
    ``/``) is matched against the indexed ``Document.path`` values so
    the refusal can name the combined ``source/path`` identity to use
    (the "did you mean …?" teaching). The documents whose ``path``
    equals *arg* (the exact bare path) or ends with ``f"/{arg}"`` (the
    file is nested deeper — the suffix match) — in catalog order (the
    :func:`all_documents` order), case-sensitive (these are file
    paths) — as ``(source, path, title)`` triples. One bulk query via
    :func:`all_documents` (at most one); called ONLY from the refusal
    path of :func:`_execute_tool` (never on the happy path) and only
    when *arg* contains ``/`` (a bare name keeps today's no-DB-lookup
    refusal). Module-level (not a method) so unit tests can
    monkeypatch it.
    """
    return [
        (doc.source, doc.path, doc.title)
        for doc in all_documents(db)
        if doc.path == arg or doc.path.endswith(f"/{arg}")
    ]


def _no_document_refusal(db: Session, arg: str) -> str:
    """The no-document refusal for an unresolved ``read`` / scoped-
    ``grep`` argument (phase 72, task 02).

    The pre-phase-72 line — the argument echoed as passed — whenever
    there is nothing to suggest: a bare argument (no ``/`` — a bare
    source name or any other bare name gets the no-DB-lookup refusal,
    byte-identical to today) or a path-like argument that matches no
    indexed document's ``path`` (zero candidates). A path-like argument
    (contains ``/``) that matches exactly one indexed document's
    ``path`` gets :data:`NO_DOCUMENT_DID_YOU_MEAN` (the combined
    identity named); two or more get :data:`NO_DOCUMENT_DID_YOU_MEAN_MANY`
    (up to :data:`SUGGESTION_LIMIT`, catalog order). Deterministic
    only: the suggestion is a pure catalog lookup, no model. A refusal
    still counts in nothing and consumes a round.
    """
    if "/" in arg:
        candidates = find_path_candidates(db, arg)
        if len(candidates) == 1:
            source, path, _title = candidates[0]
            return NO_DOCUMENT_DID_YOU_MEAN.format(arg=arg, source=source, path=path)
        if len(candidates) > 1:
            identities = ", ".join(
                f"'{source}/{path}'"
                for source, path, _title in candidates[:SUGGESTION_LIMIT]
            )
            return NO_DOCUMENT_DID_YOU_MEAN_MANY.format(arg=arg, candidates=identities)
    return f"No document at '{arg}' — check the ls output."


def _read_range_args(
    offset: Any, limit: Any
) -> tuple[bool, str, int | None, int | None]:
    """Validate ``read``'s optional line range (phase 131, task 02).

    Returns ``(ok, error_line, offset, limit)``:

    * both absent/``None`` → ``(True, "", None, None)`` — a plain read
      (no range): from the beginning, capped as pre-phase-131.
    * either PRESENT but not a whole int ≥ 1 (not an ``int`` — bools
      excluded — or ``< 1``) → ``(False, BAD_RANGE_ARGS.format(...),
      None, None)`` — the values echoed with ``!r`` (an absent one
      prints ``None``). A refusal: counts in nothing, consumes a
      round; it fires before the dedupe and the DB resolve.
    * ok → ``(True, "", offset, limit)`` — either may be ``None``:
      ``offset`` without ``limit`` reads to the END of the document,
      ``limit`` without ``offset`` is treated as ``offset=1`` (a line
      count from the start) — the caller applies both one-sided
      semantics.

    Pure (no I/O) — unit tests drive it directly.
    """

    def _whole_line_number(value: Any) -> bool:
        return (
            isinstance(value, int) and not isinstance(value, bool) and value >= 1
        )

    if offset is None and limit is None:
        return True, "", None, None
    if (offset is not None and not _whole_line_number(offset)) or (
        limit is not None and not _whole_line_number(limit)
    ):
        return (
            False,
            BAD_RANGE_ARGS.format(offset=offset, limit=limit),
            None,
            None,
        )
    return True, "", offset, limit


def covered_spans(
    records: Sequence[tuple[str, str, int, int]], source: str, path: str
) -> list[tuple[int, int]]:
    """Merged coverage spans for ONE document (phase 141, task 02).

    Selects the ``read_coverage`` records keyed by the RESOLVED
    ``(source, path)`` pair (records for other documents are ignored)
    and returns their inclusive line spans, MERGED: overlapping spans
    collapse into their union (a span contained in another is
    absorbed) and ADJACENT spans join (``(1, 10) + (11, 20)`` →
    ``(1, 20)`` — adjacency counts as continuous coverage), sorted by
    first line. Pure over holder state — no DB.
    """
    merged: list[list[int]] = []
    for first, last in sorted(
        (a, b) for s, p, a, b in records if s == source and p == path
    ):
        if merged and first <= merged[-1][1] + 1:
            merged[-1][1] = max(merged[-1][1], last)
        else:
            merged.append([first, last])
    return [(first, last) for first, last in merged]


def fully_covered(
    spans: Sequence[tuple[int, int]], first: int, last: int
) -> bool:
    """Whether the requested ``(first, last)`` span sits inside the
    merged coverage union (phase 141, task 02) — per line, inclusive:
    one merged span must start at or before ``first`` and end at or
    after ``last``. Pure — no DB.
    """
    return any(
        span_first <= first and last <= span_last for span_first, span_last in spans
    )


def format_covered_spans(spans: Sequence[tuple[int, int]]) -> str:
    """Render merged coverage spans for the :data:`REREAD_PARTIAL`
    ``{covered}`` field (phase 141, task 02): comma-joined ``a–b``
    pairs (the en dash — one span → the bare ``a–b``; a single line →
    the bare ``a``), in span order. Pure — no DB.
    """
    return ", ".join(
        f"{first}" if first == last else f"{first}–{last}"
        for first, last in spans
    )


def grep_document(content: str, pattern: str) -> list[tuple[int, str]]:
    """Every line of *content* that contains *pattern*, in file order.

    Case-insensitive **fixed substring** (owner-locked A5: no regex — no
    ReDoS surface, a simple contract for the model). Returns
    ``(1-based line number, line.rstrip())`` pairs; an empty *content*
    never matches a non-empty pattern.
    """
    needle = pattern.lower()
    return [
        (number, line.rstrip())
        for number, line in enumerate(content.split("\n"), start=1)
        if needle in line.lower()
    ]


@dataclass
class AgentHolder:
    """Per-turn agent state the API layer reads after the stream (task 04).

    ``read_docs``: the documents ``read`` added to the context, in read
    order (deduped — re-reading a document appends nothing).
    ``tool_calls``: how many tool calls executed (re-lists included);
    rejected calls (unknown tool, unknown/missing arguments or document,
    already-in-context) do not count. Drives the per-turn log line's
    ``tool_calls=N`` field (task 04).
    ``scaffold_stripped``: how many chars of tool-scaffolding the
    turn's filters removed across the turn's requests (rounds + the
    forced final + any recovery, phase 71) — drives the per-turn log
    line's ``scaffold_stripped=N`` field on grounded turns (the
    deflected path computes its own total in ``app.api.chat``).
    ``read_truncations`` (phase 95): one ``(argument, chars_shown,
    chars_total)`` tuple per TRUNCATED ``read`` — the combined
    ``source/path`` the model passed, the cap kept
    (``settings.read_max_chars``), and the document's true length — in
    execution order. For a TRUNCATED RANGE read (phase 131, task 02)
    ``chars_total`` is the SLICE's length (the requested size of the
    part that was read), not the document's. Recorded on truncation only; a read at or under the
    cap appends nothing. A truncated read is still a **successful**
    call: ``tool_calls`` increments as today and ``read_docs`` appends
    as today — this list only carries the truncation signal the agent
    loop turns into :class:`app.rag.llm.ToolResultPiece` values (task 02
    surfaces them to the UI).
    ``shown_images`` (phase 124, task 02): the documents ``show_image``
    showed, in call order — at most one in practice (LOCKED A3: a
    second call is refused with :data:`ALREADY_SHOWING`, which appends
    nothing). Read by ``app.api.chat`` after the stream for the done
    frame's ``answer_image`` field (task 04) — the ``read_docs``
    precedent (an API-layer read-after-stream signal).
    ``read_coverage`` (phase 141, task 01): one ``(source, path,
    first_line, last_line)`` record per covered span — INCLUSIVE line
    numbers over the document's ``content.splitlines()``, in read
    order, keyed by the RESOLVED ``(source, path)`` pair (the resolved
    document's own identity — the same pair ``read_docs`` keys on; by
    the :func:`_resolve_path` first-slash contract the raw argument's
    split and the resolved pair agree). One span per SUCCESSFUL
    ``read`` — a refused read records NOTHING (the refusal style of
    every refusal: it counts in nothing): a plain read at or under the
    cap covers ``(1, total_lines)``; a truncated plain read covers
    line 1 through the count from ``splitlines()`` on the ACTUAL cut
    string, ``content[:cap]`` (the lines the model actually saw; the
    char cap is never re-derived into lines arithmetically); a range
    read covers ``(first, end)`` — the TRUE extent the ``lines:
    {first}-{end} of {total}`` header reports (an overshooting
    ``limit`` is already clamped into ``end``); a truncated range
    read covers the same cut count into the slice — ``first - 1 +
    len(slice[:cap].splitlines())``, clamped to ``end``. The
    reworked dedupe matrix (phase 141, task 02) decides on these
    spans; ``read_docs`` / ``read_truncations`` keep
    their EXACT current meaning (API-layer signal, SSE matching,
    per-turn log) — this list is NEW state beside them, not a
    replacement.
    """

    read_docs: list[Document] = field(default_factory=list)
    tool_calls: int = 0
    scaffold_stripped: int = 0
    read_truncations: list[tuple[str, int, int]] = field(default_factory=list)
    shown_images: list[Document] = field(default_factory=list)
    read_coverage: list[tuple[str, str, int, int]] = field(default_factory=list)


def _execute_tool(
    db: Session,
    call: ToolCallPiece,
    seed_docs: Sequence[Document],
    holder: AgentHolder,
    settings: Settings,
) -> str:
    """Execute one tool call server-side (DB only).

    Returns the tool result text. A successful call bumps
    ``holder.tool_calls`` (a successful ``read`` also appends the
    :class:`Document` to ``holder.read_docs``; a ``grep`` never does —
    it is a locator, locked A5); rejected calls return their refusal
    line and count in nothing. A grep that ran but found nothing is
    still a successful (counted) call — its no-match line is a result,
    not a refusal. A ``read`` longer than ``settings.read_max_chars``
    (phase 95) is cut at the cap and carries the shared
    :data:`TRUNCATION_MARKER` + the pinned :data:`READ_TRUNCATION_NOTICE`
    (the rest of the document is NOT in the model's context), and a
    ``(argument, cap, total)`` tuple is appended to
    ``holder.read_truncations`` (the signal the agent loop turns into a
    :class:`app.rag.llm.ToolResultPiece`). Phase 131 (task 02): a
    ``read`` may carry the optional line range ``offset`` (1-based
    first line) + ``limit`` (line count) — validated by
    :func:`_read_range_args` (a bad value → :data:`BAD_RANGE_ARGS`,
    FIRST — before any dedupe or resolve), then the COVERAGE dedupe
    matrix (phase 141, task 02 — the raw-arg fast path is RETIRED:
    whether a span is "full" needs the document's TRUE line count,
    which only the resolve knows, so EVERY dedupe row resolves the
    document first — one cheap indexed lookup; the pre-phase-141 bug
    a false refusal dead-ended a whole stuck turn, and that cost is
    what forces the trade): a range read PROCEEDS unless the
    requested lines are FULLY covered by earlier reads (the owner's
    stuck-scenario fix — includes the phase-131 truncated-then-range
    row, a truncated first read covering only its cut span), a
    re-read of fully-covered lines gets :data:`ALREADY_IN_CONTEXT`
    (the claim is then true), a plain re-read of a PARTIALLY-covered
    one gets :data:`REREAD_PARTIAL` (the covered span(s) named), an
    ``offset`` past the end → :data:`OFFSET_PAST_END` (after the
    resolve, before the matrix), the slice shaped under a ``lines:
    {offset}-{end} of {total_lines}`` header line and capped like the
    whole document (the tuple's ``chars_total`` is then the slice's
    length). Phase 141 (task 01): every SUCCESSFUL read also appends
    its covered span — the RESOLVED pair + the inclusive line extent
    of what actually reached the model (per the four outcomes above) —
    to ``holder.read_coverage`` (the NEW state the reworked dedupe
    matrix, task 02, decides on); a refused read records nothing, and
    ``read_docs`` / ``read_truncations`` keep their exact current
    meaning. Document targets are combined ``source/path``
    strings, resolved by :func:`_resolve_path` (the canonical identity,
    phase 70).
    """
    if call.name == "ls":
        # Phase 94 (task 03): the drill-down tree — one level per call
        # (the owner-permitted tool-surface revision; the old
        # whole-catalog listing is gone). A successful listing at ANY
        # level (top/root/folder) counts; a refusal counts in nothing
        # and consumes a round like every refusal.
        raw_path = call.arguments.get("path")
        scope = raw_path.strip() if isinstance(raw_path, str) else ""
        if not scope:
            # The top level: the synced sources, registry order, each
            # with its recursive document count and its stored
            # source-root summary (``None`` → no indented line).
            holder.tool_calls += 1
            return render_ls_top(ls_top(db))
        source, _, rest = scope.partition("/")
        if source not in list_source_names(db):
            # The no-source refusal with the teaching parenthetical
            # (phase 72, the prefix byte-identical to the pre-phase-72
            # line): the FIRST segment is the source candidate — a bare
            # unknown name (the incident's ``ls(path='.')``) or the
            # source segment of a ``source/…`` argument (phase 94: a
            # ``/`` now names a folder, so the phase-72 document-path
            # teaching is deleted) — the segment echoed; counts in
            # nothing, consumes a round like every refusal.
            return NO_SOURCE_NOT_A_DIRECTORY.format(scope=source)
        if not rest:
            # The source's ROOT folder: subfolders + own file lines
            # (capped + note) — the pinned template; a registered
            # source with no documents lists its header line alone
            # (``… — 0 documents, 0 folders:`` — the old
            # ``0 documents:`` behavior preserved in spirit).
            subfolders, files, total = ls_folder(db, source, "")
            holder.tool_calls += 1
            return render_folder_listing(source, subfolders, files, total)
        rows = _source_document_rows(db, source)
        summaries = _source_folder_summaries(db, source)
        if not _folder_exists_in(rows, rest):
            # The NOT-A-FOLDER teaching (phase 94, task 03): the
            # argument echoed, the DEEPEST existing ancestor's name and
            # its direct subfolders listed (bounded — the parent's own
            # listing, so no new flood path), so the model
            # self-corrects in the next round; counts in nothing,
            # consumes a round like every refusal.
            parent = _deepest_existing_ancestor(rows, rest)
            parent_subs = group_folder_listing(source, parent, rows, summaries)[0]
            return NOT_A_FOLDER.format(
                arg=scope,
                parent=source if not parent else f"{source}/{parent}",
                subfolders=" ".join(f"{sub}/" for sub, _c, _s in parent_subs)
                or "none",
            )
        # The folder level: the same template as the root, identity =
        # source + "/" + folder.
        subfolders, files, total = group_folder_listing(source, rest, rows, summaries)
        holder.tool_calls += 1
        return render_folder_listing(f"{source}/{rest}", subfolders, files, total)
    if call.name == "read":
        raw_path = call.arguments.get("path")
        arg = raw_path.strip() if isinstance(raw_path, str) else ""
        if not arg:
            return MISSING_READ_ARGS
        # Phase 131 (task 02): the OPTIONAL line range — validated
        # after the path check and BEFORE the dedupe: a bad range
        # argument is a refusal regardless of what is in context
        # (counts in nothing, consumes a round — the phase-72 style;
        # no silent argument repair, no DB lookup).
        ok, range_error, offset, limit = _read_range_args(
            call.arguments.get("offset"), call.arguments.get("limit")
        )
        if not ok:
            return range_error
        # A one-sided range: ``limit`` without ``offset`` counts from
        # the start of the document (first line 1 — the helper's
        # documented semantics); ``offset`` without ``limit`` reads to
        # the end (the slice math below).
        ranged = offset is not None or limit is not None
        first = offset if offset is not None else 1
        # Phase 118 (A6): ``holder.read_docs`` stays the read-once
        # bookkeeping for the API layer (``done.sources`` / the
        # per-turn log's ``sources=``) — the DEDUPE no longer reads
        # it: phase 141 (task 02) decides from the COVERAGE state
        # (below). The ``seed_docs`` are SUMMARY blocks in the prompt,
        # not full text, so a FIRST read of a suggested document still
        # adds its full text through the path below.
        known = {(doc.source, doc.path) for doc in holder.read_docs}
        # Phase 141 (task 02): the raw-arg fast path is RETIRED —
        # whether a span is "full" needs the document's TRUE line
        # count, which only the resolve knows, so EVERY dedupe row
        # resolves the document first (one cheap indexed lookup; the
        # fast path's only saving was that one query, and the
        # pre-phase-141 bug — a false refusal dead-ending a whole
        # stuck turn — is what forces the trade). A fully-covered
        # plain re-read — the only row the old fast path handled —
        # still refuses with the byte-identical
        # :data:`ALREADY_IN_CONTEXT`; its implementation just goes
        # through the resolve.
        doc, _source, _path = _resolve_path(db, arg)
        if doc is None:
            # Echo the argument as passed — the model sees its own form
            # (a bare argument can never be a document, no DB lookup);
            # a path-like argument that matches an indexed document's
            # path gets the "did you mean …?" teaching (phase 72).
            return _no_document_refusal(db, arg)
        lines = doc.content.splitlines()
        total_lines = len(lines)
        # The range read's TRUE extent (``end`` clamps an overshooting
        # ``limit`` to the last line; ``offset`` without ``limit``
        # reads to the end) — computed before the matrix (the range
        # coverage row needs it) and reused by the slice shaping
        # below.
        end = total_lines if limit is None else min(first + limit - 1, total_lines)
        if ranged and first > total_lines:
            # The offset names a line past the end — teaching refusal
            # (fires after the resolve, BEFORE the coverage matrix —
            # an offset past the end is refused regardless of
            # coverage): the true line count named, grep reports the
            # exact line numbers (the locator that got the model
            # here). Counts in nothing, consumes a round; the document
            # is NOT added to the context, so a corrected retry still
            # proceeds.
            return OFFSET_PAST_END.format(
                source=doc.source,
                path=doc.path,
                total_lines=total_lines,
                offset=first,
            )
        # The COVERAGE dedupe matrix (phase 141, task 02 — the
        # rework of the phase-131 pair-set dedupe): the merged spans
        # for this RESOLVED document decide.
        #   plain, full document covered → :data:`ALREADY_IN_CONTEXT`
        #   plain, partial               → :data:`REREAD_PARTIAL`
        #   (the covered span(s) named — the REREAD_TRUNCATED role,
        #   generalized: line-based truth)
        #   range, requested span fully  → :data:`ALREADY_IN_CONTEXT`
        #   covered (the named lines ARE in the prompt — the claim is
        #   true)
        #   range, not fully covered     → PROCEEDS (the owner's
        #   stuck-scenario fix — includes the phase-131
        #   truncated-then-range row: a truncated first read covers
        #   only its cut span)
        spans = covered_spans(holder.read_coverage, doc.source, doc.path)
        if not ranged:
            if fully_covered(spans, 1, total_lines):
                return ALREADY_IN_CONTEXT
            if spans:
                return REREAD_PARTIAL.format(
                    covered=format_covered_spans(spans),
                    source=doc.source,
                    path=doc.path,
                    total=total_lines,
                )
        elif fully_covered(spans, first, end):
            return ALREADY_IN_CONTEXT
        # The checks above refused every duplicate that must be
        # refused — append at most ONCE: a first read (plain or range)
        # appends; a PROCEEDING re-read of an earlier partial read of
        # the same document (the owner's second range) does not
        # re-append.
        if (doc.source, doc.path) not in known:
            holder.read_docs.append(doc)
        holder.tool_calls += 1
        # Phase 122 (task 05): an image document's content IS the vision
        # description — the marker line (a third line between the
        # byte-identical header/date lines and the text) tells the model
        # what it is reading. A text doc's ``marker`` is "" — the result
        # stays byte-identical to pre-122.
        marker = f"{IMAGE_DOC_MARKER}\n" if doc.is_image else ""
        cap = settings.read_max_chars
        if not ranged:
            # Plain read — the pre-phase-131 bytes, untouched.
            if len(doc.content) > cap:
                # Phase 95 (owner permission 2026-09-10, ``TODO.md``
                # L5): the read cap — cut at the cap and tell the model
                # the truth: the shared marker + the pinned notice name
                # how much more exists and teach the grep →
                # read(offset/limit) workflow (phase 131 — the notice
                # itself names the follow-up call). A truncated read is
                # still a successful call (both counters above already
                # bumped); the holder tuple is the truncation signal
                # only, so the loop can yield a ToolResultPiece.
                # ``raw_path`` (a str here — ``arg`` is non-empty only
                # when it was) is the exact argument the matching SSE
                # ``tool`` frame carries, so the UI can match the two
                # (task 02); the cast keeps pyright honest about the
                # holder tuple's first element.
                holder.read_truncations.append(
                    (cast("str", raw_path), cap, len(doc.content))
                )
                # Phase 141 (task 01): coverage tracking — the lines
                # this cut actually put in the model's context: from
                # line 1 to the last line visible in the ACTUAL cut
                # string (``splitlines()`` on ``content[:cap]`` — the
                # counted-from-truth rule; a cap that cuts mid-line
                # ends the span at that line's visible prefix; the
                # char cap is never re-derived into lines
                # arithmetically). Task 02's dedupe matrix decides on
                # these spans — task 01 only records.
                holder.read_coverage.append(
                    (doc.source, doc.path, 1, len(doc.content[:cap].splitlines()))
                )
                # Phase 106 (D5): the date rides every document the
                # model sees — the ``read`` result's SECOND line; the
                # FIRST line stays ``Document {source}/{path}:``
                # BYTE-IDENTICAL (the E2E mock's ``_READ_RESULT_PREFIX``
                # header contract).
                return (
                    f"Document {doc.source}/{doc.path}:\n"
                    f"date: {doc.created_at:%Y-%m-%d}\n"
                    f"{marker}"
                    f"{doc.content[:cap]}\n"
                    f"{TRUNCATION_MARKER}\n"
                    f"{READ_TRUNCATION_NOTICE.format(shown=cap, total=len(doc.content))}"
                )
            # At or under the cap: the pre-phase-95 result plus the
            # phase-106 D5 date line (first line byte-identical — the
            # mock's header contract; no truncation marker, no notice,
            # no TRUNCATION holder entry, no ToolResultPiece) — and,
            # phase 122, the image-document marker line for image
            # docs only.
            # Phase 141 (task 01): a plain read at or under the cap
            # covers the WHOLE document — lines 1..total_lines.
            holder.read_coverage.append((doc.source, doc.path, 1, total_lines))
            return (
                f"Document {doc.source}/{doc.path}:\n"
                f"date: {doc.created_at:%Y-%m-%d}\n"
                f"{marker}"
                f"{doc.content}"
            )
        # Range read (phase 131, task 02): EXACTLY the named lines —
        # the TRUE extent on the third line (``end`` — computed
        # before the matrix above: it clamps to the last line, an
        # overshooting ``limit`` reads to the end, never past it;
        # ``offset`` without ``limit`` reads to the end), the
        # image-doc marker (when one) exactly where the plain read
        # puts it (after the header lines), the cap applied to the
        # SLICE.
        slice_text = "\n".join(lines[first - 1 : end])
        if len(slice_text) > cap:
            # The cap on the slice: the same marker + the pinned notice
            # (``{total}`` = the slice's length — how much of THIS read
            # was cut); the holder tuple's ``chars_total`` is the
            # slice's length, NOT the document's (documented on
            # :class:`AgentHolder.read_truncations`).
            holder.read_truncations.append(
                (cast("str", raw_path), cap, len(slice_text))
            )
            # Phase 141 (task 01): the truncated range read covers
            # from ``first`` to the last line visible in the cut
            # SLICE — the same counted-from-truth rule as the plain
            # read above (``splitlines()`` on the actual cut string),
            # clamped to ``end``.
            holder.read_coverage.append(
                (
                    doc.source,
                    doc.path,
                    first,
                    min(end, first - 1 + len(slice_text[:cap].splitlines())),
                )
            )
            return (
                f"Document {doc.source}/{doc.path}:\n"
                f"date: {doc.created_at:%Y-%m-%d}\n"
                f"lines: {first}-{end} of {total_lines}\n"
                f"{marker}{slice_text[:cap]}\n"
                f"{TRUNCATION_MARKER}\n"
                f"{READ_TRUNCATION_NOTICE.format(shown=cap, total=len(slice_text))}"
            )
        # Phase 141 (task 01): the range read covers exactly the TRUE
        # extent the ``lines:`` header reports (``end`` already
        # clamped an overshooting ``limit`` — the span never goes
        # past the document's last line).
        holder.read_coverage.append((doc.source, doc.path, first, end))
        return (
            f"Document {doc.source}/{doc.path}:\n"
            f"date: {doc.created_at:%Y-%m-%d}\n"
            f"lines: {first}-{end} of {total_lines}\n"
            f"{marker}{slice_text}"
        )
    if call.name == "show_image":
        # Phase 124 (task 02; LOCKED A3/A5): the 4th agent tool — the
        # model shows a standalone image document as part of its
        # answer. Offered only when ``settings.answer_images``
        # (``agent_tools``); the handler enforces the contract
        # server-side — image docs only, ONE image per answer. Every
        # refusal below counts in nothing and consumes a round, like
        # every other refusal.
        raw_path = call.arguments.get("path")
        arg = raw_path.strip() if isinstance(raw_path, str) else ""
        if not arg:
            return MISSING_SHOW_ARGS
        doc, _source, _path = _resolve_path(db, arg)
        if doc is None:
            # The same phase-72 "did you mean …?" teaching as the read
            # branch (a refusal — not counted, no context).
            return _no_document_refusal(db, arg)
        if not doc.is_image:
            # LOCKED A5: a text doc is refused, never silently
            # converted to a read — the tool shows IMAGE documents
            # only.
            return NOT_AN_IMAGE_DOC.format(arg=arg)
        if holder.shown_images:
            # LOCKED A3: one image per answer — server-enforced, not
            # just taught.
            return ALREADY_SHOWING
        holder.shown_images.append(doc)
        holder.tool_calls += 1
        # The confirming result: a NEW first line (absent from ``read``
        # — the model must know the show registered) followed by the
        # exact document block ``read`` returns for an image doc at-or-
        # under-cap (the ``Document …`` header + phase-106 date line +
        # :data:`IMAGE_DOC_MARKER` — the header contract the E2E mock
        # pins on ``read`` results is deliberately reused here). An
        # image doc's content is the summary-capped vision description
        # — short by construction (phase 122, task 03), so NO
        # truncation machinery; the plain block stays correct even if
        # it ever exceeded the cap.
        return (
            "Image shown to the user — it is now part of your answer.\n"
            f"Document {doc.source}/{doc.path}:\n"
            f"date: {doc.created_at:%Y-%m-%d}\n"
            f"{IMAGE_DOC_MARKER}\n"
            f"{doc.content}"
        )
    if call.name == "grep":
        raw_pattern = call.arguments.get("pattern")
        pattern = raw_pattern.strip() if isinstance(raw_pattern, str) else ""
        if not pattern:
            return MISSING_SEARCH_ARGS
        raw_path = call.arguments.get("path")
        scope = raw_path.strip() if isinstance(raw_path, str) else ""
        scoped_to: tuple[str, str] | None = None
        if scope:
            target, src, p = _resolve_path(db, scope)
            if target is None:
                # The same phase-72 "did you mean …?" teaching as the
                # read branch (a refusal — not counted, no context).
                return _no_document_refusal(db, scope)
            docs: list[Document] = [target]
            scoped_to = (src, p)  # the resolved (canonical) identity
        else:
            docs = all_documents(db)
        matches: list[str] = []
        for doc in docs:
            for lineno, line in grep_document(doc.content, pattern):
                matches.append(
                    f"{doc.source}/{doc.path}:{lineno}: {line[:SEARCH_LINE_LIMIT]}"
                )
                if len(matches) >= SEARCH_MAX_MATCHES:
                    break
            if len(matches) >= SEARCH_MAX_MATCHES:
                break  # the global cap is hit — stop scanning
        holder.tool_calls += 1  # the grep executed (no-match counts too)
        # Locked A5: a grep never adds context — read_docs untouched.
        if not matches:
            shown = pattern[:100]  # keep a long pattern short in the line
            # The 2026-09-05 incident teaching: a regex-shaped pattern
            # (the harness prior) can NEVER match a fixed-substring
            # grep, so a no-match for one is not "the KB lacks this" —
            # it is "the pattern was in the wrong form". Teach the
            # contract and hand over the plain-form retry hint (when the
            # reduction is non-empty); a plain pattern (or a pattern
            # that reduces to nothing) keeps the ordinary line
            # byte-identical.
            plain = plain_form(pattern) if looks_like_regex(pattern) else ""
            if plain:
                if scoped_to is not None:
                    return NO_MATCHES_REGEX_SCOPED.format(
                        pattern=shown,
                        source=scoped_to[0],
                        path=scoped_to[1],
                        plain=plain[:100],
                    )
                return NO_MATCHES_REGEX.format(pattern=shown, plain=plain[:100])
            if scoped_to is not None:
                # The scoped no-match line is keyed on the resolved
                # source/path (== the argument, stripped).
                return NO_MATCHES_SCOPED.format(
                    pattern=shown, source=scoped_to[0], path=scoped_to[1]
                )
            return NO_MATCHES.format(pattern=shown)
        return "\n".join(matches)
    return UNKNOWN_TOOL


async def run_agent(
    llm: LLMClient,
    db_factory: Callable[[], Session],
    *,
    system_prompt: str,
    user_message: str | list[dict[str, Any]],
    seed_docs: Sequence[Document],
    settings: Settings,
    holder: AgentHolder,
    history: Sequence[dict[str, Any]] = (),
) -> AsyncIterator[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece]:
    """Run the grounded-turn tool loop, yielding every stream piece.

    Every piece (``thinking`` / ``content`` / tool calls /
    :class:`RetryPiece`) is yielded as it arrives; the API layer (task 04)
    turns tool-call pieces into SSE ``tool`` events and retry pieces into
    SSE ``retry`` events. A truncated ``read`` (phase 95 — the document is
    longer than ``settings.read_max_chars``) additionally yields one
    :class:`ToolResultPiece` per truncated call, AFTER the round's
    ``tool`` frame and BEFORE the next model round (the API layer turns it
    into an SSE ``tool_result`` frame, task 02). After the loop finishes,
    *holder* carries the read documents, the executed tool-call count
    (re-lists included), the ``read_truncations`` signal list, and the
    ``read_coverage`` spans (phase 141, task 01 — the lines each
    successful read actually put in the model's context).

    History (phase 74, TODO L4): *history* is the client's prior turns
    already mapped to model messages by
    :func:`app.rag.prompts.history_to_messages` (trimmed newest-first
    against the settings budgets; assistant turns carry their prior
    thinking as ``reasoning_content``). It is spliced between the system
    prompt and the current user message —
    ``[system, *history, user]`` — and everything downstream (the tool
    rounds, the phase-71 recovery rebuilding from ``messages[1:]``, the
    retry restarts) already operates on that one ``messages`` list,
    unchanged. ``()`` (the default) keeps the pre-phase-74 two-message
    request byte-identical.

    User message (phase 123, TODO L6): *user_message* is the current
    turn's user content — the plain question string (every text-only
    turn, byte-identical to pre-phase) OR the multimodal content list
    ``[{type: "text", …}, {type: "image_url", …}]`` for a question that
    carried an image. FLOW (pinned): the API layer
    (``app.api.chat``) builds the content via its ``build_user_content``
    helper and passes it HERE — ``run_agent`` builds its OWN
    ``[system, *history, user]`` list from this value (it does NOT
    receive the already-built ``messages``; the deflected branch is the
    one that consumes chat.py's list directly), so a single value
    covers both shapes and the tool rounds / recovery / retries operate
    on it untouched. Prior turns' images are never replayed (LOCKED A7
    — *history* is text-only by construction).

    Retries (phase 67, owner-locked A2): every model request goes through
    :func:`chat_stream_retried` — a failed round is retried **before** its
    first piece (same messages, ``settings.llm_retries`` restarts, a flat
    ``settings.llm_retry_delay``); a round that already streamed pieces
    fails the turn as before.

    Scaffolding recovery (phase 71, deterministic only): every request —
    each round, the forced final, and any recovery — runs its content
    through a fresh :class:`app.rag.scaffolding.ScaffoldingFilter`. A
    round that ends with NO visible content but a non-empty strip (the
    scaffolding was the whole "answer") gets exactly ONE recovery:
    ``tools=None``, :data:`CORRECTION_INSTRUCTION` folded into the
    original single system message (the rest of the history — user
    message and tool results — unchanged), a fresh filter, the same
    retry budget. A clean recovery ends the turn; a second empty reply
    raises :class:`MalformedReplyError` (terminal — the API layer turns
    it into the dedicated error frame). A round with visible content
    plus scaffolding needs no recovery (the clean content stands), and
    a scaffolding-only round that also carried tool calls needs none
    (the tool ran) — the policy keys on the no-calls exit only. The
    per-span strip warning log (each span truncated to 200 chars) is
    the capture mechanism for new registry entries; *holder* accumulates
    the turn's ``scaffold_stripped`` total for the API layer's log line.

    ``seed_docs`` are the suggested documents whose SUMMARY blocks the
    caller put in the *system_prompt* (phase 118: the retrieval seeds
    summaries, never full texts); reading one of them ADDS its full
    text to the context through the ordinary ``read`` path — a re-read
    whose lines are FULLY covered by earlier reads is rejected with
    :data:`ALREADY_IN_CONTEXT` (the phase-72 teaching line — answer
    from the text already in the prompt; phase 141: the coverage
    state decides, so the claim is then true, and a re-read of a
    PARTIALLY-covered document gets :data:`REREAD_PARTIAL` instead —
    the covered span(s) named, read the rest with ``offset``/
    ``limit``); the rejection counts in nothing, but it still consumes
    a round.

    DB sessions (SEC-14-04): *db_factory* is a callable that returns a
    new :class:`sqlalchemy.orm.Session` (e.g. ``lambda: SessionLocal()``).
    Each tool call creates its own short-lived session via *db_factory*
    and closes it after the tool result is produced — no session is held
    across rounds, eliminating SSE-stream DB-connection pinning.
    """
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": system_prompt},
        *history,  # phase 74: the client's prior turns (empty by default)
        {"role": "user", "content": user_message},
    ]
    # Phase 45: no per-tool budgets — the tools stay offered for the
    # whole turn, bounded by the round cap. ``0`` is the no-tools kill
    # switch: exactly one request with ``tools=None`` (the pre-phase-37
    # path). Phase 124 (task 02): the offer list is :func:`agent_tools`
    # — the three base tools (byte-identical) plus ``show_image`` only
    # when ``settings.answer_images`` (``BOR_ANSWER_IMAGES``) is on;
    # off (the default) the offer is element-for-element the
    # pre-phase-124 list.
    max_rounds = settings.agent_max_rounds
    tools: list[dict[str, Any]] | None = (
        agent_tools(settings) if max_rounds > 0 else None
    )
    rounds = 0
    while True:
        calls: list[ToolCallPiece] = []
        # Phase 71: one fresh filter per round (one per model request —
        # the retry attempts of this logical request share it: a restart
        # only happens while the filter was never fed). Content-only:
        # thinking pieces pass through raw.
        round_filter = ScaffoldingFilter()
        round_content = 0  # visible (clean) content chars this round
        # Phase 48: bind the round's stream so a consumer abandon
        # (GeneratorExit into the yield below) tears down the in-flight
        # model stream deterministically — not GC-dependent. Phase 67:
        # the round goes through the retry primitive — a failure before
        # the first piece restarts the request (locked A2) after a
        # RetryPiece; closing the OUTER generator propagates GeneratorExit
        # into ``chat_stream_retried``, whose own ``finally`` closes the
        # in-flight inner ``chat_stream``, so teardown stays deterministic
        # on consumer abandon. Awaiting ``aclose()`` in the ``finally`` is
        # safe because it does not yield; on a fully consumed round it is
        # a quiet no-op.
        stream = chat_stream_retried(
            llm,
            messages,
            tools=tools,
            retries=settings.llm_retries,
            delay=settings.llm_retry_delay,
            scaffolding=round_filter,
        )
        try:
            async for piece in stream:
                if isinstance(piece, ToolCallPiece):
                    calls.append(piece)
                elif isinstance(piece, StreamPiece) and piece.kind == "content":
                    round_content += len(piece.text)
                yield piece
        finally:
            await stream.aclose()
        # Phase 71: the per-strip-event capture log — one warning per
        # stripped span, truncated to 200 chars (how a new scaffolding
        # format gets captured and added to the registry) — and the turn
        # total for the API layer's ``scaffold_stripped=N`` log field.
        holder.scaffold_stripped += round_filter.stripped_chars
        for span in round_filter.stripped_spans:
            logger.warning(
                "agent: stripped %d chars of tool-scaffolding in round %d: %r",
                len(span),
                rounds + 1,
                span[:200],
            )
        if not calls:
            if round_content > 0:
                return  # the answer was streamed (the clean content stands)
            if round_filter.stripped_chars == 0:
                # Empty/thinking-only answer — today's behavior, unchanged
                # (the UI handles it); the guardrail keys on a strip.
                return
            # Phase 71: the scaffolding was the whole "answer" — the ONE
            # bounded recovery (a fixed policy, not a conversation):
            # ``tools=None``, the correction folded into the ORIGINAL
            # single system message (provider-safe — the user message and
            # any tool history stay in place), a fresh filter, the same
            # phase-67 retry budget.
            logger.warning(
                "agent: round %d was pure tool-scaffolding (%d chars stripped) "
                "— running the one bounded recovery",
                rounds + 1,
                round_filter.stripped_chars,
            )
            messages_recovered = [
                {
                    "role": "system",
                    "content": system_prompt + "\n" + CORRECTION_INSTRUCTION,
                },
                *messages[1:],
            ]
            recovery_filter = ScaffoldingFilter()
            recovered = chat_stream_retried(
                llm,
                messages_recovered,
                tools=None,
                retries=settings.llm_retries,
                delay=settings.llm_retry_delay,
                scaffolding=recovery_filter,
            )
            recovery_content = 0
            try:
                async for piece in recovered:
                    if isinstance(piece, StreamPiece) and piece.kind == "content":
                        recovery_content += len(piece.text)
                    yield piece
            finally:
                await recovered.aclose()
            holder.scaffold_stripped += recovery_filter.stripped_chars
            for span in recovery_filter.stripped_spans:
                logger.warning(
                    "agent: stripped %d chars of tool-scaffolding in the "
                    "recovery after round %d: %r",
                    len(span),
                    rounds + 1,
                    span[:200],
                )
            if recovery_content > 0:
                return  # the recovery answered — the turn ends
            # The second empty reply is terminal (at most one recovery per
            # turn). Raised OUTSIDE the stream, so chat_stream_retried's
            # retry rule never sees it; the API layer catches it before
            # the generic LLMError handler.
            logger.warning(
                "agent: the recovery reply was still empty "
                "(scaffold_stripped=%d) — settling with a malformed-reply error",
                holder.scaffold_stripped,
            )
            raise MalformedReplyError(
                f"the model answered in raw tool-scaffolding twice in a row "
                f"(round {rounds + 1} plus one recovery) — no clean answer "
                "to stream"
            )
        call = calls[0]  # a stream can carry several calls; run the first
        # Phase 95: snapshot the truncation list BEFORE the execution so
        # only the entries THIS call added are surfaced (one read per
        # round, so at most one new entry — the loop still iterates the
        # tail, so a future multi-call round stays correct).
        trunc_before = len(holder.read_truncations)
        with db_factory() as tool_db:
            result = _execute_tool(tool_db, call, seed_docs, holder, settings)
        rounds += 1  # every call the model emits consumes a round
        logger.info(
            "agent tool=%s args=%s round=%d/%d",
            call.name,
            json.dumps(call.arguments, ensure_ascii=False)[:200],
            rounds,
            max_rounds,
        )
        messages.append(
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": call.id,
                        "type": "function",
                        "function": {
                            "name": call.name,
                            "arguments": json.dumps(call.arguments),
                        },
                    }
                ],
            }
        )
        messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
        # Phase 95: surface this round's truncation(s) to the API layer.
        # The piece lands AFTER the round's ``tool`` frame (the matching
        # ``ToolCallPiece`` was already streamed, above) and BEFORE the
        # next model round (the loop continues below) — task 02 turns it
        # into an SSE ``tool_result`` frame + the UI marker.
        for argument, shown, total in holder.read_truncations[trunc_before:]:
            yield ToolResultPiece(
                name=call.name,
                argument=argument,
                truncated=True,
                chars_shown=shown,
                chars_total=total,
            )
        if rounds >= max_rounds:
            logger.warning(
                "agent round cap reached (rounds=%d) — forcing a final "
                "no-tools answer",
                rounds,
            )
            # Phase 48: the forced final answer gets the same explicit
            # teardown as the loop rounds (consumer abandon mid-final
            # answer must still close the model's stream). Phase 67: the
            # forced call retries under the same locked-A2 rule as the
            # loop rounds. Phase 71: the forced final runs through a
            # fresh filter too — raw scaffolding can never reach the
            # user from ANY grounded request.
            final_filter = ScaffoldingFilter()
            final = chat_stream_retried(
                llm,
                messages,
                tools=None,
                retries=settings.llm_retries,
                delay=settings.llm_retry_delay,
                scaffolding=final_filter,
            )
            final_content = 0
            try:
                async for piece in final:
                    if isinstance(piece, StreamPiece) and piece.kind == "content":
                        final_content += len(piece.text)
                    yield piece
            finally:
                await final.aclose()
            # Phase 71: the same capture log + turn total; a
            # scaffolding-only forced final (this turn used no recovery,
            # so nothing is doubled up) settles with the same terminal
            # malformed-reply error rather than a silently empty answer.
            holder.scaffold_stripped += final_filter.stripped_chars
            for span in final_filter.stripped_spans:
                logger.warning(
                    "agent: stripped %d chars of tool-scaffolding in the "
                    "forced final answer (round %d): %r",
                    len(span),
                    rounds + 1,
                    span[:200],
                )
            if final_content == 0 and final_filter.stripped_chars > 0:
                logger.warning(
                    "agent: the forced final answer was pure tool-scaffolding "
                    "— settling with a malformed-reply error"
                )
                raise MalformedReplyError(
                    "the forced final answer was raw tool-scaffolding — no "
                    "clean answer to stream"
                )
            return