loader Public

Watch 0 Fork 0 Star 0
Python · 5951 bytes Raw Blame History
1 """Tests for the runtime-first internal handle below Agent."""
2
3 from __future__ import annotations
4
5 from pathlib import Path
6
7 import pytest
8
9 from loader.agent.loop import AgentConfig
10 from loader.llm.base import CompletionResponse, StreamChunk
11 from loader.runtime.bootstrap import RuntimeBootstrapView, build_runtime_context
12 from loader.runtime.conversation import ConversationRuntime
13 from loader.runtime.launcher import RuntimeLauncher, build_runtime_launcher
14 from loader.runtime.runtime_handle import RuntimeHandle
15 from tests.helpers.runtime_harness import ScriptedBackend, run_explore_scenario, run_scenario
16
17
18 def test_runtime_handle_builds_runtime_bootstrap_contract(
19 temp_dir: Path,
20 ) -> None:
21 handle = RuntimeHandle(
22 backend=ScriptedBackend(),
23 config=AgentConfig(auto_context=False),
24 project_root=temp_dir,
25 )
26
27 launcher = build_runtime_launcher(handle)
28 context = build_runtime_context(handle)
29
30 assert isinstance(launcher, RuntimeLauncher)
31 assert isinstance(launcher.source, RuntimeBootstrapView)
32 assert launcher.source is not handle
33 assert launcher.source.metadata == {
34 "owner_type": "RuntimeHandle",
35 "owner_path": "runtime-handle",
36 }
37 assert context.project_root == temp_dir.resolve()
38 assert context.backend is handle.backend
39 assert context.registry is handle.registry
40 assert context.session is handle.session
41 assert context.permission_policy is handle.permission_policy
42 assert context.workflow_mode == handle.workflow_mode
43
44
45 @pytest.mark.asyncio
46 async def test_runtime_handle_runs_conversation_runtime_without_agent(
47 temp_dir: Path,
48 ) -> None:
49 handle = RuntimeHandle(
50 backend=ScriptedBackend(
51 completions=[CompletionResponse(content="Runtime handle reply.")]
52 ),
53 config=AgentConfig(auto_context=False, stream=False),
54 project_root=temp_dir,
55 )
56 runtime = ConversationRuntime(handle)
57 events = []
58
59 async def emit(event) -> None:
60 events.append(event)
61
62 summary = await runtime.run_turn(
63 "Explain why runtime-first seams matter here.",
64 emit,
65 requested_mode="execute",
66 )
67
68 assert summary.final_response == "Runtime handle reply."
69 assert runtime.source.metadata == {
70 "owner_type": "RuntimeHandle",
71 "owner_path": "runtime-handle",
72 }
73 assert any(event.type == "response" for event in events)
74
75
76 @pytest.mark.asyncio
77 async def test_runtime_handle_runs_public_shell_entrypoint_without_agent(
78 temp_dir: Path,
79 ) -> None:
80 handle = RuntimeHandle(
81 backend=ScriptedBackend(
82 completions=[CompletionResponse(content="Runtime handle shell reply.")]
83 ),
84 config=AgentConfig(auto_context=False, stream=False),
85 project_root=temp_dir,
86 )
87 events = []
88
89 async def emit(event) -> None:
90 events.append(event)
91
92 response = await handle.run(
93 "Summarize the runtime-first shell path.",
94 on_event=emit,
95 use_plan=False,
96 )
97
98 assert response == "Runtime handle shell reply."
99 assert handle.last_turn_summary is not None
100 assert handle.last_turn_summary.final_response == "Runtime handle shell reply."
101 assert any(event.type == "response" for event in events)
102
103
104 @pytest.mark.asyncio
105 async def test_runtime_handle_runs_explore_and_streaming_entrypoints_without_agent(
106 temp_dir: Path,
107 ) -> None:
108 handle = RuntimeHandle(
109 backend=ScriptedBackend(
110 completions=[CompletionResponse(content="Explore with runtime handle.")],
111 streams=[
112 [
113 StreamChunk(content="Streamed ", is_done=False),
114 StreamChunk(
115 content="runtime reply.",
116 full_content="Streamed runtime reply.",
117 is_done=True,
118 ),
119 ]
120 ],
121 ),
122 config=AgentConfig(auto_context=False),
123 project_root=temp_dir,
124 )
125
126 stream_events = [event async for event in handle.run_streaming("thanks")]
127 explore_response = await handle.run_explore(
128 "Where should I start in this repo?",
129 )
130
131 assert any(
132 event.type == "response" and event.content == "Streamed runtime reply."
133 for event in stream_events
134 )
135 assert explore_response == "Explore with runtime handle."
136 assert handle.last_turn_summary is not None
137 assert handle.last_turn_summary.workflow_mode == "explore"
138
139
140 @pytest.mark.asyncio
141 async def test_runtime_harness_uses_runtime_handle_for_scripted_runs(
142 temp_dir: Path,
143 ) -> None:
144 run = await run_scenario(
145 "Summarize the runtime-first harness.",
146 ScriptedBackend(
147 completions=[CompletionResponse(content="Runtime harness reply.")]
148 ),
149 config=AgentConfig(auto_context=False, stream=False),
150 project_root=temp_dir,
151 )
152 explore_run = await run_explore_scenario(
153 "Where should I start?",
154 ScriptedBackend(
155 completions=[CompletionResponse(content="Explore harness reply.")]
156 ),
157 config=AgentConfig(auto_context=False, stream=False),
158 project_root=temp_dir,
159 )
160
161 assert isinstance(run.agent, RuntimeHandle)
162 assert run.response == "Runtime harness reply."
163 assert isinstance(explore_run.agent, RuntimeHandle)
164 assert explore_run.response == "Explore harness reply."
165
166
167 def test_runtime_handle_exposes_public_shell_steering_contract(
168 temp_dir: Path,
169 ) -> None:
170 handle = RuntimeHandle(
171 backend=ScriptedBackend(),
172 config=AgentConfig(auto_context=False),
173 project_root=temp_dir,
174 )
175
176 assert handle.is_running is False
177 assert handle.steer("stay in runtime") is False
178
179 handle.steering.mark_running()
180
181 assert handle.is_running is True
182 assert handle.steer("stay in runtime") is True
183 assert handle.drain_steering_messages() == ["stay in runtime"]