feat(busy): add 'steer' as a third display.busy_input_mode option (#16279)

Enter while the agent is busy can now inject the typed text via /steer —
arriving at the agent after the next tool call — instead of interrupting
(current default) or queueing for the next turn.

Changes:
- cli.py: keybinding honors busy_input_mode='steer' by calling
  agent.steer(text) on the UI thread (thread-safe), with automatic
  fallback to 'queue' when the agent is missing, steer() is unavailable,
  images are attached, or steer() rejects the payload. /busy accepts
  'steer' as a fourth argument alongside queue/interrupt/status.
- gateway/run.py: busy-message handler and the PRIORITY running-agent
  path both route through running_agent.steer() when the mode is 'steer',
  with the same fallback-to-queue safety net. Ack wording tells users
  their message was steered into the current run. Restart-drain queueing
  now also activates for 'steer' so messages aren't lost across restarts.
- agent/onboarding.py: first-touch hint has a steer branch for both
  CLI and gateway.
- hermes_cli/commands.py: /busy args_hint updated to include steer,
  and 'steer' is registered as a subcommand (completions).
- hermes_cli/web_server.py: dashboard select widget offers steer.
- hermes_cli/config.py, cli-config.yaml.example, hermes_cli/tips.py:
  inline docs updated.
- website/docs/user-guide/cli.md + messaging/index.md: documented.
- Tests: steer set/status path for /busy; onboarding hints;
  _load_busy_input_mode accepts steer; busy-session ack exercises
  steer success + two fallback-to-queue branches.

Requested on X by @CodingAcct.

Default is unchanged (interrupt).

tests/agent/test_onboarding.py

@@ -117,6 +117,12 @@ class TestHintMessages:
         assert "/busy interrupt" in msg
         assert "queued" in msg.lower()
 
+    def test_busy_input_hint_gateway_steer(self):
+        msg = busy_input_hint_gateway("steer")
+        assert "/busy interrupt" in msg
+        assert "/busy queue" in msg
+        assert "steer" in msg.lower()
+
     def test_busy_input_hint_cli_interrupt(self):
         msg = busy_input_hint_cli("interrupt")
         assert "/busy queue" in msg
@@ -125,6 +131,12 @@ class TestHintMessages:
         msg = busy_input_hint_cli("queue")
         assert "/busy interrupt" in msg
 
+    def test_busy_input_hint_cli_steer(self):
+        msg = busy_input_hint_cli("steer")
+        assert "/busy interrupt" in msg
+        assert "/busy queue" in msg
+        assert "steer" in msg.lower()
+
     def test_tool_progress_hints_mention_verbose(self):
         assert "/verbose" in tool_progress_hint_gateway()
         assert "/verbose" in tool_progress_hint_cli()
@@ -133,8 +145,10 @@ class TestHintMessages:
         for hint in (
             busy_input_hint_gateway("queue"),
             busy_input_hint_gateway("interrupt"),
+            busy_input_hint_gateway("steer"),
             busy_input_hint_cli("queue"),
             busy_input_hint_cli("interrupt"),
+            busy_input_hint_cli("steer"),
             tool_progress_hint_gateway(),
             tool_progress_hint_cli(),
         ):

tests/cli/test_busy_input_mode_command.py

@@ -65,6 +65,35 @@ class TestHandleBusyCommand(unittest.TestCase):
         self.assertEqual(stub.busy_input_mode, "interrupt")
         mock_save.assert_called_once_with("display.busy_input_mode", "interrupt")
 
+    def test_steer_argument_sets_steer_mode_and_saves(self):
+        cli_mod = _import_cli()
+        stub = self._make_cli("interrupt")
+        with (
+            patch.object(cli_mod, "_cprint") as mock_cprint,
+            patch.object(cli_mod, "save_config_value", return_value=True) as mock_save,
+        ):
+            cli_mod.HermesCLI._handle_busy_command(stub, "/busy steer")
+
+        self.assertEqual(stub.busy_input_mode, "steer")
+        mock_save.assert_called_once_with("display.busy_input_mode", "steer")
+        printed = " ".join(str(c) for c in mock_cprint.call_args_list)
+        self.assertIn("steer", printed.lower())
+
+    def test_status_reports_steer_behavior(self):
+        cli_mod = _import_cli()
+        stub = self._make_cli("steer")
+        with (
+            patch.object(cli_mod, "_cprint") as mock_cprint,
+            patch.object(cli_mod, "save_config_value") as mock_save,
+        ):
+            cli_mod.HermesCLI._handle_busy_command(stub, "/busy status")
+
+        mock_save.assert_not_called()
+        printed = " ".join(str(c) for c in mock_cprint.call_args_list)
+        self.assertIn("steer", printed.lower())
+        # The usage line should also advertise the steer option
+        self.assertIn("steer", printed)
+
     def test_invalid_argument_prints_usage(self):
         cli_mod = _import_cli()
         stub = self._make_cli()
@@ -90,5 +119,5 @@ class TestBusyCommandRegistry(unittest.TestCase):
         from hermes_cli.commands import COMMAND_REGISTRY
 
         busy = next(c for c in COMMAND_REGISTRY if c.name == "busy")
-        assert busy.args_hint == "[queue|interrupt|status]"
+        assert busy.args_hint == "[queue|steer|interrupt|status]"
         assert busy.category == "Configuration"

tests/gateway/test_busy_session_ack.py

@@ -186,6 +186,91 @@ class TestBusySessionAck:
         assert "respond once the current task finishes" in content
         assert "Interrupting" not in content
 
+    @pytest.mark.asyncio
+    async def test_steer_mode_calls_agent_steer_no_interrupt_no_queue(self):
+        """busy_input_mode='steer' injects via agent.steer() and skips queueing."""
+        runner, sentinel = _make_runner()
+        runner._busy_input_mode = "steer"
+        adapter = _make_adapter()
+
+        event = _make_event(text="also check the tests")
+        sk = build_session_key(event.source)
+        runner.adapters[event.source.platform] = adapter
+
+        agent = MagicMock()
+        agent.steer = MagicMock(return_value=True)
+        runner._running_agents[sk] = agent
+
+        with patch("gateway.run.merge_pending_message_event") as mock_merge:
+            await runner._handle_active_session_busy_message(event, sk)
+
+        # VERIFY: Agent was steered, NOT interrupted
+        agent.steer.assert_called_once_with("also check the tests")
+        agent.interrupt.assert_not_called()
+
+        # VERIFY: No queueing — successful steer must NOT replay as next turn
+        mock_merge.assert_not_called()
+
+        # VERIFY: Ack mentions steer wording
+        adapter._send_with_retry.assert_called_once()
+        call_kwargs = adapter._send_with_retry.call_args
+        content = call_kwargs.kwargs.get("content") or call_kwargs[1].get("content", "")
+        assert "Steered" in content or "steer" in content.lower()
+        assert "Interrupting" not in content
+
+    @pytest.mark.asyncio
+    async def test_steer_mode_falls_back_to_queue_when_agent_rejects(self):
+        """If agent.steer() returns False, fall back to queue behavior."""
+        runner, sentinel = _make_runner()
+        runner._busy_input_mode = "steer"
+        adapter = _make_adapter()
+
+        event = _make_event(text="empty or rejected")
+        sk = build_session_key(event.source)
+        runner.adapters[event.source.platform] = adapter
+
+        agent = MagicMock()
+        agent.steer = MagicMock(return_value=False)  # rejected
+        runner._running_agents[sk] = agent
+
+        with patch("gateway.run.merge_pending_message_event") as mock_merge:
+            await runner._handle_active_session_busy_message(event, sk)
+
+        agent.steer.assert_called_once()
+        agent.interrupt.assert_not_called()
+        # Fell back to queue semantics: event was merged into pending messages
+        mock_merge.assert_called_once()
+
+        # Ack uses queue-mode wording (not steer, not interrupt)
+        call_kwargs = adapter._send_with_retry.call_args
+        content = call_kwargs.kwargs.get("content") or call_kwargs[1].get("content", "")
+        assert "Queued for the next turn" in content
+        assert "Steered" not in content
+
+    @pytest.mark.asyncio
+    async def test_steer_mode_falls_back_to_queue_when_agent_pending(self):
+        """If agent is still starting (sentinel), steer mode falls back to queue."""
+        runner, sentinel = _make_runner()
+        runner._busy_input_mode = "steer"
+        adapter = _make_adapter()
+
+        event = _make_event(text="arrived too early")
+        sk = build_session_key(event.source)
+        runner.adapters[event.source.platform] = adapter
+
+        # Agent is still being set up — sentinel in place
+        runner._running_agents[sk] = sentinel
+
+        with patch("gateway.run.merge_pending_message_event") as mock_merge:
+            await runner._handle_active_session_busy_message(event, sk)
+
+        # Event was queued instead of steered
+        mock_merge.assert_called_once()
+
+        call_kwargs = adapter._send_with_retry.call_args
+        content = call_kwargs.kwargs.get("content") or call_kwargs[1].get("content", "")
+        assert "Queued for the next turn" in content
+
     @pytest.mark.asyncio
     async def test_debounce_suppresses_rapid_acks(self):
         """Second message within 30s should NOT send another ack."""

tests/gateway/test_restart_drain.py

@@ -90,9 +90,21 @@ def test_load_busy_input_mode_prefers_env_then_config_then_default(tmp_path, mon
     )
     assert gateway_run.GatewayRunner._load_busy_input_mode() == "queue"
 
+    (tmp_path / "config.yaml").write_text(
+        "display:\n  busy_input_mode: steer\n", encoding="utf-8"
+    )
+    assert gateway_run.GatewayRunner._load_busy_input_mode() == "steer"
+
     monkeypatch.setenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "interrupt")
     assert gateway_run.GatewayRunner._load_busy_input_mode() == "interrupt"
 
+    monkeypatch.setenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "steer")
+    assert gateway_run.GatewayRunner._load_busy_input_mode() == "steer"
+
+    # Unknown values fall through to the safe default
+    monkeypatch.setenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "bogus")
+    assert gateway_run.GatewayRunner._load_busy_input_mode() == "interrupt"
+
 
 def test_load_restart_drain_timeout_prefers_env_then_config_then_default(
     tmp_path, monkeypatch, caplog