"""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, TOOLS_TRIGGER, _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). TOOLS = [{"type": "function", "function": {"name": "list_documents"}}] #: The agent's ``list_documents`` output for a two-document KB # (``app/rag/agent.py`` ``_execute_tool``): one ``source/path — title`` #: line per document, ``(source, path)`` order. CATALOG_2 = ( "2 documents:\n" "Deployments/example-record-file.json — Example Record File\n" "Homelab/aws-route53.md — AWS Route 53 Notes" ) CATALOG_1 = "1 documents:\nDeployments/example-record-file.json — Example Record File" CATALOG_3 = ( "3 documents:\n" "Deployments/aaa.md — AAA\n" "Deployments/bbb.md — BBB\n" "Homelab/ccc.md — 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": "list_documents", "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), rsplit convention. assert flow == ("read", "Deployments", "aaa.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