brain-of-reese/tests/unit/test_mock_tool_flow.py

"""Unit tests for the E2E mock's tool-flow classifier (phase 45, task 02).

The mock (``tests/e2e/mock_llm.py``) classifies marker requests
statelessly into one step of the agent tool flow. This file pins the
classification at unit speed — no Playwright, no LLM process:

* the phase-37/94 SINGLE-READ flow (``TOOLS_TRIGGER`` only): list (the
  top-level source listing) → drill (``ls`` scoped to the first source
  — phase 94: the top level carries sources only, so the flow drills
  one level for the file lines) → read (first file line, the combined
  ``source/path``) → answer;
* the phase-45 MULTI-READ flow (``TOOLS_TRIGGER`` +
  ``MULTI_READ_TRIGGER``) classifies by the count of ``tool``-role read
  results: list → drill → read #1 (first file line) → read #2 (second
  file line) → the byte-stable ``multi_answer`` naming both read paths;
* the degenerate empty-KB case: every listed source already drilled
  with no file lines → the flow falls back to the re-list loop (the
  round cap settles it — the phase-70 empty-catalog behavior).
"""
from __future__ import annotations

from typing import Any

from tests.e2e.mock_llm import (
    MULTI_READ_TRIGGER,
    SEARCH_PATTERN,
    SEARCH_TRIGGER,
    TOOLS_TRIGGER,
    _search_flow,
    _tool_flow,
)

# --------------------------------------------------------------------------
# Wire fixtures — byte-identical to what app/rag/agent.py produces
# --------------------------------------------------------------------------

#: The ``<tools>`` section marks the HIGH prompt (app/rag/prompts.py).
SYSTEM_HIGH = "<relevance>HIGH</relevance>\n<documents>\n</documents>\n<tools>\n…\n</tools>"
SYSTEM_LOW = "<relevance>LOW</relevance>\n"

#: A minimal truthy ``tools`` parameter (the mock only checks presence;
#: the phase-70 harness-aligned names).
TOOLS = [{"type": "function", "function": {"name": "ls"}}]

#: The agent's drill-down ``ls`` output for a two-source KB
#: (``app/rag/agent.py`` ``render_ls_top`` / ``render_folder_listing``,
#: phase 94): the top level lists the registered sources (registry
#: order, recursive counts — no file lines); the folder level carries
#: the file lines (``source: X | path: Y | title: Z`` — the phase-63
#: labeled fields, unchanged), ``path`` order.
TOP_LEVEL_2 = (
    "2 sources:\n"
    "\n"
    "Deployments — 1 documents\n"
    "Homelab — 1 documents"
)

#: The first source's root folder: one file line (the single-read
#: flow's read target).
FOLDER_DEPLOYMENTS = (
    "Deployments — 1 documents, 0 folders:\n"
    "\n"
    "source: Deployments | path: example-record-file.json | title: Example Record File"
)

#: Two file lines in the first source (the multi-read flow's reads).
FOLDER_DEPLOYMENTS_2 = (
    "Deployments — 2 documents, 0 folders:\n"
    "\n"
    "source: Deployments | path: aaa.md | title: AAA\n"
    "source: Deployments | path: bbb.md | title: BBB"
)

#: Three file lines in the first source (the listing-order pin: read
#: #2 is the SECOND line, not the last).
FOLDER_DEPLOYMENTS_3 = (
    "Deployments — 3 documents, 0 folders:\n"
    "\n"
    "source: Deployments | path: aaa.md | title: AAA\n"
    "source: Deployments | path: bbb.md | title: BBB\n"
    "source: Deployments | path: ccc.md | title: CCC"
)

#: Empty folder levels (a registered source with no documents — the
#: header line alone; the drill's degenerate arm).
FOLDER_DEPLOYMENTS_EMPTY = "Deployments — 0 documents, 0 folders:"
FOLDER_HOMELAB_EMPTY = "Homelab — 0 documents, 0 folders:"

DOC1_SP = "Deployments/example-record-file.json"
DOC1_CONTENT = (
    "The record file keeps every hosted zone record — first line is longer "
    "than eighty characters so the quote truncation below is observable.\n"
    "second line of the document content"
)
assert len(DOC1_CONTENT) > 80

DOC2_SP = "Homelab/aws-route53.md"
DOC2_CONTENT = "Route 53 notes — the second read, short on purpose."

