fix(cli): dispatch /steer inline while agent is running (#13354)

Classic-CLI /steer typed during an active agent run was queued through
self._pending_input alongside ordinary user input.  process_loop, which
drains that queue, is blocked inside self.chat() for the entire run,
so the queued command was not pulled until AFTER _agent_running had
flipped back to False — at which point process_command() took the idle
fallback ("No agent running; queued as next turn") and delivered the
steer as an ordinary next-turn user message.

From Utku's bug report on PR #13205: mid-run /steer arrived minutes
later at the end of the turn as a /queue-style message, completely
defeating its purpose.

Fix: add _should_handle_steer_command_inline() gating — when
_agent_running is True and the user typed /steer, dispatch
process_command(text) directly from the prompt_toolkit Enter handler
on the UI thread instead of queueing.  This mirrors the existing
_should_handle_model_command_inline() pattern for /model and is
safe because agent.steer() is thread-safe (uses _pending_steer_lock,
no prompt_toolkit state mutation, instant return).

No changes to the idle-path behavior: /steer typed with no active
agent still takes the normal queue-and-drain route so the fallback
"No agent running; queued as next turn" message is preserved.

Validation:
- 7 new unit tests in tests/cli/test_cli_steer_busy_path.py covering
  the detector, dispatch path, and idle-path control behavior.
- All 21 existing tests in tests/run_agent/test_steer.py still pass.
- Live PTY end-to-end test with real agent + real openrouter model:
    22:36:22 API call #1 (model requested execute_code)
    22:36:26 ENTER FIRED: agent_running=True, text='/steer ...'
    22:36:26 INLINE STEER DISPATCH fired
    22:36:43 agent.log: 'Delivered /steer to agent after tool batch'
    22:36:44 API call #2 included the steer; response contained marker
  Same test on the tip of main without this fix shows the steer
  landing as a new user turn ~20s after the run ended.

cli.py

@@ -5256,6 +5256,30 @@ class HermesCLI:
         except Exception:
             return False
 
+    def _should_handle_steer_command_inline(self, text: str, has_images: bool = False) -> bool:
+        """Return True when /steer should be dispatched immediately while the agent is running.
+
+        /steer MUST bypass the normal _pending_input → process_loop path when
+        the agent is active, because process_loop is blocked inside
+        self.chat() for the duration of the run.  By the time the queued
+        command is pulled from _pending_input, _agent_running has already
+        flipped back to False, and process_command() takes the idle
+        fallback — delivering the steer as a next-turn message instead of
+        injecting it mid-run.  Dispatching inline on the UI thread calls
+        agent.steer() directly, which is thread-safe (uses _pending_steer_lock).
+        """
+        if not text or has_images or not _looks_like_slash_command(text):
+            return False
+        if not getattr(self, "_agent_running", False):
+            return False
+        try:
+            from hermes_cli.commands import resolve_command
+            base = text.split(None, 1)[0].lower().lstrip('/')
+            cmd = resolve_command(base)
+            return bool(cmd and cmd.name == "steer")
+        except Exception:
+            return False
+
     def _show_model_and_providers(self):
         """Show current model + provider and list all authenticated providers.
 
@@ -9068,6 +9092,17 @@ class HermesCLI:
                     event.app.current_buffer.reset(append_to_history=True)
                     return
 
+                # Handle /steer while the agent is running immediately on the
+                # UI thread.  Queuing through _pending_input would deadlock the
+                # steer until after the agent loop finishes (process_loop is
+                # blocked inside self.chat()), which turns /steer into a
+                # post-run next-turn message — defeating mid-run injection.
+                # agent.steer() is thread-safe (holds _pending_steer_lock).
+                if self._should_handle_steer_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()

tests/cli/test_cli_steer_busy_path.py

@@ -0,0 +1,146 @@
+"""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"])