"""Integration: the agent DB accessors against real Postgres (phase 37;
the harness-aligned ``ls``/``read``/``grep`` surface, phase 70; the
drill-down tree ``ls``, phase 94).

``_source_document_rows`` must order a source's rows by ``path`` (the
file lines' catalog order), ``list_source_names`` must resolve the
registered source names (the registry join), and ``find_document``
must resolve a hit to the full document row (content included, for the
never-truncated read) and return ``None`` for unknown pairs. Phase 70:
the ``ls``/``read``/``grep`` tools are pinned here too — the locked
parameter shape in ``AGENT_TOOLS``, and scripted ``ToolCallPiece``s
executed through ``run_agent`` against the real DB. Phase 94: the
drill-down ``ls`` against the REAL tables — ``ls()`` lists the
registered sources (registry order, recursive counts, stored
source-root summaries from ``folder_summaries``), ``ls(source)`` /
``ls(source/folder)`` list one folder level (the SQL prefix logic:
subfolders = slash-boundary prefixes, counts = the recursive subtree,
file lines in catalog order, capped at 50 + the grep-pointer note),
and the refusals (unknown source segment → the no-source refusal; an
unknown folder → NOT-A_FOLDER with the parent's subfolders).
``read`` runs on the canonical combined ``source/path`` form
(first-slash split; a bare source name and an unknown identity get the
no-document refusal), and ``grep`` (``all_documents`` for a whole-KB
search, ``find_document`` for a scoped one) — both byte-identical
across the phase-94 change. Phase 106 (D5): the ``ls`` FILE line ends
with the appended `` | date: YYYY-MM-DD`` field and the ``read``
result carries the ``date: YYYY-MM-DD`` second line (first line
byte-identical) — the fixture documents carry a fixed ``created_at``
so the pins stay deterministic.

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

import asyncio
import uuid
from collections.abc import AsyncIterator, Callable, Iterator
from copy import deepcopy
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any, cast

import pytest
from sqlalchemy import delete, text
from sqlalchemy.orm import Session

from app.config import Settings
from app.db import SessionLocal
from app.models import Document, FolderSummary, GitSource
from app.rag import agent
from app.rag.agent import AGENT_TOOLS, AgentHolder, run_agent
from app.rag.llm import (
    LLMClient,
    RetryPiece,
    StreamPiece,
    ToolCallPiece,
    ToolResultPiece,
)

if TYPE_CHECKING:
    from app.rag.scaffolding import ScaffoldingFilter

#: The fixture documents' fixed creation date (phase 106, D5) — the
#: ``ls`` file line and the ``read`` second line format its UTC date
#: part; a fixed value keeps the pins deterministic.
_FIXTURE_CREATED_AT = datetime(2024, 6, 15, 12, 0, 0, tzinfo=UTC)


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


@pytest.fixture()
def kb(db) -> Iterator[None]:
    """Fresh documents + folder_summaries tables (chunks first — the FK)
    for these accessors (phase 94: the drill-down ``ls`` reads the
    stored summaries too)."""
    db.execute(text("TRUNCATE chunks, documents, folder_summaries"))
    db.commit()
    yield
    db.execute(text("TRUNCATE chunks, documents, folder_summaries"))
    db.commit()


@pytest.fixture()
def registry(db) -> Iterator[None]:
    """A FRESH two-source registry (phase 94): the drill-down ``ls``
    top level IS the registry, so the table is truncated and re-seeded
    around the tests in a controlled ``(added_at, id)`` order —
    ``Deployments`` before ``Homelab`` (the top-level listing order)."""
    db.execute(text("TRUNCATE git_sources"))
    db.commit()
    db.add(GitSource(url="https://github.com/reese/Deployments.git", kind="git"))
    db.commit()
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git"))
    db.commit()
    yield
    db.execute(text("TRUNCATE git_sources"))
    db.commit()


@pytest.fixture()
def src(db) -> Iterator[GitSource]:
    """One registered git source — the scoped ``ls`` source-name check
    reads the real registry, so the row is inserted and deleted around
    the tests (``repo_name`` resolves the URL to ``Homelab``)."""
    row = GitSource(url="https://github.com/reese/Homelab.git", kind="git")
    db.add(row)
    db.commit()
    yield row
    db.execute(delete(GitSource).where(GitSource.id == row.id))
    db.commit()


def test_source_document_rows_order_by_path_within_the_source(kb, db) -> None:
    """Phase 94: the file lines' order — the source's rows in ``path``
    order (the old ``list_catalog``'s per-source ordering, now the
    ``ls`` folder-level accessor's contract; a different source's rows
    never leak in)."""
    _doc(db, "Zeta", "b/second.md", "Zeta B", "ZB")
    _doc(db, "Zeta", "a/first.md", "Zeta A", "ZA")
    _doc(db, "Alpha", "c/third.md", "Alpha C", "AC")
    db.commit()

    assert agent._source_document_rows(db, "Zeta") == [
        ("a/first.md", "Zeta A", "2024-06-15"),
        ("b/second.md", "Zeta B", "2024-06-15"),
    ]


def test_source_document_rows_is_empty_without_rows(kb, db) -> None:
    assert agent._source_document_rows(db, "Zeta") == []


def test_list_source_names_resolves_registry_rows(db) -> None:
    """The real registry: git names resolve through the import pipeline's
    ``repo_name`` (trailing ``.git`` stripped); a second row resolving to
    the same name (the phase-69 sibling case) is listed once."""
    a = GitSource(url="https://github.com/reese/Homelab.git", kind="git")
    b = GitSource(url="https://github.com/reese/Homelab", kind="git")  # sibling
    c = GitSource(url="https://e.com/deployments", kind="git")
    db.add_all([a, b, c])
    db.commit()
    try:
        assert agent.list_source_names(db).count("Homelab") == 1  # deduped
        assert "deployments" in agent.list_source_names(db)
    finally:
        db.execute(delete(GitSource).where(GitSource.id.in_([a.id, b.id, c.id])))
        db.commit()


def test_find_document_hit_returns_full_row(kb, db) -> None:
    created = _doc(db, "Alpha", "deep/nested/doc.md", "The Doc", "FULL-TEXT")
    db.commit()

    found = agent.find_document(db, "Alpha", "deep/nested/doc.md")
    assert found is not None
    assert found.id == created.id
    assert found.source == "Alpha"
    assert found.path == "deep/nested/doc.md"
    assert found.title == "The Doc"
    assert found.content == "FULL-TEXT"  # the read tool feeds this, untruncated


def test_find_document_none_for_unknown_pairs(kb, db) -> None:
    _doc(db, "Alpha", "x.md", "X", "X-CONTENT")
    db.commit()

    assert agent.find_document(db, "Alpha", "nope.md") is None  # wrong path
    assert agent.find_document(db, "Beta", "x.md") is None  # wrong source
    assert agent.find_document(db, "nope", "nope.md") is None  # nothing at all


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


def test_agent_tools_offers_the_harness_aligned_surface() -> None:
    by_name = {t["function"]["name"]: t for t in AGENT_TOOLS}
    assert list(by_name) == [  # the harness order, phase 70
        "ls",
        "read",
        "grep",
    ]
    ls = by_name["ls"]["function"]["parameters"]
    assert ls["type"] == "object"
    assert ls["required"] == []  # path is optional
    assert set(ls["properties"]) == {"path"}
    read = by_name["read"]["function"]["parameters"]
    assert read["type"] == "object"
    assert read["required"] == ["path"]
    assert set(read["properties"]) == {"path"}
    grep = by_name["grep"]["function"]["parameters"]
    assert grep["type"] == "object"
    assert grep["required"] == ["pattern"]
    assert set(grep["properties"]) == {"pattern", "path"}
    assert all(p["type"] == "string" for p in grep["properties"].values())


class ScriptedToolLLM:
    """One scripted tool-call stream, then one canned answer stream.
    Records every ``chat_stream`` request's messages and tools."""

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

    async def chat_stream(
        self,
        messages: list[dict[str, str]],
        tools: list[dict[str, Any]] | None = None,
        scaffolding: ScaffoldingFilter | None = None,  # phase 71 pass-through
    ) -> AsyncIterator[StreamPiece | ToolCallPiece]:
        self.requests.append((deepcopy(messages), deepcopy(tools)))
        if len(self.requests) == 1:
            yield self.call
        else:
            yield StreamPiece("content", "ans")


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