SINGLE_USER = "Use your tools: what is the exact shape of the record file?"
#: Carries BOTH markers — ``use your tools`` then ``read two documents``.
MULTI_USER = "Use your tools and read two documents: compare the zone notes with the record file."
#: The multi marker alone — no ``use your tools``.
MULTI_ONLY_USER = "Please read two documents and compare them."
PLAIN_USER = "How does the sync job push records to the zone?"
assert TOOLS_TRIGGER in SINGLE_USER.lower() and MULTI_READ_TRIGGER not in SINGLE_USER.lower()
assert TOOLS_TRIGGER in MULTI_USER.lower() and MULTI_READ_TRIGGER in MULTI_USER.lower()


def _read_result(sp: str, content: str) -> str:
    """The agent's read-result text (``_execute_tool`` prefix)."""
    return f"Document {sp}:\n{content}"


def _body(
    user: str,
    tool_msgs: tuple[str, ...] = (),
    tools: Any = TOOLS,
    system: str = SYSTEM_HIGH,
) -> dict[str, Any]:
    """A chat-completion body: system + user + the tool results in order."""
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    for i, content in enumerate(tool_msgs):
        messages.append(
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": f"call_{i}",
                        "type": "function",
                        "function": {"name": "ls", "arguments": "{}"},
                    }
                ],
            }
        )
        messages.append({"role": "tool", "tool_call_id": f"call_{i}", "content": content})
    return {"messages": messages, "tools": tools}


# --------------------------------------------------------------------------
# Phase-37 single-read flow — must stay byte-identical
# --------------------------------------------------------------------------


def test_single_flow_list_step() -> None:
    assert _tool_flow(_body(SINGLE_USER)) == ("list", "", "")


def test_single_flow_drill_step_after_top_level() -> None:
    # Phase 94: the top level lists SOURCES only — the flow drills one
    # level into the FIRST source (listing order = registry order).
    flow = _tool_flow(_body(SINGLE_USER, (TOP_LEVEL_2,)))
    assert flow == ("drill", "Deployments", "call_1")


def test_single_flow_drill_skips_already_drilled_source() -> None:
    # The first source's folder level is already in the messages (an
    # empty listing — header only, no file lines): the drill proceeds
    # to the NEXT un-drilled source.
    flow = _tool_flow(_body(SINGLE_USER, (TOP_LEVEL_2, FOLDER_DEPLOYMENTS_EMPTY)))
    assert flow == ("drill", "Homelab", "call_1")


def test_single_flow_all_sources_drilled_empty_falls_back_to_list() -> None:
    # Degenerate: every listed source already drilled, no file lines
    # anywhere — the flow falls back to the re-list loop (settled at the
    # round cap, the phase-70 empty-catalog behavior).
    flow = _tool_flow(
        _body(
            SINGLE_USER,
            (TOP_LEVEL_2, FOLDER_DEPLOYMENTS_EMPTY, FOLDER_HOMELAB_EMPTY),
        )
    )
    assert flow == ("list", "", "")


def test_single_flow_read_step_first_file_line() -> None:
    # The folder level reached: the FIRST file line (Deployments/aaa.md),
    # labeled fields, the combined ``source/path`` join.
    flow = _tool_flow(_body(SINGLE_USER, (TOP_LEVEL_2, FOLDER_DEPLOYMENTS_2)))
    assert flow == ("read", "Deployments", "aaa.md", "call_2")


def test_read_step_nested_path_stays_intact() -> None:
    # Phase 63 bug report: the path itself contains ``/`` — the old
    # ``source/path — title`` + ``rpartition("/")`` parse misread the
    # split (``source=brain-of-reese-main/homelab``). The labeled fields
    # recover the nested path intact, however deep.
    listing = (
        "brain-of-reese-main — 1 documents, 0 folders:\n"
        "\n"
        "source: brain-of-reese-main | path: homelab/aws-route53.md | title: aws-route53"
    )
    flow = _tool_flow(_body(SINGLE_USER, (TOP_LEVEL_2, listing)))
    assert flow == ("read", "brain-of-reese-main", "homelab/aws-route53.md", "call_2")


def test_single_flow_answer_step_with_tools_offered() -> None:
    # Phase 45: the round cap keeps the tools offered until it is hit —
    # the answer step fires regardless of the ``tools`` parameter.
    flow = _tool_flow(
        _body(SINGLE_USER, (TOP_LEVEL_2, FOLDER_DEPLOYMENTS, _read_result(DOC1_SP, DOC1_CONTENT)))
    )
    assert flow == ("answer", DOC1_SP, DOC1_CONTENT)


def test_single_flow_answer_step_without_tools() -> None:
    flow = _tool_flow(
        _body(
            SINGLE_USER,
            (TOP_LEVEL_2, FOLDER_DEPLOYMENTS, _read_result(DOC1_SP, DOC1_CONTENT)),
            tools=None,
        )
    )
    assert flow == ("answer", DOC1_SP, DOC1_CONTENT)


