loader Public

Watch 0 Fork 0 Star 0
Python · 9589 bytes Raw Blame History
1 """Tests for the public runtime launcher seam."""
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, ReasoningConfig
10 from loader.llm.base import CompletionResponse, StreamChunk
11 from loader.runtime.bootstrap import RuntimeBootstrapView
12 from loader.runtime.launcher import RuntimeLauncher, build_runtime_launcher
13 from loader.runtime.public_shell import get_runtime_shell_system_message
14 from loader.runtime.runtime_handle import RuntimeHandle
15 from loader.utils.todos import active_todo_store_path
16 from tests.helpers.runtime_harness import ScriptedBackend
17
18
19 def test_build_runtime_launcher_returns_launcher_for_agent_source(
20 temp_dir: Path,
21 ) -> None:
22 agent = Agent(
23 backend=ScriptedBackend(),
24 config=AgentConfig(auto_context=False, stream=False),
25 project_root=temp_dir,
26 )
27
28 launcher = build_runtime_launcher(agent)
29
30 assert isinstance(launcher, RuntimeLauncher)
31 assert isinstance(launcher.source, RuntimeBootstrapView)
32 assert launcher.source is not agent
33 assert launcher.source.metadata == {
34 "owner_type": "Agent",
35 "owner_path": "public-agent",
36 }
37
38
39 @pytest.mark.asyncio
40 async def test_runtime_launcher_runs_conversation_turn(
41 temp_dir: Path,
42 ) -> None:
43 handle = RuntimeHandle(
44 backend=ScriptedBackend(
45 completions=[CompletionResponse(content="Hello back.")]
46 ),
47 config=AgentConfig(auto_context=False, stream=False),
48 project_root=temp_dir,
49 )
50 launcher = build_runtime_launcher(handle)
51 events = []
52
53 async def emit(event) -> None:
54 events.append(event)
55
56 summary = await launcher.run_turn("Hello there", emit)
57
58 assert summary.final_response == "Hello back."
59 assert any(event.type == "response" for event in events)
60
61
62 @pytest.mark.asyncio
63 async def test_runtime_launcher_runs_explore_query(
64 temp_dir: Path,
65 ) -> None:
66 backend = ScriptedBackend(
67 completions=[
68 CompletionResponse(content="Quick repo summary.")
69 ]
70 )
71 handle = RuntimeHandle(
72 backend=backend,
73 config=AgentConfig(auto_context=False, stream=False),
74 project_root=temp_dir,
75 )
76 launcher = build_runtime_launcher(handle)
77 events = []
78
79 async def emit(event) -> None:
80 events.append(event)
81
82 summary = await launcher.run_explore("Give me a quick repo summary.", emit)
83
84 assert summary.workflow_mode == "explore"
85 assert summary.final_response == "Quick repo summary."
86 assert any(event.type == "response" for event in events)
87
88
89 @pytest.mark.asyncio
90 async def test_runtime_launcher_runs_decomposition_fallback_turn(
91 temp_dir: Path,
92 ) -> None:
93 backend = ScriptedBackend(
94 completions=[
95 CompletionResponse(
96 content=(
97 '{"subtasks": [{"id": "1", "description": "Ship the feature", '
98 '"verification": "Done"}]}'
99 )
100 ),
101 CompletionResponse(content="Feature shipped directly."),
102 ]
103 )
104 handle = RuntimeHandle(
105 backend=backend,
106 config=AgentConfig(
107 auto_context=False,
108 stream=False,
109 reasoning=ReasoningConfig(decomposition=True),
110 ),
111 project_root=temp_dir,
112 )
113 launcher = build_runtime_launcher(handle)
114 events = []
115
116 async def emit(event) -> None:
117 events.append(event)
118
119 response = await launcher.run_decomposed(
120 "Ship the feature",
121 emit,
122 requested_mode="execute",
123 original_task="Ship the feature",
124 )
125
126 assert response == "Feature shipped directly."
127 assert events[0].type == "thinking"
128 assert any(event.type == "response" for event in events)
129 assert not any(event.type == "decomposition" for event in events)
130 assert handle.session.messages[0].content == "Ship the feature"
131
132
133 @pytest.mark.asyncio
134 async def test_runtime_launcher_routes_user_message_to_conversational_fast_path(
135 temp_dir: Path,
136 ) -> None:
137 handle = RuntimeHandle(
138 backend=ScriptedBackend(
139 streams=[
140 [
141 StreamChunk(content="Quick ", is_done=False),
142 StreamChunk(
143 content="reply.",
144 full_content="Quick reply.",
145 is_done=True,
146 ),
147 ]
148 ]
149 ),
150 config=AgentConfig(auto_context=False),
151 project_root=temp_dir,
152 )
153 launcher = build_runtime_launcher(handle)
154 events = []
155
156 async def emit(event) -> None:
157 events.append(event)
158
159 response = await launcher.run_user_message("thanks", emit)
160
161 assert response == "Quick reply."
162 assert handle.current_task is None
163 assert any(event.type == "response" and event.content == "Quick reply." for event in events)
164
165
166 @pytest.mark.asyncio
167 async def test_runtime_launcher_routes_user_message_to_direct_runtime_turn(
168 temp_dir: Path,
169 ) -> None:
170 handle = RuntimeHandle(
171 backend=ScriptedBackend(
172 completions=[CompletionResponse(content="Feature shipped directly.")]
173 ),
174 config=AgentConfig(
175 auto_context=False,
176 stream=False,
177 reasoning=ReasoningConfig(completion_check=False),
178 ),
179 project_root=temp_dir,
180 )
181 launcher = build_runtime_launcher(handle)
182 events = []
183
184 async def emit(event) -> None:
185 events.append(event)
186
187 response = await launcher.run_user_message(
188 "Write a short release-note style summary of what Loader does well.",
189 emit,
190 use_plan=False,
191 )
192
193 assert response == "Feature shipped directly."
194 assert handle.current_task == "Write a short release-note style summary of what Loader does well."
195 assert handle.last_turn_summary is not None
196 assert handle.last_turn_summary.final_response == "Feature shipped directly."
197 assert (
198 handle.session.messages[0].content
199 == "Write a short release-note style summary of what Loader does well."
200 )
201 assert any(event.type == "response" for event in events)
202
203
204 @pytest.mark.asyncio
205 async def test_runtime_launcher_routes_user_message_through_decomposition_lane(
206 temp_dir: Path,
207 monkeypatch,
208 ) -> None:
209 handle = RuntimeHandle(
210 backend=ScriptedBackend(),
211 config=AgentConfig(
212 auto_context=False,
213 stream=False,
214 reasoning=ReasoningConfig(decomposition=True, completion_check=False),
215 ),
216 project_root=temp_dir,
217 )
218 launcher = build_runtime_launcher(handle)
219 events = []
220 calls = []
221
222 async def emit(event) -> None:
223 events.append(event)
224
225 async def fake_run_decomposed(
226 task: str,
227 emit,
228 *,
229 on_confirmation=None,
230 on_user_question=None,
231 requested_mode: str | None = None,
232 original_task: str | None = None,
233 ) -> str:
234 calls.append(
235 {
236 "task": task,
237 "requested_mode": requested_mode,
238 "original_task": original_task,
239 }
240 )
241 return "All done."
242
243 monkeypatch.setattr(launcher, "run_decomposed", fake_run_decomposed)
244
245 response = await launcher.run_user_message(
246 "Read the spec and implement the feature",
247 emit,
248 use_plan=False,
249 )
250
251 assert response == "All done."
252 assert handle.current_task == "Read the spec and implement the feature"
253 assert calls == [
254 {
255 "task": "Read the spec and implement the feature",
256 "requested_mode": "execute",
257 "original_task": "Read the spec and implement the feature",
258 }
259 ]
260 assert events == []
261
262
263 @pytest.mark.asyncio
264 async def test_runtime_launcher_resets_task_scoped_state_for_new_top_level_prompt(
265 temp_dir: Path,
266 ) -> None:
267 backend = ScriptedBackend(
268 completions=[CompletionResponse(content="Penguins page shipped.")]
269 )
270 handle = RuntimeHandle(
271 backend=backend,
272 config=AgentConfig(
273 auto_context=False,
274 stream=False,
275 reasoning=ReasoningConfig(completion_check=False),
276 ),
277 project_root=temp_dir,
278 )
279 handle.current_task = "Create a collection of animal pages."
280 cached_prompt = get_runtime_shell_system_message(handle)
281 assert "Create a collection of animal pages." in cached_prompt.content
282
283 todo_store = active_todo_store_path(temp_dir)
284 todo_store.parent.mkdir(parents=True, exist_ok=True)
285 todo_store.write_text(
286 '[{"content": "Build cat page", "active_form": "Building cat page", "status": "pending"}]'
287 )
288
289 launcher = build_runtime_launcher(handle)
290 events = []
291
292 async def emit(event) -> None:
293 events.append(event)
294
295 response = await launcher.run_user_message(
296 "Generate penguins.html and penguins.css for the new page.",
297 emit,
298 use_plan=False,
299 )
300
301 assert response == "Penguins page shipped."
302 assert handle.current_task == "Generate penguins.html and penguins.css for the new page."
303 assert not todo_store.exists()
304 assert any(
305 event.type == "todo_update" and event.todo_items == []
306 for event in events
307 )
308 invocation = backend.invocations[-1]
309 assert (
310 "Current task: Generate penguins.html and penguins.css for the new page."
311 in invocation.messages[0].content
312 )
313 assert "Current task: Create a collection of animal pages." not in invocation.messages[0].content