class ScriptedToolCallsLLM:
    """N scripted tool-call rounds (one ``ToolCallPiece`` each), then one
    canned answer; records every request's messages and tools (the
    phase-72 task-02 two-round self-correction cases: the refusal round,
    then the corrected call)."""

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

    async def chat_stream(
        self,
        messages: list[dict[str, str]],
        tools: list[dict[str, Any]] | None = None,
        scaffolding: ScaffoldingFilter | None = None,
    ) -> AsyncIterator[StreamPiece | ToolCallPiece]:
        self.requests.append((deepcopy(messages), deepcopy(tools)))
        index = len(self.requests) - 1
        if index < len(self.calls):
            yield self.calls[index]
        else:
            yield StreamPiece("content", "ans")


def _run_call(
    db: Session, name: str, arguments: dict[str, Any]
) -> tuple[AgentHolder, ScriptedToolLLM]:
    """Drive one scripted tool call through ``run_agent``.

    SEC-14-04: the session factory creates a short-lived session per tool
    call — the fixture session (*db*) is used to seed the KB, but each
    tool round opens its own session via ``SessionLocal()``, executes the
    tool, and closes it (the same pattern as production).
    """
    holder = AgentHolder()
    llm = ScriptedToolLLM(ToolCallPiece(id="call_1", name=name, arguments=arguments))
    # Create a factory that opens a fresh short-lived session per call
    def _db_factory() -> Session:
        return SessionLocal()

    asyncio.run(_consume(cast("LLMClient", llm), _db_factory, holder))
    return holder, llm


