loader Public

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