"""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 SINGLE-READ flow (``TOOLS_TRIGGER`` only) stays byte-identical: list → read (first catalog line, ``call_1``) → answer; * the phase-45 MULTI-READ flow (``TOOLS_TRIGGER`` + ``MULTI_READ_TRIGGER``) classifies by the count of ``tool``-role read results: list → read #1 (``call_1``) → read #2 (second catalog line, ``call_2``) → the byte-stable ``multi_answer`` naming both read paths. """ 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 ``ls`` output for a two-document KB #: (``app/rag/agent.py`` ``_execute_tool``): one #: ``source: X | path: Y | title: Z`` line per document (phase 63: labeled, #: unambiguous fields), ``(source, path)`` order. CATALOG_2 = ( "2 documents:\n" "source: Deployments | path: example-record-file.json | title: Example Record File\n" "source: Homelab | path: aws-route53.md | title: AWS Route 53 Notes" ) CATALOG_1 = ( "1 documents:\n" "source: Deployments | path: example-record-file.json | title: Example Record File" ) CATALOG_3 = ( "3 documents:\n" "source: Deployments | path: aaa.md | title: AAA\n" "source: Deployments | path: bbb.md | title: BBB\n" "source: Homelab | path: ccc.md | title: CCC" ) 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_read_step_first_catalog_line() -> None: flow = _tool_flow(_body(SINGLE_USER, (CATALOG_3,))) # The FIRST listing line (Deployments/aaa.md), labeled fields. assert flow == ("read", "Deployments", "aaa.md", "call_1") 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. catalog = ( "1 documents:\n" "source: brain-of-reese-main | path: homelab/aws-route53.md | title: aws-route53" ) flow = _tool_flow(_body(SINGLE_USER, (catalog,))) assert flow == ("read", "brain-of-reese-main", "homelab/aws-route53.md", "call_1") 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, (CATALOG_2, _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, (CATALOG_2, _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_read_first_step() -> None: flow = _tool_flow(_body(MULTI_USER, (CATALOG_2,))) assert flow == ("read", DOC1_SP.split("/", 1)[0], DOC1_SP.rsplit("/", 1)[1], "call_1") def test_multi_flow_read_second_step_skips_already_read() -> None: flow = _tool_flow(_body(MULTI_USER, (CATALOG_2, _read_result(DOC1_SP, DOC1_CONTENT)))) # The second catalog line — the first line differing from DOC1. assert flow == ("read", "Homelab", "aws-route53.md", "call_2") def test_multi_flow_read_second_is_listing_order_not_last() -> None: # Three-doc catalog, first doc read: read #2 is the SECOND line # (Deployments/bbb.md), not the last one. flow = _tool_flow(_body(MULTI_USER, (CATALOG_3, _read_result("Deployments/aaa.md", "x")))) assert flow == ("read", "Deployments", "bbb.md", "call_2") def test_multi_flow_answer_step_names_both_paths() -> None: flow = _tool_flow( _body( MULTI_USER, ( CATALOG_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, ( CATALOG_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_document_catalog_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, (CATALOG_1, _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: # A catalog (labeled lines) and a read result ("Document …" prefix) # are NOT search results — the flow stays at the search step. flow = _search_flow(_body(SEARCH_USER, (CATALOG_2, _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