| 1 | """Tests for runtime-first CLI owner selection.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import sys |
| 6 | from types import ModuleType, SimpleNamespace |
| 7 | |
| 8 | import pytest |
| 9 | |
| 10 | import loader.agent.loop as agent_loop_module |
| 11 | import loader.cli.main as cli_main_module |
| 12 | import loader.runtime.runtime_api as runtime_api_module |
| 13 | import loader.runtime.runtime_handle as runtime_handle_module |
| 14 | from loader.agent.loop import AgentConfig |
| 15 | |
| 16 | |
| 17 | class _FakeBackend: |
| 18 | def __init__(self, **kwargs) -> None: |
| 19 | self.model = kwargs.get("model", "fake-model") |
| 20 | self.timeout = kwargs.get("timeout", 60) |
| 21 | |
| 22 | async def health_check(self) -> bool: |
| 23 | return True |
| 24 | |
| 25 | async def describe_model(self) -> dict[str, object]: |
| 26 | return {} |
| 27 | |
| 28 | def supports_native_tools(self) -> bool: |
| 29 | return True |
| 30 | |
| 31 | |
| 32 | def _install_fake_ollama_module(monkeypatch: pytest.MonkeyPatch) -> None: |
| 33 | module = ModuleType("loader.llm.ollama") |
| 34 | module.OllamaBackend = _FakeBackend |
| 35 | monkeypatch.setitem(sys.modules, "loader.llm.ollama", module) |
| 36 | |
| 37 | |
| 38 | def test_build_runtime_shell_owner_uses_runtime_handle_for_runtime_paths( |
| 39 | monkeypatch: pytest.MonkeyPatch, |
| 40 | ) -> None: |
| 41 | seen: list[dict[str, object]] = [] |
| 42 | |
| 43 | class FakeHandle: |
| 44 | def __init__(self, **kwargs) -> None: |
| 45 | seen.append(kwargs) |
| 46 | |
| 47 | class FakeAgent: |
| 48 | def __init__(self, **kwargs) -> None: |
| 49 | raise AssertionError("Agent should not be used for internal CLI paths") |
| 50 | |
| 51 | monkeypatch.setattr(runtime_handle_module, "RuntimeHandle", FakeHandle) |
| 52 | monkeypatch.setattr(agent_loop_module, "Agent", FakeAgent) |
| 53 | |
| 54 | owner = runtime_api_module.build_runtime_shell_owner( |
| 55 | backend="backend", |
| 56 | registry="registry", |
| 57 | config="config", |
| 58 | owner_kind="runtime", |
| 59 | ) |
| 60 | |
| 61 | assert isinstance(owner, FakeHandle) |
| 62 | assert seen == [ |
| 63 | { |
| 64 | "backend": "backend", |
| 65 | "registry": "registry", |
| 66 | "config": "config", |
| 67 | "project_root": None, |
| 68 | } |
| 69 | ] |
| 70 | |
| 71 | |
| 72 | def test_build_runtime_shell_owner_uses_agent_for_public_compat_paths( |
| 73 | monkeypatch: pytest.MonkeyPatch, |
| 74 | ) -> None: |
| 75 | seen: list[dict[str, object]] = [] |
| 76 | |
| 77 | class FakeHandle: |
| 78 | def __init__(self, **kwargs) -> None: |
| 79 | raise AssertionError("RuntimeHandle should not be used for public TUI paths") |
| 80 | |
| 81 | class FakeAgent: |
| 82 | def __init__(self, **kwargs) -> None: |
| 83 | seen.append(kwargs) |
| 84 | |
| 85 | monkeypatch.setattr(runtime_handle_module, "RuntimeHandle", FakeHandle) |
| 86 | monkeypatch.setattr(agent_loop_module, "Agent", FakeAgent) |
| 87 | |
| 88 | owner = runtime_api_module.build_runtime_shell_owner( |
| 89 | backend="backend", |
| 90 | registry="registry", |
| 91 | config="config", |
| 92 | owner_kind="public-compat", |
| 93 | ) |
| 94 | |
| 95 | assert isinstance(owner, FakeAgent) |
| 96 | assert seen == [ |
| 97 | { |
| 98 | "backend": "backend", |
| 99 | "registry": "registry", |
| 100 | "config": "config", |
| 101 | "project_root": None, |
| 102 | } |
| 103 | ] |
| 104 | |
| 105 | |
| 106 | @pytest.mark.asyncio |
| 107 | async def test_main_uses_runtime_first_owner_for_tui_launch( |
| 108 | monkeypatch: pytest.MonkeyPatch, |
| 109 | ) -> None: |
| 110 | fake_owner = SimpleNamespace( |
| 111 | capability_profile=SimpleNamespace( |
| 112 | preferred_tool_call_format="native", |
| 113 | verification_strictness="strict", |
| 114 | ), |
| 115 | workflow_mode="execute", |
| 116 | active_permission_mode="workspace-write", |
| 117 | session=SimpleNamespace(session_id="session-123", active_turn_phase=""), |
| 118 | project_context=None, |
| 119 | resume_session=lambda session_id=None: False, |
| 120 | ) |
| 121 | owner_calls: list[dict[str, object]] = [] |
| 122 | app_calls: list[dict[str, object]] = [] |
| 123 | |
| 124 | class FakeApp: |
| 125 | def __init__(self, **kwargs) -> None: |
| 126 | app_calls.append(kwargs) |
| 127 | |
| 128 | async def run_async(self) -> None: |
| 129 | app_calls.append({"ran": True}) |
| 130 | |
| 131 | _install_fake_ollama_module(monkeypatch) |
| 132 | monkeypatch.setattr("loader.config.get_default_model", lambda: "fake-model") |
| 133 | monkeypatch.setattr("loader.config.get_last_model", lambda: None) |
| 134 | monkeypatch.setattr("loader.config.set_last_model", lambda model: None) |
| 135 | monkeypatch.setattr( |
| 136 | "loader.tools.base.create_default_registry", |
| 137 | lambda: SimpleNamespace(skip_confirmation=False), |
| 138 | ) |
| 139 | fake_ui_app = ModuleType("loader.ui.app") |
| 140 | fake_ui_app.LoaderApp = FakeApp |
| 141 | monkeypatch.setitem(sys.modules, "loader.ui.app", fake_ui_app) |
| 142 | |
| 143 | def fake_build_owner(*, backend, registry, config, owner_kind): |
| 144 | owner_calls.append( |
| 145 | { |
| 146 | "backend": backend, |
| 147 | "registry": registry, |
| 148 | "config": config, |
| 149 | "owner_kind": owner_kind, |
| 150 | } |
| 151 | ) |
| 152 | return fake_owner |
| 153 | |
| 154 | monkeypatch.setattr(cli_main_module, "build_runtime_shell_owner", fake_build_owner) |
| 155 | |
| 156 | await cli_main_module._main( |
| 157 | model="fake-model", |
| 158 | select_model=False, |
| 159 | backend="ollama", |
| 160 | yes=False, |
| 161 | permission_mode="workspace-write", |
| 162 | react=False, |
| 163 | no_context=True, |
| 164 | plan=False, |
| 165 | clarify=False, |
| 166 | resume_target=None, |
| 167 | no_recover=False, |
| 168 | no_tui=False, |
| 169 | ctx=8192, |
| 170 | gpu=-1, |
| 171 | timeout=60, |
| 172 | decompose=False, |
| 173 | critique=False, |
| 174 | confidence=False, |
| 175 | verify=False, |
| 176 | reason=False, |
| 177 | prompt=None, |
| 178 | ) |
| 179 | |
| 180 | assert owner_calls and owner_calls[0]["owner_kind"] == "runtime" |
| 181 | assert app_calls[0]["shell_owner"] is fake_owner |
| 182 | assert app_calls[-1] == {"ran": True} |
| 183 | |
| 184 | |
| 185 | @pytest.mark.asyncio |
| 186 | async def test_main_uses_runtime_first_owner_for_single_prompt( |
| 187 | monkeypatch: pytest.MonkeyPatch, |
| 188 | ) -> None: |
| 189 | seen: list[dict[str, object]] = [] |
| 190 | fake_owner = SimpleNamespace( |
| 191 | capability_profile=SimpleNamespace( |
| 192 | preferred_tool_call_format="native", |
| 193 | verification_strictness="strict", |
| 194 | ), |
| 195 | workflow_mode="execute", |
| 196 | active_permission_mode="workspace-write", |
| 197 | session=SimpleNamespace(session_id="session-123"), |
| 198 | project_context=None, |
| 199 | resume_session=lambda session_id=None: False, |
| 200 | ) |
| 201 | |
| 202 | _install_fake_ollama_module(monkeypatch) |
| 203 | monkeypatch.setattr("loader.config.get_default_model", lambda: "fake-model") |
| 204 | monkeypatch.setattr("loader.config.get_last_model", lambda: None) |
| 205 | monkeypatch.setattr("loader.config.set_last_model", lambda model: None) |
| 206 | monkeypatch.setattr( |
| 207 | "loader.tools.base.create_default_registry", |
| 208 | lambda: SimpleNamespace(skip_confirmation=False), |
| 209 | ) |
| 210 | |
| 211 | def fake_build_owner(*, backend, registry, config, owner_kind): |
| 212 | seen.append( |
| 213 | { |
| 214 | "backend": backend, |
| 215 | "registry": registry, |
| 216 | "config": config, |
| 217 | "owner_kind": owner_kind, |
| 218 | } |
| 219 | ) |
| 220 | return fake_owner |
| 221 | |
| 222 | captured = {} |
| 223 | |
| 224 | async def fake_run_once(owner, prompt: str, skip_confirmation: bool = False) -> None: |
| 225 | captured["owner"] = owner |
| 226 | captured["prompt"] = prompt |
| 227 | captured["skip_confirmation"] = skip_confirmation |
| 228 | |
| 229 | monkeypatch.setattr(cli_main_module, "build_runtime_shell_owner", fake_build_owner) |
| 230 | monkeypatch.setattr(cli_main_module, "run_once", fake_run_once) |
| 231 | |
| 232 | await cli_main_module._main( |
| 233 | model="fake-model", |
| 234 | select_model=False, |
| 235 | backend="ollama", |
| 236 | yes=False, |
| 237 | permission_mode="workspace-write", |
| 238 | react=False, |
| 239 | no_context=True, |
| 240 | plan=False, |
| 241 | clarify=False, |
| 242 | resume_target=None, |
| 243 | no_recover=False, |
| 244 | no_tui=False, |
| 245 | ctx=8192, |
| 246 | gpu=-1, |
| 247 | timeout=60, |
| 248 | decompose=False, |
| 249 | critique=False, |
| 250 | confidence=False, |
| 251 | verify=False, |
| 252 | reason=False, |
| 253 | prompt="Summarize the runtime-first shell path.", |
| 254 | ) |
| 255 | |
| 256 | assert seen and seen[0]["owner_kind"] == "runtime" |
| 257 | assert isinstance(seen[0]["config"], AgentConfig) |
| 258 | assert captured == { |
| 259 | "owner": fake_owner, |
| 260 | "prompt": "Summarize the runtime-first shell path.", |
| 261 | "skip_confirmation": False, |
| 262 | } |
| 263 | |
| 264 | |
| 265 | @pytest.mark.asyncio |
| 266 | async def test_explore_main_uses_runtime_first_owner( |
| 267 | monkeypatch: pytest.MonkeyPatch, |
| 268 | ) -> None: |
| 269 | seen: list[dict[str, object]] = [] |
| 270 | |
| 271 | class FakeExploreOwner: |
| 272 | use_react = False |
| 273 | |
| 274 | async def run_explore(self, prompt: str, on_event=None, *, fresh: bool = False) -> str: |
| 275 | seen.append({"prompt": prompt, "fresh": fresh, "on_event": on_event}) |
| 276 | return "Explore reply." |
| 277 | |
| 278 | _install_fake_ollama_module(monkeypatch) |
| 279 | monkeypatch.setattr("loader.config.get_default_model", lambda: "fake-model") |
| 280 | monkeypatch.setattr("loader.config.get_last_model", lambda: None) |
| 281 | monkeypatch.setattr("loader.config.set_last_model", lambda model: None) |
| 282 | |
| 283 | owner_calls: list[dict[str, object]] = [] |
| 284 | |
| 285 | def fake_build_owner(*, backend, registry, config, owner_kind): |
| 286 | owner_calls.append( |
| 287 | { |
| 288 | "backend": backend, |
| 289 | "registry": registry, |
| 290 | "config": config, |
| 291 | "owner_kind": owner_kind, |
| 292 | } |
| 293 | ) |
| 294 | return FakeExploreOwner() |
| 295 | |
| 296 | monkeypatch.setattr(cli_main_module, "build_runtime_shell_owner", fake_build_owner) |
| 297 | |
| 298 | await cli_main_module._explore_main( |
| 299 | model="fake-model", |
| 300 | select_model=False, |
| 301 | backend="ollama", |
| 302 | react=False, |
| 303 | no_context=True, |
| 304 | fresh=True, |
| 305 | ctx=8192, |
| 306 | gpu=-1, |
| 307 | timeout=60, |
| 308 | prompt="Where should I start?", |
| 309 | ) |
| 310 | |
| 311 | assert owner_calls and owner_calls[0]["owner_kind"] == "runtime" |
| 312 | assert isinstance(owner_calls[0]["config"], AgentConfig) |
| 313 | assert len(seen) == 1 |
| 314 | assert seen[0]["prompt"] == "Where should I start?" |
| 315 | assert seen[0]["fresh"] is True |
| 316 | assert callable(seen[0]["on_event"]) |