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