"""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 ```` section marks the HIGH prompt (app/rag/prompts.py). SYSTEM_HIGH = "HIGH\n\n\n\n…\n" SYSTEM_LOW = "LOW\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 + 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 + 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 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