fix(gateway): bypass slash commands during pending update prompts

tests/gateway/test_session_boundary_security_state.py

@@ -76,6 +76,7 @@ def _make_resume_runner():
     runner._running_agents_ts = {}
     runner._busy_ack_ts = {}
     runner._pending_approvals = {}
+    runner._update_prompt_pending = {}
     runner._agent_cache_lock = None
     runner.session_store = MagicMock()
     runner.session_store.get_or_create_session.return_value = current_entry
@@ -102,6 +103,7 @@ def _make_branch_runner():
     runner._running_agents_ts = {}
     runner._busy_ack_ts = {}
     runner._pending_approvals = {}
+    runner._update_prompt_pending = {}
     runner._agent_cache_lock = None
     runner.session_store = MagicMock()
     runner.session_store.get_or_create_session.return_value = current_entry
@@ -127,6 +129,8 @@ async def test_resume_clears_session_scoped_approval_and_yolo_state():
     enable_session_yolo(other_key)
     runner._pending_approvals[session_key] = {"command": "rm -rf /tmp/demo"}
     runner._pending_approvals[other_key] = {"command": "rm -rf /tmp/other"}
+    runner._update_prompt_pending[session_key] = True
+    runner._update_prompt_pending[other_key] = True
 
     result = await runner._handle_resume_command(_make_event("/resume Resumed Work"))
 
@@ -134,9 +138,11 @@ async def test_resume_clears_session_scoped_approval_and_yolo_state():
     assert is_approved(session_key, "recursive delete") is False
     assert is_session_yolo_enabled(session_key) is False
     assert session_key not in runner._pending_approvals
+    assert session_key not in runner._update_prompt_pending
     assert is_approved(other_key, "recursive delete") is True
     assert is_session_yolo_enabled(other_key) is True
     assert other_key in runner._pending_approvals
+    assert other_key in runner._update_prompt_pending
 
 
 @pytest.mark.asyncio
@@ -150,6 +156,8 @@ async def test_branch_clears_session_scoped_approval_and_yolo_state():
     enable_session_yolo(other_key)
     runner._pending_approvals[session_key] = {"command": "rm -rf /tmp/demo"}
     runner._pending_approvals[other_key] = {"command": "rm -rf /tmp/other"}
+    runner._update_prompt_pending[session_key] = True
+    runner._update_prompt_pending[other_key] = True
 
     result = await runner._handle_branch_command(_make_event("/branch"))
 
@@ -157,9 +165,11 @@ async def test_branch_clears_session_scoped_approval_and_yolo_state():
     assert is_approved(session_key, "recursive delete") is False
     assert is_session_yolo_enabled(session_key) is False
     assert session_key not in runner._pending_approvals
+    assert session_key not in runner._update_prompt_pending
     assert is_approved(other_key, "recursive delete") is True
     assert is_session_yolo_enabled(other_key) is True
     assert other_key in runner._pending_approvals
+    assert other_key in runner._update_prompt_pending
 
 
 def test_clear_session_boundary_security_state_is_scoped():
@@ -172,6 +182,7 @@ def test_clear_session_boundary_security_state_is_scoped():
 
     runner = object.__new__(GatewayRunner)
     runner._pending_approvals = {}
+    runner._update_prompt_pending = {}
 
     source = _make_source()
     session_key = build_session_key(source)
@@ -183,6 +194,8 @@ def test_clear_session_boundary_security_state_is_scoped():
     enable_session_yolo(other_key)
     runner._pending_approvals[session_key] = {"command": "rm -rf /tmp/demo"}
     runner._pending_approvals[other_key] = {"command": "rm -rf /tmp/other"}
+    runner._update_prompt_pending[session_key] = True
+    runner._update_prompt_pending[other_key] = True
 
     runner._clear_session_boundary_security_state(session_key)
 
@@ -190,11 +203,14 @@ def test_clear_session_boundary_security_state_is_scoped():
     assert is_approved(session_key, "recursive delete") is False
     assert is_session_yolo_enabled(session_key) is False
     assert session_key not in runner._pending_approvals
+    assert session_key not in runner._update_prompt_pending
     # Other session untouched
     assert is_approved(other_key, "recursive delete") is True
     assert is_session_yolo_enabled(other_key) is True
     assert other_key in runner._pending_approvals
+    assert other_key in runner._update_prompt_pending
 
     # Empty session_key is a no-op
     runner._clear_session_boundary_security_state("")
     assert is_approved(other_key, "recursive delete") is True
+    assert other_key in runner._update_prompt_pending

tests/gateway/test_update_streaming.py

@@ -251,7 +251,7 @@ class TestWatchUpdateProgress:
                    "session_key": "agent:main:telegram:dm:111"}
         (hermes_home / ".update_pending.json").write_text(json.dumps(pending))
         # Write output
-        (hermes_home / ".update_output.txt").write_text("→ Fetching updates...\n")
+        (hermes_home / ".update_output.txt").write_text("→ Fetching updates...\n", encoding="utf-8")
 
         mock_adapter = AsyncMock()
         runner.adapters = {Platform.TELEGRAM: mock_adapter}
@@ -261,7 +261,7 @@ class TestWatchUpdateProgress:
             await asyncio.sleep(0.3)
             (hermes_home / ".update_output.txt").write_text(
                 "→ Fetching updates...\n✓ Code updated!\n"
-            )
+            , encoding="utf-8")
             (hermes_home / ".update_exit_code").write_text("0")
 
         with patch("gateway.run._hermes_home", hermes_home):
@@ -489,6 +489,28 @@ class TestUpdatePromptInterception:
         # Should clear the pending flag
         assert session_key not in runner._update_prompt_pending
 
+    @pytest.mark.asyncio
+    async def test_recognized_slash_command_bypasses_pending_update_prompt(self, tmp_path):
+        """Known slash commands must dispatch normally instead of being consumed."""
+        runner = _make_runner()
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+
+        event = _make_event(text="/new", chat_id="67890")
+        session_key = "agent:main:telegram:dm:67890"
+        runner._update_prompt_pending[session_key] = True
+        runner._is_user_authorized = MagicMock(return_value=True)
+        runner._session_key_for_source = MagicMock(return_value=session_key)
+        runner._handle_reset_command = AsyncMock(return_value="reset ok")
+
+        with patch("gateway.run._hermes_home", hermes_home):
+            result = await runner._handle_message(event)
+
+        assert result == "reset ok"
+        runner._handle_reset_command.assert_awaited_once_with(event)
+        assert not (hermes_home / ".update_response").exists()
+        assert runner._update_prompt_pending[session_key] is True
+
     @pytest.mark.asyncio
     async def test_normal_message_when_no_prompt_pending(self, tmp_path):
         """Messages pass through normally when no prompt is pending."""