def test_single_flow_no_tools_no_results_is_not_the_flow() -> None:
    # agent_max_rounds=0 path: marker + <tools> prompt, but the request
    # carries no tools and no tool results — regular answer, not a flow.
    assert _tool_flow(_body(SINGLE_USER, tools=None)) is None


def test_single_flow_marker_without_tools_section_is_none() -> None:
    assert _tool_flow(_body(SINGLE_USER, system=SYSTEM_LOW)) is None


def test_single_flow_plain_question_is_none() -> None:
    assert _tool_flow(_body(PLAIN_USER)) is None


# --------------------------------------------------------------------------
# Phase-45 multi-read flow (task 02)
# --------------------------------------------------------------------------


def test_multi_flow_list_step() -> None:
    assert _tool_flow(_body(MULTI_USER)) == ("list", "", "")


def test_multi_flow_drill_step_after_top_level() -> None:
    # Phase 94: the top level lists SOURCES only — the multi flow drills
    # too, before its first read.
    flow = _tool_flow(_body(MULTI_USER, (TOP_LEVEL_2,)))
    assert flow == ("drill", "Deployments", "call_1")


def test_multi_flow_read_first_step() -> None:
    flow = _tool_flow(_body(MULTI_USER, (TOP_LEVEL_2, FOLDER_DEPLOYMENTS)))
    assert flow == ("read", DOC1_SP.split("/", 1)[0], DOC1_SP.rsplit("/", 1)[1], "call_2")


def test_multi_flow_read_second_step_skips_already_read() -> None:
    flow = _tool_flow(
        _body(
            MULTI_USER,
            (TOP_LEVEL_2, FOLDER_DEPLOYMENTS_2, _read_result("Deployments/aaa.md", DOC1_CONTENT)),
        )
    )
    # The second file line — the first line differing from the read doc.
    assert flow == ("read", "Deployments", "bbb.md", "call_3")


def test_multi_flow_read_second_is_listing_order_not_last() -> None:
    # Three-file listing, first file read: read #2 is the SECOND line
    # (Deployments/bbb.md), not the last one.
    flow = _tool_flow(
        _body(
            MULTI_USER,
            (TOP_LEVEL_2, FOLDER_DEPLOYMENTS_3, _read_result("Deployments/aaa.md", "x")),
        )
    )
    assert flow == ("read", "Deployments", "bbb.md", "call_3")


def test_multi_flow_answer_step_names_both_paths() -> None:
    flow = _tool_flow(
        _body(
            MULTI_USER,
            (
                TOP_LEVEL_2,
                FOLDER_DEPLOYMENTS_2,
                _read_result(DOC1_SP, DOC1_CONTENT),
                _read_result(DOC2_SP, DOC2_CONTENT),
            ),
        )
    )
    assert flow is not None
    assert flow[0] == "multi_answer"
    # Byte-stable: the single-read shape quoting the FIRST read result
    # (first 80 chars), plus both read paths in read order.
    assert flow[2] == f"Read {DOC1_SP}. {DOC1_CONTENT[:80]} I read {DOC1_SP} and {DOC2_SP}."


def test_multi_flow_answer_step_without_tools_offered() -> None:
    # The forced answer is content, not a tool call — it must not be
    # gated on the ``tools`` parameter.
    flow = _tool_flow(
        _body(
            MULTI_USER,
            (
                TOP_LEVEL_2,
                FOLDER_DEPLOYMENTS_2,
                _read_result(DOC1_SP, DOC1_CONTENT),
                _read_result(DOC2_SP, DOC2_CONTENT),
            ),
            tools=None,
        )
    )
    assert flow is not None
    assert flow[0] == "multi_answer"


def test_multi_flow_one_file_listing_degenerates_to_single_answer() -> None:
    # Nothing second to read — the single-read answer shape, quoting the
    # only read result.
    flow = _tool_flow(
        _body(MULTI_USER, (TOP_LEVEL_2, FOLDER_DEPLOYMENTS, _read_result(DOC1_SP, DOC1_CONTENT)))
    )
    assert flow == ("answer", DOC1_SP, DOC1_CONTENT)


def test_multi_flow_no_tools_no_results_is_not_the_flow() -> None:
    assert _tool_flow(_body(MULTI_USER, tools=None)) is None


def test_multi_trigger_without_tools_trigger_is_none() -> None:
    # The multi marker alone (no ``use your tools``) is not the flow.
    assert MULTI_READ_TRIGGER in MULTI_ONLY_USER
    assert TOOLS_TRIGGER not in MULTI_ONLY_USER.lower()
    assert _tool_flow(_body(MULTI_ONLY_USER)) is None


