Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -7819,6 +7819,36 @@ def _should_handle_steer_command_inline(self, text: str, has_images: bool = Fals
except Exception:
return False

def _should_handle_agents_command_inline(self, text: str, has_images: bool = False) -> bool:
"""Return True when /agents (alias /tasks) should be dispatched inline while busy.

Same constraint as /steer: while the agent is running, ``process_loop`` is
blocked inside ``self.chat()`` and never drains ``_pending_input`` until
the run completes. /agents is a read-only introspection command — the
whole point is to monitor in-flight delegations — so queueing makes the
command effectively silent during exactly the window it exists for.
``_handle_agents_command`` only reads ``process_registry`` plus a couple
of CLI attributes and emits through ``_cprint`` (which routes through
``run_in_terminal``), so it's safe to dispatch on the UI thread.
"""
if not text or has_images or not _looks_like_slash_command(text):
return False
if not getattr(self, "_agent_running", False):
return False
# /agents takes no args (unlike /steer which carries a payload).
# Inputs like "/agents foo" should fall through to the normal path
# where _handle_agents_command can surface a usage error.
parts = text.split()
if len(parts) != 1:
return False
try:
from hermes_cli.commands import resolve_command
base = parts[0].lower().lstrip('/')
cmd = resolve_command(base)
Comment on lines +7844 to +7847
return bool(cmd and cmd.name == "agents")
except Exception:
return False
Comment on lines +7834 to +7850

def _output_console(self):
"""Use prompt_toolkit-safe Rich rendering once the TUI is live."""
if getattr(self, "_app", None):
Expand Down Expand Up @@ -12704,6 +12734,16 @@ def handle_enter(event):
event.app.current_buffer.reset(append_to_history=True)
return

# Handle /agents (alias /tasks) immediately when the agent is
# running — same deadlock as /steer above. This command is the
# in-flight delegation monitor, so queueing it for after the run
# makes it useless: the user sees "nothing happens" until the
# whole delegation chain finishes. See #32477.
if self._should_handle_agents_command_inline(text, has_images=has_images):
self.process_command(text)
event.app.current_buffer.reset(append_to_history=True)
return

# Snapshot and clear attached images
images = list(self._attached_images)
self._attached_images.clear()
Expand Down
154 changes: 154 additions & 0 deletions tests/cli/test_cli_agents_busy_path.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
"""Regression tests for classic-CLI mid-run /agents (alias /tasks) dispatch.

Background
----------
/agents (and its /tasks alias) is the in-flight introspection command —
users type it precisely because they want to monitor delegations *while*
the agent loop is running. Without the inline-dispatch path the keystroke
flows through ``_pending_input`` to ``process_loop``, which is blocked
inside ``self.chat()`` for the duration of the run. The command therefore
sits silently in the queue and never executes until the whole delegation
chain finishes — from the user's perspective "nothing happens" (#32477).

The fix mirrors the existing /steer pattern: detect the command on the UI
thread inside ``handle_enter`` and call ``process_command`` directly,
since ``_handle_agents_command`` is read-only and ``_cprint`` already
routes thread-unsafe output through ``run_in_terminal``.
"""

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 TestAgentsInlineDetector:
"""_should_handle_agents_command_inline gates the busy-path fast dispatch."""

def test_detects_agents_when_agent_running(self):
cli = _make_cli()
cli._agent_running = True
assert cli._should_handle_agents_command_inline("/agents") is True

def test_detects_tasks_alias_when_agent_running(self):
"""/tasks is an alias for /agents — both should dispatch inline."""
cli = _make_cli()
cli._agent_running = True
assert cli._should_handle_agents_command_inline("/tasks") is True

def test_ignores_agents_when_agent_idle(self):
"""Idle-path /agents should fall through to the normal process_loop
dispatch — the queue isn't blocked, so the inline shortcut is
unnecessary and would just duplicate work."""
cli = _make_cli()
cli._agent_running = False
assert cli._should_handle_agents_command_inline("/agents") is False
assert cli._should_handle_agents_command_inline("/tasks") is False

def test_ignores_non_slash_input(self):
cli = _make_cli()
cli._agent_running = True
assert cli._should_handle_agents_command_inline("agents") is False
assert cli._should_handle_agents_command_inline("") is False

def test_ignores_other_slash_commands(self):
cli = _make_cli()
cli._agent_running = True
assert cli._should_handle_agents_command_inline("/queue hello") is False
assert cli._should_handle_agents_command_inline("/stop") is False
assert cli._should_handle_agents_command_inline("/help") is False
assert cli._should_handle_agents_command_inline("/steer focus") is False

def test_ignores_agents_with_attached_images(self):
"""Image payloads take the normal path; /agents takes no args."""
cli = _make_cli()
cli._agent_running = True
assert cli._should_handle_agents_command_inline("/agents", has_images=True) is False

def test_ignores_agents_with_trailing_args(self):
"""/agents takes no args, so '/agents foo' must fall through to the
normal path where _handle_agents_command can surface a usage error
rather than silently swallowing the argument on the busy path."""
cli = _make_cli()
cli._agent_running = True
assert cli._should_handle_agents_command_inline("/agents foo") is False
assert cli._should_handle_agents_command_inline("/tasks something") is False


class TestAgentsBusyPathDispatch:
"""When the detector fires, /agents dispatch through process_command must
reach _handle_agents_command directly rather than being queued."""

def test_process_command_routes_to_agents_handler(self):
"""With _agent_running=True, /agents calls _handle_agents_command and
does NOT enqueue onto _pending_input."""
cli = _make_cli()
cli._agent_running = True
cli._pending_input = MagicMock()
cli._handle_agents_command = MagicMock()

cli.process_command("/agents")

cli._handle_agents_command.assert_called_once()
cli._pending_input.put.assert_not_called()

def test_tasks_alias_routes_to_agents_handler(self):
"""/tasks resolves to canonical 'agents' and dispatches the same."""
cli = _make_cli()
cli._agent_running = True
cli._pending_input = MagicMock()
cli._handle_agents_command = MagicMock()

cli.process_command("/tasks")

cli._handle_agents_command.assert_called_once()
Comment on lines +122 to +147
Comment on lines +122 to +147
cli._pending_input.put.assert_not_called()


if __name__ == "__main__": # pragma: no cover
import pytest

pytest.main([__file__, "-v"])
Loading