fix(acp): block state-mutating commands while a turn is running

prompt() dispatches slash commands on a worker thread before claiming the
turn, so /reset and /compress could clear or rebind state.history (and null
agent._session_db) underneath a live run_conversation. /model and
session/set_model could likewise swap state.agent mid-turn, which also made
_finish_turn emit a spurious compression-rotation update.

Mutating commands now hold a per-session command_op flag for their whole
run: they are rejected while a turn or another op is in flight, prompts
arriving mid-op queue instead of claiming the turn, and the queue drains
through the same helper _finish_turn uses. Gateway parity: these commands
are idle-only there.

acp_adapter/commands.py

@@ -36,6 +36,15 @@ def _queue_prompt(state: SessionState, text: str) -> int:
         return len(state.queued_prompts)
 
 
+# Commands that mutate shared turn state must not run beside a live turn or beside each
+# other: slash dispatch happens on a worker thread while the turn iterates state.history and
+# reads agent._session_db, so clearing, rebinding, or swapping state.agent underneath
+# run_conversation tears the running turn. Gateway parity: all three are idle-only there.
+# The command_op flag is held for the whole handler so a turn cannot claim the session in
+# the check-then-act window (the /compress LLM call and /model agent rebuild take seconds).
+_MID_TURN_BLOCKED_COMMANDS = frozenset({"reset", "compress", "model"})
+
+
 class SlashCommandsMixin:
     """Slash-command surface for ``HermesACPAgent``; relies on ``_conn``, ``_send``, ``_schedule_soon``,
     ``session_manager`` and ``_switch_model`` from the host class."""
@@ -93,6 +102,13 @@ class SlashCommandsMixin:
 
         if cmd not in self._COMMANDS:
             return None
+        mutating = cmd in _MID_TURN_BLOCKED_COMMANDS
+        if mutating:
+            with state.runtime_lock:
+                if state.is_running or state.command_op:
+                    return (f"⏳ Session is busy; /{cmd} only works while the session is "
+                            "idle. Wait for the current response or cancel first.")
+                state.command_op = True
         handler = getattr(self, f"_cmd_{cmd}")
 
         # Handlers run outside the per-turn cwd-pinning context. ``/compress``
@@ -112,6 +128,10 @@ class SlashCommandsMixin:
         except Exception as e:
             logger.error("Slash command /%s error: %s", cmd, e, exc_info=True)
             return f"Error executing /{cmd}: {e}"
+        finally:
+            if mutating:
+                with state.runtime_lock:
+                    state.command_op = False
 
     def _cmd_help(self, args: str, state: SessionState) -> str:
         lines = ["Available commands:", ""]

acp_adapter/server.py

@@ -728,11 +728,12 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         """Mark the session running; if a turn is active, redirect it (text-only, supported
         runtime) or queue it. Returns the client message when absorbed, else None."""
         with state.runtime_lock:
-            if not state.is_running:
+            if not state.is_running and not state.command_op:
                 state.is_running = True
                 state.current_prompt_text = user_text or "[Image attachment]"
                 return None
