fix(acp): ignore idle cancel before next prompt

acp_adapter/server.py

@@ -1211,16 +1211,32 @@ async def resume_session(
     async def cancel(self, session_id: str, **kwargs: Any) -> None:
         state = self.session_manager.get_session(session_id)
         if state and state.cancel_event:
+            should_interrupt = False
             with state.runtime_lock:
-                if state.is_running and state.current_prompt_text:
+                should_interrupt = bool(state.is_running)
+                if should_interrupt and state.current_prompt_text:
                     state.interrupted_prompt_text = state.current_prompt_text
-            state.cancel_event.set()
-            try:
-                if getattr(state, "agent", None) and hasattr(state.agent, "interrupt"):
-                    state.agent.interrupt()
-            except Exception:
-                logger.debug("Failed to interrupt ACP session %s", session_id, exc_info=True)
-            logger.info("Cancelled session %s", session_id)
+            if should_interrupt:
+                state.cancel_event.set()
+                try:
+                    if getattr(state, "agent", None) and hasattr(state.agent, "interrupt"):
+                        state.agent.interrupt()
+                except Exception:
+                    logger.debug("Failed to interrupt ACP session %s", session_id, exc_info=True)
+                logger.info("Cancelled session %s", session_id)
+            else:
+                # Some ACP clients send a best-effort cancel immediately before
+                # submitting the next prompt, even when the previous turn is
+                # already idle. Treat that as a no-op; otherwise a stale
+                # AIAgent interrupt poisons the next prompt and it returns
+                # interrupted_by_user without ever calling the model.
+                state.cancel_event.clear()
+                try:
+                    if getattr(state, "agent", None) and hasattr(state.agent, "clear_interrupt"):
+                        state.agent.clear_interrupt()
+                except Exception:
+                    logger.debug("Failed to clear idle ACP interrupt for %s", session_id, exc_info=True)
+                logger.info("Ignored idle cancel for session %s", session_id)
 
     async def fork_session(
         self,

tests/acp_adapter/test_acp_commands.py

@@ -18,12 +18,25 @@ def __init__(self):
         self.valid_tool_names = set()
         self.steers = []
         self.runs = []
+        self.interrupted = False
+        self.interrupt_calls = 0
+        self.clear_interrupt_calls = 0
 
     def steer(self, text):
         self.steers.append(text)
         return True
 
+    def interrupt(self):
+        self.interrupted = True
+        self.interrupt_calls += 1
+
+    def clear_interrupt(self):
+        self.interrupted = False
+        self.clear_interrupt_calls += 1
+
     def run_conversation(self, *, user_message, conversation_history, task_id, **kwargs):
+        if self.interrupted:
+            return {"final_response": "", "messages": list(conversation_history or []), "interrupted": True}
         self.runs.append(user_message)
         messages = list(conversation_history or [])
         messages.append({"role": "user", "content": user_message})
@@ -196,3 +209,34 @@ async def test_acp_prompt_drains_queued_turns_after_current_run():
     assert state.queued_prompts == []
     agent_messages = [u for _sid, u in conn.updates if getattr(u, "session_update", None) == "agent_message_chunk"]
     assert len(agent_messages) >= 2
+
+
+@pytest.mark.asyncio
+async def test_acp_idle_cancel_does_not_poison_next_prompt():
+    acp_agent, state, fake, _conn = make_agent_and_state()
+
+    await acp_agent.cancel(state.session_id)
+    response = await acp_agent.prompt(
+        session_id=state.session_id,
+        prompt=[TextContentBlock(type="text", text="second question")],
+    )
+
+    assert response.stop_reason == "end_turn"
+    assert fake.runs == ["second question"]
+    assert fake.interrupt_calls == 0
+    assert fake.clear_interrupt_calls == 1
+    assert not state.cancel_event.is_set()
+
+
+@pytest.mark.asyncio
+async def test_acp_running_cancel_still_interrupts_current_prompt():
+    acp_agent, state, fake, _conn = make_agent_and_state()
+    state.is_running = True
+    state.current_prompt_text = "long first question"
+
+    await acp_agent.cancel(state.session_id)
+
+    assert fake.interrupt_calls == 1
+    assert fake.clear_interrupt_calls == 0
+    assert state.cancel_event.is_set()
+    assert state.interrupted_prompt_text == "long first question"