test_cli_steer_busy_path.py
1 """Regression tests for classic-CLI mid-run /steer dispatch. 2 3 Background 4 ---------- 5 /steer sent while the agent is running used to be queued through 6 ``self._pending_input`` alongside ordinary user input. ``process_loop`` 7 pulls from that queue and calls ``process_command()`` — but while the 8 agent is running, ``process_loop`` is blocked inside ``self.chat()``. 9 By the time the queued /steer was pulled, ``_agent_running`` had 10 already flipped back to False, so ``process_command()`` took the idle 11 fallback (``"No agent running; queued as next turn"``) and delivered 12 the steer as an ordinary next-turn message. 13 14 The fix dispatches /steer inline on the UI thread when the agent is 15 running — matching the existing pattern for /model — so the steer 16 reaches ``agent.steer()`` (thread-safe) without touching the queue. 17 18 These tests exercise the detector + inline dispatch without starting a 19 prompt_toolkit app. 20 """ 21 22 from __future__ import annotations 23 24 import importlib 25 import sys 26 from unittest.mock import MagicMock, patch 27 28 29 def _make_cli(): 30 """Create a HermesCLI instance with prompt_toolkit stubbed out.""" 31 _clean_config = { 32 "model": { 33 "default": "anthropic/claude-opus-4.6", 34 "base_url": "https://openrouter.ai/api/v1", 35 "provider": "auto", 36 }, 37 "display": {"compact": False, "tool_progress": "all"}, 38 "agent": {}, 39 "terminal": {"env_type": "local"}, 40 } 41 clean_env = {"LLM_MODEL": "", "HERMES_MAX_ITERATIONS": ""} 42 prompt_toolkit_stubs = { 43 "prompt_toolkit": MagicMock(), 44 "prompt_toolkit.history": MagicMock(), 45 "prompt_toolkit.styles": MagicMock(), 46 "prompt_toolkit.patch_stdout": MagicMock(), 47 "prompt_toolkit.application": MagicMock(), 48 "prompt_toolkit.layout": MagicMock(), 49 "prompt_toolkit.layout.processors": MagicMock(), 50 "prompt_toolkit.filters": MagicMock(), 51 "prompt_toolkit.layout.dimension": MagicMock(), 52 "prompt_toolkit.layout.menus": MagicMock(), 53 "prompt_toolkit.widgets": MagicMock(), 54 "prompt_toolkit.key_binding": MagicMock(), 55 "prompt_toolkit.completion": MagicMock(), 56 "prompt_toolkit.formatted_text": MagicMock(), 57 "prompt_toolkit.auto_suggest": MagicMock(), 58 } 59 with patch.dict(sys.modules, prompt_toolkit_stubs), patch.dict( 60 "os.environ", clean_env, clear=False 61 ): 62 import cli as _cli_mod 63 64 _cli_mod = importlib.reload(_cli_mod) 65 with patch.object(_cli_mod, "get_tool_definitions", return_value=[]), patch.dict( 66 _cli_mod.__dict__, {"CLI_CONFIG": _clean_config} 67 ): 68 return _cli_mod.HermesCLI() 69 70 71 class TestSteerInlineDetector: 72 """_should_handle_steer_command_inline gates the busy-path fast dispatch.""" 73 74 def test_detects_steer_when_agent_running(self): 75 cli = _make_cli() 76 cli._agent_running = True 77 assert cli._should_handle_steer_command_inline("/steer focus on error handling") is True 78 79 def test_ignores_steer_when_agent_idle(self): 80 """Idle-path /steer should fall through to the normal process_loop 81 dispatch so the queue-style fallback message is emitted.""" 82 cli = _make_cli() 83 cli._agent_running = False 84 assert cli._should_handle_steer_command_inline("/steer do something") is False 85 86 def test_ignores_non_slash_input(self): 87 cli = _make_cli() 88 cli._agent_running = True 89 assert cli._should_handle_steer_command_inline("steer without slash") is False 90 assert cli._should_handle_steer_command_inline("") is False 91 92 def test_ignores_other_slash_commands(self): 93 cli = _make_cli() 94 cli._agent_running = True 95 assert cli._should_handle_steer_command_inline("/queue hello") is False 96 assert cli._should_handle_steer_command_inline("/stop") is False 97 assert cli._should_handle_steer_command_inline("/help") is False 98 99 def test_ignores_steer_with_attached_images(self): 100 """Image payloads take the normal path; steer doesn't accept images.""" 101 cli = _make_cli() 102 cli._agent_running = True 103 assert cli._should_handle_steer_command_inline("/steer text", has_images=True) is False 104 105 106 class TestSteerBusyPathDispatch: 107 """When the detector fires, process_command('/steer ...') must call 108 agent.steer() directly rather than the idle-path fallback.""" 109 110 def test_process_command_routes_to_agent_steer(self): 111 """With _agent_running=True and agent.steer present, /steer reaches 112 agent.steer(payload), NOT _pending_input.""" 113 cli = _make_cli() 114 cli._agent_running = True 115 cli.agent = MagicMock() 116 cli.agent.steer = MagicMock(return_value=True) 117 # Make sure the idle-path fallback would be observable if taken 118 cli._pending_input = MagicMock() 119 120 cli.process_command("/steer focus on errors") 121 122 cli.agent.steer.assert_called_once_with("focus on errors") 123 cli._pending_input.put.assert_not_called() 124 125 def test_idle_path_queues_as_next_turn(self): 126 """Control — when the agent is NOT running, /steer correctly falls 127 back to next-turn queue semantics. Demonstrates why the fix was 128 needed: the queue path only works when you can actually drain it.""" 129 cli = _make_cli() 130 cli._agent_running = False 131 cli.agent = MagicMock() 132 cli.agent.steer = MagicMock(return_value=True) 133 cli._pending_input = MagicMock() 134 135 cli.process_command("/steer would-be-next-turn") 136 137 # Idle path does NOT call agent.steer 138 cli.agent.steer.assert_not_called() 139 # It puts the payload in the queue as a normal next-turn message 140 cli._pending_input.put.assert_called_once_with("would-be-next-turn") 141 142 143 if __name__ == "__main__": # pragma: no cover 144 import pytest 145 146 pytest.main([__file__, "-v"])