loader Public

Watch 0 Fork 0 Star 0
Python · 5598 bytes Raw Blame History
1 """Tests for assistant-cycle iteration orchestration."""
2
3 from __future__ import annotations
4
5 from pathlib import Path
6
7 import pytest
8
9 from loader.agent.loop import Agent, AgentConfig
10 from loader.llm.base import CompletionResponse, Message, Role, ToolCall
11 from loader.runtime.conversation import ConversationRuntime
12 from loader.runtime.turn_iteration import (
13 TurnIterationAction,
14 _successful_progress_after_latest_empty_retry,
15 )
16 from tests.helpers.runtime_harness import ScriptedBackend
17
18
19 def non_streaming_config() -> AgentConfig:
20 """Shared config for direct turn-iteration tests."""
21
22 return AgentConfig(auto_context=False, stream=False, max_iterations=8)
23
24
25 async def _run_iteration(
26 runtime: ConversationRuntime,
27 *,
28 task: str,
29 requested_mode: str = "execute",
30 original_task: str | None = None,
31 ) -> tuple:
32 events = []
33
34 async def capture(event) -> None:
35 events.append(event)
36
37 prepared = await runtime.turn_preparation.prepare(
38 task=task,
39 emit=capture,
40 requested_mode=requested_mode,
41 original_task=original_task,
42 on_user_question=None,
43 )
44 decision = await runtime.turn_iteration.run_iteration(
45 task=prepared.task,
46 effective_task=prepared.effective_task,
47 original_task=original_task,
48 effective_max_tokens=prepared.effective_max_tokens,
49 iterations=1,
50 max_iterations=runtime.context.config.max_iterations,
51 actions_taken=[],
52 continuation_count=0,
53 empty_retry_count=0,
54 max_empty_retries=5,
55 extracted_iterations=0,
56 max_extracted_iterations=3,
57 consecutive_errors=0,
58 dod=prepared.definition_of_done,
59 emit=capture,
60 summary=prepared.summary,
61 executor=prepared.executor,
62 rollback_plan=prepared.rollback_plan,
63 on_confirmation=None,
64 on_user_question=None,
65 emit_confirmation=runtime._emit_confirmation(capture),
66 )
67 return prepared, decision, events
68
69
70 @pytest.mark.asyncio
71 async def test_turn_iteration_completes_react_final_answer(
72 temp_dir: Path,
73 ) -> None:
74 backend = ScriptedBackend(
75 completions=[
76 CompletionResponse(
77 content="Thought: I have enough context.\nFinal Answer: All set.",
78 )
79 ],
80 supports_native_tools=False,
81 )
82 agent = Agent(
83 backend=backend,
84 config=non_streaming_config(),
85 project_root=temp_dir,
86 )
87 runtime = ConversationRuntime(agent)
88
89 prepared, decision, events = await _run_iteration(
90 runtime,
91 task="Explain whether the turn iteration controller works.",
92 )
93
94 assert agent.use_react
95 assert decision.action == TurnIterationAction.COMPLETE
96 assert prepared.summary.final_response == "All set."
97 assert prepared.summary.assistant_messages[-1].content.endswith("Final Answer: All set.")
98 assert any(event.type == "response" and event.content == "All set." for event in events)
99
100
101 @pytest.mark.asyncio
102 async def test_turn_iteration_executes_native_tool_batch_and_continues(
103 temp_dir: Path,
104 ) -> None:
105 readme = temp_dir / "README.md"
106 readme.write_text("Loader runtime notes\n")
107
108 backend = ScriptedBackend(
109 completions=[
110 CompletionResponse(
111 content="I'll inspect the README first.",
112 tool_calls=[
113 ToolCall(
114 id="read-1",
115 name="read",
116 arguments={"file_path": str(readme)},
117 )
118 ],
119 )
120 ]
121 )
122 agent = Agent(
123 backend=backend,
124 config=non_streaming_config(),
125 project_root=temp_dir,
126 )
127 runtime = ConversationRuntime(agent)
128
129 prepared, decision, events = await _run_iteration(
130 runtime,
131 task="Inspect the README before continuing.",
132 )
133
134 assert decision.action == TurnIterationAction.CONTINUE
135 assert decision.new_actions_taken == [f"read: {{'file_path': '{readme}'}}"]
136 assert prepared.summary.assistant_messages[-1].tool_calls[0].name == "read"
137 assert len(prepared.summary.tool_result_messages) == 1
138 assert "Loader runtime notes" in prepared.summary.tool_result_messages[0].content
139 assert any(event.type == "tool_call" and event.tool_name == "read" for event in events)
140 assert any(
141 event.type == "tool_result" and "Loader runtime notes" in event.content
142 for event in events
143 )
144
145
146 def test_empty_retry_episode_resets_after_successful_tool_progress() -> None:
147 messages = [
148 Message(role=Role.USER, content="[EMPTY ASSISTANT RESPONSE] retry 1/6"),
149 Message.tool_result_message(
150 tool_call_id="write-1",
151 display_content="Observation [write]: Result: Successfully wrote file",
152 result_content="Observation [write]: Result: Successfully wrote file",
153 is_error=False,
154 ),
155 ]
156
157 assert _successful_progress_after_latest_empty_retry(messages) is True
158
159
160 def test_empty_retry_episode_does_not_reset_after_blocked_tool_result() -> None:
161 messages = [
162 Message(role=Role.USER, content="[EMPTY ASSISTANT RESPONSE] retry 1/6"),
163 Message.tool_result_message(
164 tool_call_id="write-1",
165 display_content="[Blocked - HTML content contains placeholder or stub text]",
166 result_content="[Blocked - HTML content contains placeholder or stub text]",
167 is_error=True,
168 ),
169 ]
170
171 assert _successful_progress_after_latest_empty_retry(messages) is False