def test_multi_flow_requires_tools_section() -> None:
    assert _tool_flow(_body(MULTI_USER, system=SYSTEM_LOW)) is None


# --------------------------------------------------------------------------
# Phase-68 search flow (task 03)
# --------------------------------------------------------------------------

#: Carries ONLY the search trigger (never ``use your tools`` — the
#: phase-68 suite's live question shape, regression-safe by assertion).
SEARCH_USER = (
    "Search your documents for the vault passphrase marker in my homelab "
    "kubernetes backup notes?"
)
assert SEARCH_TRIGGER in SEARCH_USER.lower()
assert TOOLS_TRIGGER not in SEARCH_USER.lower()

#: The agent's ``grep`` result for the e2e fixture (phase 70 renamed
#: the phase-68 tool; the line format is unchanged)
#: (``app/rag/agent.py`` ``_execute_tool``): one ``source/path:LINE: text``
#: match line (the sentinel line, 200-char-capped server-side).
SEARCH_RESULT = (
    f"search_docs/reese-notes.md:6: The offsite vault passphrase marker "
    f"is {SEARCH_PATTERN}."
)
#: The agent's no-match line quotes the pattern — the sentinel-only
#: shape ``_search_result_line`` also recognizes (degenerate path).
SEARCH_NO_MATCH = f"No matches for '{SEARCH_PATTERN}' in the knowledge base."


def test_search_flow_search_step() -> None:
    # tools offered, no search result yet: the model greps.
    assert _search_flow(_body(SEARCH_USER)) == ("search",)


def test_search_flow_search_step_requires_tools_offered() -> None:
    # agent_max_rounds=0 path: trigger + <tools> prompt, but no tools
    # and no search result — regular answer, not a flow.
    assert _search_flow(_body(SEARCH_USER, tools=None)) is None


def test_search_flow_found_step_quotes_first_match_line() -> None:
    flow = _search_flow(_body(SEARCH_USER, (SEARCH_RESULT,)))
    assert flow == ("found", f"The offsite vault passphrase marker is {SEARCH_PATTERN}.")


def test_search_flow_found_step_with_nested_path() -> None:
    # A nested path (``/`` in it) stays intact in the match-line parse.
    result = f"search_docs/deep/nested-note.md:12: line with {SEARCH_PATTERN} inside"
    flow = _search_flow(_body(SEARCH_USER, (result,)))
    assert flow == ("found", f"line with {SEARCH_PATTERN} inside")


def test_search_flow_found_step_without_tools_offered() -> None:
    # The answer is content, not a tool call — it must not be gated on
    # the ``tools`` parameter (phase 45 keeps the tools offered until
    # the round cap, but the no-tools final request must still answer).
    flow = _search_flow(_body(SEARCH_USER, (SEARCH_RESULT,), tools=None))
    assert flow == ("found", f"The offsite vault passphrase marker is {SEARCH_PATTERN}.")


def test_search_flow_ignores_catalog_and_read_results() -> None:
    # Listings (top-level + folder level, labeled file lines) and a read
    # result ("Document …" prefix) are NOT search results — the flow
    # stays at the search step.
    flow = _search_flow(
        _body(SEARCH_USER, (TOP_LEVEL_2, FOLDER_DEPLOYMENTS, _read_result(DOC1_SP, DOC1_CONTENT)))
    )
    assert flow == ("search",)


def test_search_flow_sentinel_only_result_is_a_search_result() -> None:
    # The no-match line quotes the pattern — sentinel-only recognition
    # (degenerate path; the e2e fixture always matches).
    flow = _search_flow(_body(SEARCH_USER, (SEARCH_NO_MATCH,)))
    assert flow == ("found", SEARCH_NO_MATCH)


def test_search_flow_requires_tools_section() -> None:
    # Deflected turns never carry the <tools> section.
    assert _search_flow(_body(SEARCH_USER, system=SYSTEM_LOW)) is None


def test_search_flow_plain_question_is_none() -> None:
    assert _search_flow(_body(PLAIN_USER)) is None


def test_search_trigger_does_not_shadow_the_tool_flow() -> None:
    # The search question carries no ``use your tools`` — the phase-37
    # classifier must stay inert on it (regression-safe marker).
    assert _tool_flow(_body(SEARCH_USER)) is None


def test_tool_trigger_does_not_shadow_the_search_flow() -> None:
    # The phase-37/45 questions carry no ``search your documents`` —
    # the search classifier must stay inert on them.
    assert _search_flow(_body(SINGLE_USER)) is None
    assert _search_flow(_body(MULTI_USER)) is None