async def _consume(
    llm: LLMClient,
    db_factory: Callable[[], Session],
    holder: AgentHolder,
) -> list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece]:
    out: list[StreamPiece | ToolCallPiece | RetryPiece | ToolResultPiece] = []
    async for piece in run_agent(
        llm,
        db_factory,  # SEC-14-04: session factory (short-lived sessions)
        system_prompt="SYSTEM_PROMPT",
        user_message="QUESTION",
        seed_docs=[],
        settings=_settings(),
        holder=holder,
    ):
        out.append(piece)
    return out


# ---------- ls (the drill-down tree, phase 94 — the real registry + DB) ----------


def test_ls_top_level_lists_registered_sources_through_run_agent(
    kb, registry, db
) -> None:
    """No path: the TOP level against the real tables — registry order
    (``(added_at, id)`` — Deployments before Homelab), recursive counts
    (all of a source's documents), the stored source-root summary
    (``folder_path = ''``) shown only when stored."""
    _doc(db, "Deployments", "a/one.md", "A1", "A1-CONTENT")
    _doc(db, "Homelab", "x.md", "X", "X-CONTENT")
    _doc(db, "Homelab", "y/z.md", "Z", "Z-CONTENT")
    db.add(FolderSummary(source="Homelab", folder_path="", summary="Home lab notes."))
    db.commit()

    holder, llm = _run_call(db, "ls", {})

    assert llm.requests[0][1] == AGENT_TOOLS
    assert llm.requests[1][0][3]["content"] == (
        "2 sources:\n"
        "\n"
        "Deployments — 1 documents\n"
        "Homelab — 2 documents\n"
        "  Home lab notes."
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []


def test_ls_top_level_empty_registry_through_run_agent(
    kb, db, monkeypatch: pytest.MonkeyPatch
) -> None:
    """No registered sources: the top level is the header line alone
    (``0 sources:`` — the old ``0 documents:`` behavior preserved in
    spirit), still counted. The env fallback (``BOR_GIT_SOURCES`` — the
    operator's ``.env`` may name sources) is emptied for the test, so
    the registry is genuinely empty."""
    import app.rag.git_sources as git_sources_mod

    db.execute(text("TRUNCATE git_sources"))
    db.commit()
    monkeypatch.setattr(
        git_sources_mod,
        "get_settings",
        lambda: _settings(git_sources=""),
    )
    _doc(db, "Orphan", "a.md", "A", "A-CONTENT")  # indexed but unregistered
    db.commit()

    holder, llm = _run_call(db, "ls", {})
    assert llm.requests[1][0][3]["content"] == "0 sources:"
    assert holder.tool_calls == 1


def test_ls_source_scope_lists_root_folder_through_run_agent(kb, registry, db) -> None:
    """A registered source name: the source's ROOT folder — the direct
    subfolders (path order, recursive counts, stored summaries attached)
    + the root's own file lines in catalog order — against the real
    tables; a registered source with no documents lists its header
    line alone."""
    _doc(db, "Homelab", "backups/cron.md", "Cron", "CRON")
    _doc(db, "Homelab", "backups/restic.md", "Restic", "RESTIC")
    _doc(db, "Homelab", "networking/lan.md", "LAN", "LAN")
    _doc(db, "Homelab", "readme.md", "Readme", "README")
    db.add(
        FolderSummary(
            source="Homelab", folder_path="backups", summary="Backup notes."
        )
    )
    db.commit()

    holder, llm = _run_call(db, "ls", {"path": "Homelab"})

    assert llm.requests[1][0][3]["content"] == (
        "Homelab — 1 documents, 2 folders:\n"
        "\n"
        "  backups/ — 2 documents: Backup notes.\n"
        "  networking/ — 1 documents\n"
        "\n"
        "source: Homelab | path: readme.md | title: Readme | date: 2024-06-15"
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []

    # A registered source with no documents: the header line alone.
    holder0, llm0 = _run_call(db, "ls", {"path": "Deployments"})
    assert llm0.requests[1][0][3]["content"] == "Deployments — 0 documents, 0 folders:"
    assert holder0.tool_calls == 1


def test_ls_nested_folder_scope_drills_one_level_through_run_agent(
    kb, registry, db
) -> None:
    """A ``source/folder`` path: that folder's subfolders + own file
    lines (identity = ``source/folder``) — the drill-down against the
    real tables."""
    _doc(db, "Homelab", "networking/lan/a.md", "A", "A")
    _doc(db, "Homelab", "networking/lan/b.md", "B", "B")
    _doc(db, "Homelab", "networking/vpn/c.md", "C", "C")
    db.commit()

    holder, llm = _run_call(db, "ls", {"path": "Homelab/networking"})

    assert llm.requests[1][0][3]["content"] == (
        "Homelab/networking — 0 documents, 2 folders:\n"
        "\n"
        "  networking/lan/ — 2 documents\n"
        "  networking/vpn/ — 1 documents"
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []

    # One level deeper.
    holder2, llm2 = _run_call(db, "ls", {"path": "Homelab/networking/lan"})
    assert llm2.requests[1][0][3]["content"] == (
        "Homelab/networking/lan — 2 documents, 0 folders:\n"
        "\n"
        "source: Homelab | path: networking/lan/a.md | title: A | date: 2024-06-15\n"
        "source: Homelab | path: networking/lan/b.md | title: B | date: 2024-06-15"
    )
    assert holder2.tool_calls == 1


def test_ls_folder_file_cap_through_run_agent(kb, registry, db) -> None:
    """The cap end-to-end: 51 direct files in one folder cost 50 file
    lines + the deterministic grep-pointer note, never 51."""
    for i in range(51):
        _doc(db, "Homelab", f"big/f{i:03d}.md", f"T{i}", "BODY")
    db.commit()

    holder, llm = _run_call(db, "ls", {"path": "Homelab/big"})

    content = llm.requests[1][0][3]["content"]
    lines = content.splitlines()
    assert lines[0] == "Homelab/big — 51 documents, 0 folders:"
    assert lines[2] == (
        "source: Homelab | path: big/f000.md | title: T0 | date: 2024-06-15"
    )
    assert lines[51] == (
        "source: Homelab | path: big/f049.md | title: T49 | date: 2024-06-15"
    )
    assert lines[52] == (
        "…and 1 more documents in this folder — use grep (pattern) to "
        "find a specific one."
    )
    assert len(lines) == 53
    assert holder.tool_calls == 1


def test_ls_scoped_unknown_source_refused_through_run_agent(kb, src, db) -> None:
    """A ``path`` without ``/`` matching no source name is a refusal —
    the extended line with the teaching parenthetical (phase 72, the
    prefix byte-identical to the pre-phase-72 line); still not counted."""
    _doc(db, "Homelab", "a.md", "A", "A-CONTENT")
    db.commit()

    holder, llm = _run_call(db, "ls", {"path": "Ghost"})

    assert (
        llm.requests[1][0][3]["content"]
        == agent.NO_SOURCE_NOT_A_DIRECTORY.format(scope="Ghost")
    )
    assert holder.tool_calls == 0 and holder.read_docs == []


def test_ls_path_like_scope_unknown_source_gets_no_source_refusal(kb, src, db) -> None:
    """Phase 94: a ``/`` now names a folder — the phase-72 document-path
    teaching is DELETED; a ``source/…`` argument whose FIRST segment
    names no registered source gets the no-source refusal (the segment
    echoed), not counted, the tools stay offered."""
    _doc(db, "Homelab", "a.md", "A", "A-CONTENT")
    db.commit()

    holder, llm = _run_call(db, "ls", {"path": "app/rag/importer.py"})

    assert (
        llm.requests[1][0][3]["content"]
        == agent.NO_SOURCE_NOT_A_DIRECTORY.format(scope="app")
    )
    assert holder.tool_calls == 0 and holder.read_docs == []
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


def test_ls_unknown_folder_gets_not_a_folder_with_parents_subfolders(
    kb, src, db,
) -> None:
    """Phase 94: a folder segment matching no indexed prefix gets the
    NOT-A_FOLDER teaching — the argument echoed, the source named, its
    DIRECT subfolders listed (the self-correction list), not counted,
    the tools stay offered."""
    _doc(db, "Homelab", "backups/cron.md", "Cron", "CRON")
    _doc(db, "Homelab", "containers/caddy.md", "Caddy", "CADDY")
    _doc(db, "Homelab", "networking/lan.md", "LAN", "LAN")
    _doc(db, "Homelab", "readme.md", "Readme", "README")
    db.commit()

    holder, llm = _run_call(db, "ls", {"path": "Homelab/netwoking"})

    assert llm.requests[1][0][3]["content"] == (
        "'Homelab/netwoking' is not a folder — Homelab has: "
        "backups/ containers/ networking/"
    )
    assert holder.tool_calls == 0 and holder.read_docs == []
    assert llm.requests[1][1] == AGENT_TOOLS  # rejected → tools stay offered


# ---------- read (the canonical combined source/path form) ----------


def test_read_combined_path_through_run_agent(kb, db) -> None:
    """The combined ``source/path`` identity resolves at the FIRST slash
    against the REAL table (a path with further slashes included): the
    read executes, the holder records the row, the result header carries
    the true source/path."""
    created = _doc(db, "Alpha", "deep/nested/doc.md", "The Doc", "FULL-TEXT")
    db.commit()

    holder, llm = _run_call(db, "read", {"path": "Alpha/deep/nested/doc.md"})

    # Phase 106 (D5): the date rides every read — the SECOND line (the
    # first line stays the byte-identical header).
    assert llm.requests[1][0][3]["content"] == (
        "Document Alpha/deep/nested/doc.md:\n"
        "date: 2024-06-15\n"
        "FULL-TEXT"
    )
    assert holder.tool_calls == 1
    # SEC-14-04: short-lived session loads fresh copies
    assert [d.id for d in holder.read_docs] == [created.id]


def test_read_bare_source_name_refused_through_run_agent(kb, db) -> None:
    """A bare source name (no '/') can never be a document — the
    no-document refusal echoing the argument as passed; the old
    split-teaching refusal is gone (phase 70)."""
    _doc(db, "Alpha", "deep/nested/doc.md", "The Doc", "FULL-TEXT")
    db.commit()

    holder, llm = _run_call(db, "read", {"path": "Alpha"})

    assert (
        llm.requests[1][0][3]["content"] == "No document at 'Alpha' — check the ls output."
    )
    assert holder.tool_calls == 0 and holder.read_docs == []


def test_read_unknown_combined_path_refused_through_run_agent(kb, db) -> None:
    """A combined identity that matches NOTHING — not a document and not
    any indexed document's ``path`` (zero candidates) — gets today's
    no-document refusal byte-identical (the argument echoed as passed —
    the model sees its own form)."""
    _doc(db, "Alpha", "x.md", "X", "X-CONTENT")
    db.commit()

    holder, llm = _run_call(db, "read", {"path": "Alpha/nope/deep.md"})

    assert (
        llm.requests[1][0][3]["content"]
        == "No document at 'Alpha/nope/deep.md' — check the ls output."
    )
    assert holder.tool_calls == 0 and holder.read_docs == []


def test_read_bare_path_single_source_suggestion_then_corrected_read(kb, db) -> None:
    """Phase 72, task 02: a bare path under ONE source (exact ``path``
    match, the source prefix missing) → the single-identity suggestion
    (a refusal — not counted); the scripted corrected call (round 2, the
    suggested combined identity) then succeeds against real Postgres —
    the two-round self-correction."""
    created = _doc(db, "Alpha", "deep/nested/doc.md", "The Doc", "FULL-TEXT")
    db.commit()

    llm = ScriptedToolCallsLLM(
        [
            ToolCallPiece(
                id="call_1", name="read", arguments={"path": "deep/nested/doc.md"}
            ),
            ToolCallPiece(
                id="call_2",
                name="read",
                arguments={"path": "Alpha/deep/nested/doc.md"},
            ),
        ]
    )
    holder = AgentHolder()
    asyncio.run(_consume(cast("LLMClient", llm), lambda: SessionLocal(), holder))

    # Round 1: the bare path resolves to no combined identity, but it IS
    # the indexed document's path — the refusal names the one combined
    # identity to use (not counted, the tools stay offered).
    assert llm.requests[1][0][3]["content"] == (
        "No document at 'deep/nested/doc.md' — "
        "did you mean 'Alpha/deep/nested/doc.md'?"
    )
    assert llm.requests[1][1] == AGENT_TOOLS
    # Round 2: the corrected combined identity succeeds — the full
    # content (plus the phase-106 D5 date line), the holder records the
    # row, and it counts.
    assert llm.requests[2][0][5]["content"] == (
        "Document Alpha/deep/nested/doc.md:\n"
        "date: 2024-06-15\n"
        "FULL-TEXT"
    )
    assert llm.requests[2][1] == AGENT_TOOLS
    # SEC-14-04: short-lived session loads fresh copies
    assert [d.id for d in holder.read_docs] == [created.id]
    assert holder.tool_calls == 1  # only the corrected read executed


def test_read_bare_path_two_sources_one_of_suggestion_then_corrected_read(
    kb, db,
) -> None:
    """Phase 72, task 02: the same bare path under TWO sources → the
    ``one of`` line (up to ``SUGGESTION_LIMIT`` identities, catalog
    order — Alpha before Beta); the scripted corrected call (round 2,
    the first suggested identity) then succeeds."""
    a = _doc(db, "Alpha", "shared/x.md", "Alpha X", "A-TEXT")
    _doc(db, "Beta", "shared/x.md", "Beta X", "B-TEXT")
    db.commit()

    llm = ScriptedToolCallsLLM(
        [
            ToolCallPiece(id="call_1", name="read", arguments={"path": "shared/x.md"}),
            ToolCallPiece(
                id="call_2", name="read", arguments={"path": "Alpha/shared/x.md"}
            ),
        ]
    )
    holder = AgentHolder()
    asyncio.run(_consume(cast("LLMClient", llm), lambda: SessionLocal(), holder))

    assert llm.requests[1][0][3]["content"] == (
        "No document at 'shared/x.md' — did you mean one of: "
        "'Alpha/shared/x.md', 'Beta/shared/x.md'?"
    )
    assert llm.requests[2][0][5]["content"] == (
        "Document Alpha/shared/x.md:\ndate: 2024-06-15\nA-TEXT"
    )
    # SEC-14-04: short-lived session loads fresh copies
    assert [d.id for d in holder.read_docs] == [a.id]
    assert holder.tool_calls == 1  # only the corrected read executed


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


def test_all_documents_orders_by_source_then_path(kb, db) -> None:
    _doc(db, "Zeta", "b/second.md", "Zeta B", "ZB")
    _doc(db, "Zeta", "a/first.md", "Zeta A", "ZA")
    _doc(db, "Alpha", "c/third.md", "Alpha C", "AC")
    db.commit()

    docs = agent.all_documents(db)
    assert [(d.source, d.path) for d in docs] == [
        ("Alpha", "c/third.md"),
        ("Zeta", "a/first.md"),
        ("Zeta", "b/second.md"),
    ]
    assert [d.content for d in docs] == ["AC", "ZA", "ZB"]  # full rows


def test_grep_whole_kb_through_run_agent(kb, db) -> None:
    _doc(db, "Beta", "b/two.md", "Two", "no hit\nNEEDLE in two\nlast")
    _doc(db, "Alpha", "a/one.md", "One", "first\nneedle in one\nthird")
    db.commit()

    holder, llm = _run_call(db, "grep", {"pattern": "needle"})

    # Offered: the first request carries AGENT_TOOLS (the 3-tool list).
    assert llm.requests[0][1] == AGENT_TOOLS
    # Executed against the real DB: catalog order, grep-style lines.
    assert llm.requests[1][0][3]["content"] == (
        "Alpha/a/one.md:2: needle in one\n"
        "Beta/b/two.md:2: NEEDLE in two"
    )
    assert holder.tool_calls == 1
    assert holder.read_docs == []  # locked A5: grep adds no context


def test_grep_scoped_through_run_agent(kb, db) -> None:
    _doc(db, "Alpha", "a/one.md", "One", "first\nNeedle here\nthird")
    _doc(db, "Beta", "b/two.md", "Two", "NEEDLE too")
    db.commit()

    holder, llm = _run_call(db, "grep", {"pattern": "needle", "path": "Alpha/a/one.md"})

    # Only the named document is searched — the other one's hit is absent.
    assert llm.requests[1][0][3]["content"] == "Alpha/a/one.md:2: Needle here"
    assert holder.tool_calls == 1
    assert holder.read_docs == []


def test_grep_scoped_missing_doc_refused_through_run_agent(kb, db) -> None:
    """A scoped ``grep`` miss that matches no indexed document's ``path``
    (zero candidates) keeps today's line byte-identical — a refusal,
    not counted."""
    _doc(db, "Alpha", "a/one.md", "One", "nothing")
    db.commit()

    holder, llm = _run_call(db, "grep", {"pattern": "needle", "path": "Alpha/ghost.md"})

    assert (
        llm.requests[1][0][3]["content"]
        == "No document at 'Alpha/ghost.md' — check the ls output."
    )
    assert holder.tool_calls == 0 and holder.read_docs == []


def test_grep_no_matches_through_run_agent(kb, db) -> None:
    _doc(db, "Alpha", "a/one.md", "One", "nothing matching")
    db.commit()

    holder, llm = _run_call(db, "grep", {"pattern": "zebra"})

    assert llm.requests[1][0][3]["content"] == (
        "No matches for 'zebra' in the knowledge base."
    )
    assert holder.tool_calls == 1  # an executed grep with zero hits
    assert holder.read_docs == []