-            if text_only and isinstance(user_content, str) and hasattr(state.agent, "redirect") and (
+            # Redirect steers a live turn; a state-mutating command (command_op) has none.
+            if state.is_running and text_only and isinstance(user_content, str) and hasattr(state.agent, "redirect") and (
                 getattr(state.agent, "_supports_active_turn_redirect", False) is True
             ):
                 try:
@@ -831,6 +832,8 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
                 if self._conn:
                     await self._conn.session_update(session_id, acp.update_agent_message_text(response_text))
                     await self._send_usage_update(state)
+                # A mutating command held command_op; prompts that arrived mid-op are queued.
+                await self._drain_queued_prompts(state, session_id, self._conn)
                 return PromptResponse(stop_reason="end_turn")
 
         absorbed = self._claim_turn_or_queue(state, session_id, user_text, user_content, text_only_prompt)
@@ -983,14 +986,7 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
             with state.runtime_lock:
                 state.is_running = False
                 state.current_prompt_text = ""
-            while True:
-                with state.runtime_lock:
-                    if not state.queued_prompts:
-                        break
-                    next_prompt = state.queued_prompts.pop(0)
-                if conn:
-                    await conn.session_update(session_id, acp.update_user_message_text(next_prompt))
-                await self.prompt(prompt=[TextContentBlock(type="text", text=next_prompt)], session_id=session_id)
+            await self._drain_queued_prompts(state, session_id, conn)
 
         usage = None
         if any(result.get(k) is not None for k in ("prompt_tokens", "completion_tokens", "total_tokens")):
@@ -1002,12 +998,34 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
         await self._send_usage_update(state)
         return PromptResponse(stop_reason="cancelled" if cancelled else "end_turn", usage=usage)
 
+    async def _drain_queued_prompts(self, state: SessionState, session_id: str, conn: Any) -> None:
+        """Run queued prompts while the session is idle. Reached from ``_finish_turn`` and
+        from the slash path after a state-mutating command releases ``command_op``."""
+        while True:
+            with state.runtime_lock:
+                if state.is_running or state.command_op or not state.queued_prompts:
+                    return
+                next_prompt = state.queued_prompts.pop(0)
+            if conn:
+                await conn.session_update(session_id, acp.update_user_message_text(next_prompt))
+            await self.prompt(prompt=[TextContentBlock(type="text", text=next_prompt)], session_id=session_id)
+
     # ---- Session settings (ACP protocol methods) -----------------------------
 
     async def set_session_model(self, model_id: str, session_id: str, **kwargs: Any) -> SetSessionModelResponse | None:
         """Switch the model for a session (called by ACP protocol)."""
         state = await asyncio.to_thread(self.session_manager.get_session, session_id)
-        if state:
+        if state is None:
+            logger.warning("Session %s: model switch requested for missing session", session_id)
+            return None
+        # The picker swaps state.agent wholesale; mid-turn that strands the running agent and
+        # makes _finish_turn emit a spurious compression-rotation update. Same exclusion as
+        # the /model slash command.
+        with state.runtime_lock:
+            if state.is_running or state.command_op:
+                raise acp.RequestError(-32603, "Session is busy; switch models while the session is idle")
+            state.command_op = True
+        try:
             # switch_model() does synchronous network I/O (models.dev, custom-endpoint probes,
             # ~10 s cold) — off the loop, like the gateway, so other ACP sessions keep flowing.
             try:
@@ -1020,12 +1038,14 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
                 # (disabled provider, context window below the floor) stays on the -32603 path.
                 from acp.exceptions import RequestError
                 raise RequestError.invalid_params({"details": str(exc)}) from exc
-            logger.info(
-                "Session %s: model switched to %s via provider %s", session_id, resolved_model, requested_provider
-            )
-            return SetSessionModelResponse()
-        logger.warning("Session %s: model switch requested for missing session", session_id)
-        return None
+        finally:
+            with state.runtime_lock:
+                state.command_op = False
+            await self._drain_queued_prompts(state, session_id, self._conn)
+        logger.info(
+            "Session %s: model switched to %s via provider %s", session_id, resolved_model, requested_provider
+        )
+        return SetSessionModelResponse()
 
     async def set_session_mode(self, mode_id: str, session_id: str, **kwargs: Any) -> SetSessionModeResponse | None:
         """Persist the editor-requested mode so ACP clients do not fail on mode switches."""

acp_adapter/session.py

@@ -139,6 +139,11 @@ class SessionState:
     history: List[Dict[str, Any]] = field(default_factory=list)
     cancel_event: Any = None  # threading.Event
     is_running: bool = False
+    # A state-mutating slash command (/reset, /compress, /model) is in flight. Turn claims
+    # must queue behind it: /compress's LLM call and /model's agent rebuild take seconds, so
+    # a bare is_running check in the slash thread would leave a check-then-act window where
+    # a prompt claims the turn mid-mutation.
+    command_op: bool = False
     queued_prompts: List[str] = field(default_factory=list)
     runtime_lock: Any = field(default_factory=threading.Lock)
     current_prompt_text: str = ""

tests/acp_adapter/test_acp_commands.py

@@ -1,5 +1,7 @@
+import asyncio
 import sys
 from types import ModuleType, SimpleNamespace
+from unittest.mock import patch
 
 import pytest
 from acp.schema import TextContentBlock
@@ -135,6 +137,85 @@ async def test_acp_steer_slash_command_injects_into_running_agent():
     assert fake.runs == []
 
 
+@pytest.mark.asyncio
+async def test_acp_reset_rejected_while_turn_running():
+    """prompt() dispatches slash commands before the is_running claim; /reset
+    must be refused there rather than clearing state.history mid-turn."""
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    state.is_running = True
+    state.history = [{"role": "user", "content": "earlier"}]
+
+    response = await acp_agent.prompt(
+        session_id=state.session_id,
+        prompt=[TextContentBlock(type="text", text="/reset")],
+    )
+
+    assert response.stop_reason == "end_turn"
+    assert state.history == [{"role": "user", "content": "earlier"}]
+    assert fake.runs == []
+    assert state.queued_prompts == []
+
+
+@pytest.mark.asyncio
+async def test_acp_compress_rejected_while_turn_running():
+    """Mid-turn /compress would compress a torn history and rebind
+    state.history while the live turn still appends to the old list."""
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    state.is_running = True
+    state.history = [{"role": "user", "content": "earlier"}]
+    sentinel_db = object()
+    fake._session_db = sentinel_db
+    fake._cached_system_prompt = "sys"
+    fake._compress_context = lambda *a, **k: ([{"role": "user", "content": "summary"}], "new-sys")
+
+    response = await acp_agent.prompt(
+        session_id=state.session_id,
+        prompt=[TextContentBlock(type="text", text="/compress")],
+    )
+
+    assert response.stop_reason == "end_turn"
+    assert state.history == [{"role": "user", "content": "earlier"}]
+    assert fake._session_db is sentinel_db
+    assert fake.runs == []
+
+
+@pytest.mark.asyncio
+async def test_acp_prompt_during_mutating_command_queues_then_runs():
+    """The command_op flag closes the check-then-act window: a prompt arriving while
+    /reset is mid-flight must queue behind it, then run on the cleared history."""
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    loop = asyncio.get_running_loop()
+    state.history = [{"role": "user", "content": "earlier"}]
+
+    # Inject inside _cmd_reset: at that point _handle_slash_command already holds
+    # command_op, so the concurrent prompt must queue rather than claim the turn.
+    orig_reset = acp_agent._cmd_reset
+
+    def patched_reset(args, st):
+        fut = asyncio.run_coroutine_threadsafe(
+            acp_agent.prompt(
+                session_id=st.session_id,
+                prompt=[TextContentBlock(type="text", text="follow-up")],
+            ), loop)
+        fut.result(timeout=10)
+        return orig_reset(args, st)
+
+    with patch.object(acp_agent, "_cmd_reset", patched_reset):
+        response = await acp_agent.prompt(
+            session_id=state.session_id,
+            prompt=[TextContentBlock(type="text", text="/reset")],
+        )
+
+    assert response.stop_reason == "end_turn"
+    # The follow-up queued behind the op, then ran on the cleared history.
+    assert fake.runs == ["follow-up"]
+    assert state.history == [
+        {"role": "user", "content": "follow-up"},
+        {"role": "assistant", "content": "ran: follow-up"},
+    ]
+    assert state.command_op is False
+
+
 
 
 

tests/acp_adapter/test_acp_dashboard_model_switch_validation.py

@@ -34,8 +34,10 @@ def _acp_agent():
 
 
 def _state(**agent_attrs):
+    import threading
     return types.SimpleNamespace(
         session_id="s1", cwd=".", model="claude-sonnet-5",
+        is_running=False, command_op=False, queued_prompts=[], runtime_lock=threading.Lock(),
         agent=types.SimpleNamespace(
             provider="anthropic", base_url="https://api.anthropic.com", api_key="k", **agent_attrs))
 
@@ -99,6 +101,29 @@ def test_acp_set_session_model_runs_switch_model_off_the_event_loop(monkeypatch)
     assert seen["thread"] is not loop_thread
 
 
+def test_acp_set_session_model_rejected_while_turn_running(monkeypatch):
+    """The picker swaps state.agent wholesale; mid-turn that strands the running agent and
+    makes _finish_turn emit a spurious compression-rotation update."""
+    import acp
+    import asyncio
+
+    called = {}
+    monkeypatch.setattr(
+        "hermes_cli.model_switch.switch_model",
+        lambda **kw: called.setdefault("hit", kw))
+    agent, made = _acp_agent()
+    state = _state()
+    state.is_running = True
+    agent.session_manager.get_session = lambda sid: state
+
+    with pytest.raises(acp.RequestError):
+        asyncio.run(agent.set_session_model("anthropic:claude-sonnet-5", "s1"))
+
+    assert called == {} and made == {}  # no resolution, no rebuild
+    assert state.model == "claude-sonnet-5"
+    assert state.command_op is False
+
+
 def test_acp_switch_model_carries_the_live_agent_toolsets_into_the_rebuild(monkeypatch):
     """Regression for #42719: ACP-provided MCP servers live only on the running agent's toolsets
     (``_register_session_mcp_servers``); a rebuild that re-derived them from config.yaml dropped

tests/acp_adapter/test_server.py

@@ -596,6 +596,69 @@ class TestSlashCommands:
         assert "cleared" in result.lower()
         assert len(state.history) == 0
 
+    def test_reset_rejected_mid_turn(self, agent, mock_manager):
+        """Slash dispatch runs on a worker thread beside the live turn; clearing
+        state.history underneath run_conversation tears the running turn."""
+        state = self._make_state(mock_manager)
+        state.history = [{"role": "user", "content": "hello"}]
+        state.is_running = True
+        result = agent._handle_slash_command("/reset", state)
+        assert "busy" in result
+        assert state.history == [{"role": "user", "content": "hello"}]
+
+    def test_compress_rejected_mid_turn(self, agent, mock_manager):
+        """Mid-turn /compress must not rebind state.history or null
+        agent._session_db underneath the running turn."""
+        state = self._make_state(mock_manager)
+        state.history = [{"role": "user", "content": "one"}]
+        state.is_running = True
+        state.agent._compress_context = MagicMock()
+        sentinel_db = object()
+        state.agent._session_db = sentinel_db
+        result = agent._handle_slash_command("/compress", state)
+        assert "busy" in result
+        state.agent._compress_context.assert_not_called()
+        assert state.agent._session_db is sentinel_db
+        assert state.history == [{"role": "user", "content": "one"}]
+
+    def test_model_switch_rejected_mid_turn(self, agent, mock_manager):
+        """/model swaps state.agent wholesale; mid-turn the switch strands the
+        running agent and corrupts the turn-finish provenance check."""
+        state = self._make_state(mock_manager)
+        old_agent = state.agent
+        state.is_running = True
+        result = agent._handle_slash_command("/model gpt-5", state)
+        assert "busy" in result
+        assert state.agent is old_agent
+
+    def test_mutating_command_rejected_during_command_op(self, agent, mock_manager):
+        """A second mutating command must not interleave with one in flight."""
+        state = self._make_state(mock_manager)
+        state.command_op = True
+        state.history = [{"role": "user", "content": "hello"}]
+        result = agent._handle_slash_command("/reset", state)
+        assert "busy" in result
+        assert state.history == [{"role": "user", "content": "hello"}]
+        assert state.command_op is True
+
+    def test_command_op_released_on_handler_error(self, agent, mock_manager):
+        state = self._make_state(mock_manager)
+        state.history = [{"role": "user", "content": "hello"}]
+        with patch.object(agent, "_cmd_reset", side_effect=RuntimeError("boom")):
+            result = agent._handle_slash_command("/reset", state)
+        assert "Error executing /reset" in result
+        assert state.command_op is False
+
+    def test_prompt_queues_without_redirect_during_command_op(self, agent, mock_manager):
+        """A prompt arriving mid-op queues; redirect only applies to a live turn."""
+        state = self._make_state(mock_manager)
+        state.command_op = True
+        result = agent._claim_turn_or_queue(state, state.session_id, "next", "next", True)
+        assert "Queued" in result
+        assert state.queued_prompts == ["next"]
+        state.agent.redirect.assert_not_called()
+        assert state.is_running is False
+