fix: handle /reasoning while gateway session is busy

gateway/run.py

@@ -1755,10 +1755,15 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                 return await self._handle_status_command(event)
 
             # Resolve the command once for all early-intercept checks below.
+            # Busy-safe meta commands like /reasoning should still execute
+            # immediately instead of being queued as plain user text.
             from hermes_cli.commands import resolve_command as _resolve_cmd_inner
             _evt_cmd = event.get_command()
             _cmd_def_inner = _resolve_cmd_inner(_evt_cmd) if _evt_cmd else None
 
+            if _cmd_def_inner and _cmd_def_inner.name == "reasoning":
+                return await self._handle_reasoning_command(event)
+
             # /stop must hard-kill the session when an agent is running.
             # A soft interrupt (agent.interrupt()) doesn't help when the agent
             # is truly hung — the executor thread is blocked and never checks

tests/gateway/test_session_race_guard.py

@@ -309,7 +309,70 @@ async def test_stop_clears_pending_messages():
 
 
 # ------------------------------------------------------------------
-# Test 7: Shutdown skips sentinel entries
+# Test 7: /reasoning should still work while a session is busy
+# ------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_reasoning_command_during_sentinel_dispatches_immediately():
+    """If /reasoning arrives while the session is still in the sentinel
+    startup window, it should be handled immediately instead of being queued
+    as plain text for the agent."""
+    runner = _make_runner()
+    runner.hooks = MagicMock()
+    runner.hooks.emit = AsyncMock()
+    runner._handle_reasoning_command = AsyncMock(return_value="reasoning settings")
+
+    event1 = _make_event(text="hello")
+    session_key = build_session_key(event1.source)
+    barrier = asyncio.Event()
+
+    async def slow_inner(self_inner, ev, src, qk):
+        await barrier.wait()
+        return "ok"
+
+    with patch.object(GatewayRunner, "_handle_message_with_agent", slow_inner):
+        task1 = asyncio.create_task(runner._handle_message(event1))
+        await asyncio.sleep(0)
+
+        assert runner._running_agents.get(session_key) is _AGENT_PENDING_SENTINEL
+
+        result = await runner._handle_message(_make_event(text="/reasoning"))
+
+        assert result == "reasoning settings"
+        runner._handle_reasoning_command.assert_awaited_once()
+        assert runner._running_agents.get(session_key) is _AGENT_PENDING_SENTINEL
+        assert session_key not in runner.adapters[Platform.TELEGRAM]._pending_messages
+
+        barrier.set()
+        await task1
+
+
+@pytest.mark.asyncio
+async def test_reasoning_command_during_running_agent_does_not_interrupt():
+    """A busy-session /reasoning command should bypass the interrupt path and
+    return its command response directly."""
+    runner = _make_runner()
+    runner.hooks = MagicMock()
+    runner.hooks.emit = AsyncMock()
+    runner._handle_reasoning_command = AsyncMock(return_value="reasoning settings")
+
+    source = SessionSource(platform=Platform.TELEGRAM, chat_id="12345", chat_type="dm")
+    session_key = build_session_key(source)
+    fake_agent = MagicMock()
+    runner._running_agents[session_key] = fake_agent
+
+    result = await runner._handle_message(
+        MessageEvent(text="/reasoning", message_type=MessageType.COMMAND, source=source)
+    )
+
+    assert result == "reasoning settings"
+    runner._handle_reasoning_command.assert_awaited_once()
+    assert runner._running_agents.get(session_key) is fake_agent
+    fake_agent.interrupt.assert_not_called()
+    assert session_key not in runner._pending_messages
+
+
+# ------------------------------------------------------------------
+# Test 8: Shutdown skips sentinel entries
 # ------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_shutdown_skips_sentinel():