From 4987f668aeea19765ac38075bf37bfb557e8a542 Mon Sep 17 00:00:00 2001 From: beardthelion Date: Sat, 19 Sep 2026 10:28:12 -0500 Subject: [PATCH] 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 | 20 +++++ acp_adapter/server.py | 54 +++++++++---- acp_adapter/session.py | 5 ++ tests/acp_adapter/test_acp_commands.py | 81 +++++++++++++++++++ ...t_acp_dashboard_model_switch_validation.py | 25 ++++++ tests/acp_adapter/test_server.py | 63 +++++++++++++++ 6 files changed, 231 insertions(+), 17 deletions(-) diff --git a/acp_adapter/commands.py b/acp_adapter/commands.py index 5a79309d92..12bebb01ba 100644 --- a/acp_adapter/commands.py +++ b/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:", ""] diff --git a/acp_adapter/server.py b/acp_adapter/server.py index 4f0affc5a2..16a4b10198 100644 --- a/acp_adapter/server.py +++ b/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.""" diff --git a/acp_adapter/session.py b/acp_adapter/session.py index 0a323c50f5..0011deda18 100644 --- a/acp_adapter/session.py +++ b/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 = "" diff --git a/tests/acp_adapter/test_acp_commands.py b/tests/acp_adapter/test_acp_commands.py index ff51128cc3..60ce83bbd8 100644 --- a/tests/acp_adapter/test_acp_commands.py +++ b/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 + + diff --git a/tests/acp_adapter/test_acp_dashboard_model_switch_validation.py b/tests/acp_adapter/test_acp_dashboard_model_switch_validation.py index 85458aed88..f3247fd8f5 100644 --- a/tests/acp_adapter/test_acp_dashboard_model_switch_validation.py +++ b/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 diff --git a/tests/acp_adapter/test_server.py b/tests/acp_adapter/test_server.py index dad364e7da..24f94c5f81 100644 --- a/tests/acp_adapter/test_server.py +++ b/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 +