fix: cancel stopped agent runs immediately (#9602)

* fix: cancel stopped agent runs immediately

* refactor: simplify stopped run history

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -104,6 +104,7 @@ class _ToolExecutionInterrupted(Exception):
 
 
 ToolExecutorResultT = T.TypeVar("ToolExecutorResultT")
+AwaitableResultT = T.TypeVar("AwaitableResultT")
 
 
 class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
@@ -112,10 +113,8 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
     EMPTY_OUTPUT_RETRY_ATTEMPTS = 3
     EMPTY_OUTPUT_RETRY_WAIT_MIN_S = 1
     EMPTY_OUTPUT_RETRY_WAIT_MAX_S = 4
-    USER_INTERRUPTION_MESSAGE = (
-        "[SYSTEM: User actively interrupted the response generation. "
-        "Partial output before interruption is preserved.]"
-    )
+    USER_INTERRUPTION_REQUEST = "Stop output."
+    USER_INTERRUPTION_MESSAGE = "Output stopped."
     FOLLOW_UP_NOTICE_TEMPLATE = (
         "\n\n[SYSTEM NOTICE] User sent follow-up messages while tool execution "
         "was in progress. Prioritize these follow-up instructions in your next "
@@ -459,6 +458,45 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
             preview = preview[:next_len]
         return preview
 
+    async def _await_or_stop(
+        self,
+        awaitable: T.Awaitable[AwaitableResultT],
+    ) -> AwaitableResultT | None:
+        """Await work while allowing a stop request to cancel it immediately.
+
+        Args:
+            awaitable: Provider or context-processing operation to execute.
+
+        Returns:
+            The operation result, or None when user cancellation wins the race.
+
+        Raises:
+            asyncio.CancelledError: If the outer Agent task is cancelled.
+            Exception: Any exception raised by the awaited operation.
+        """
+        operation_task = asyncio.create_task(awaitable)
+        abort_task = asyncio.create_task(self._abort_signal.wait())
+        try:
+            done, _ = await asyncio.wait(
+                {operation_task, abort_task},
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+            if abort_task in done:
+                if not operation_task.done():
+                    operation_task.cancel()
+                await asyncio.gather(operation_task, return_exceptions=True)
+                return None
+            return operation_task.result()
+        except asyncio.CancelledError:
+            if not operation_task.done():
+                operation_task.cancel()
+            await asyncio.gather(operation_task, return_exceptions=True)
+            raise
+        finally:
+            if not abort_task.done():
+                abort_task.cancel()
+            await asyncio.gather(abort_task, return_exceptions=True)
+
     async def _iter_llm_responses(
         self, *, include_model: bool = True
     ) -> T.AsyncGenerator[LLMResponse, None]:
@@ -476,10 +514,21 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
             payload["model"] = self.req.model
         if self.streaming:
             stream = self.provider.text_chat_stream(**payload)
-            async for resp in stream:  # type: ignore
-                yield resp
+            try:
+                while True:
+                    try:
+                        resp = await self._await_or_stop(anext(stream))  # type: ignore
+                    except StopAsyncIteration:
+                        return
+                    if resp is None:
+                        return
+                    yield resp
+            finally:
+                await self._close_executor(stream)
         else:
-            yield await self.provider.text_chat(**payload)
+            resp = await self._await_or_stop(self.provider.text_chat(**payload))
+            if resp is not None:
+                yield resp
 
     async def _iter_llm_responses_with_fallback(
         self,
@@ -491,6 +540,8 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
         last_err_response: LLMResponse | None = None
 
         for idx, candidate in enumerate(candidates):
+            if self._is_stop_requested():
+                return
             candidate_id = candidate.provider_config.get("id", "<unknown>")
             is_last_candidate = idx == total_candidates - 1
             if idx > 0:
@@ -513,6 +564,8 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                 )
 
                 async for attempt in retrying:
+                    if self._is_stop_requested():
+                        return
                     has_stream_output = False
                     with attempt:
                         try:
@@ -556,6 +609,8 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                                     self.EMPTY_OUTPUT_RETRY_ATTEMPTS,
                                 )
                             raise
+                    if self._is_stop_requested():
+                        return
             except Exception as exc:  # noqa: BLE001
                 last_exception = exc
                 logger.warning(
@@ -749,9 +804,16 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
         # Process request-time context before sending it to the provider.
         token_usage = self.req.conversation.token_usage if self.req.conversation else 0
         self._simple_print_message_role("[BefCompact]", self.run_context.messages)
-        self.run_context.messages = await self.request_context_manager.process(
-            self.run_context.messages, trusted_token_usage=token_usage
+        processed_messages = await self._await_or_stop(
+            self.request_context_manager.process(
+                self.run_context.messages,
+                trusted_token_usage=token_usage,
+            )
         )
+        if processed_messages is None:
+            yield await self._finalize_aborted_step()
+            return
+        self.run_context.messages = processed_messages
         self._simple_print_message_role("[AftCompact]", self.run_context.messages)
 
         async for llm_response in self._iter_llm_responses_with_fallback():
@@ -781,12 +843,6 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                         ),
                     )
                 if self._is_stop_requested():
-                    llm_resp_result = LLMResponse(
-                        role="assistant",
-                        completion_text=self.USER_INTERRUPTION_MESSAGE,
-                        reasoning_content=llm_response.reasoning_content,
-                        reasoning_signature=llm_response.reasoning_signature,
-                    )
                     break
                 continue
             llm_resp_result = llm_response
@@ -814,14 +870,11 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
             )
             break  # got final response
 
-        if not llm_resp_result:
-            if self._is_stop_requested():
-                llm_resp_result = LLMResponse(role="assistant", completion_text="")
-            else:
-                return
-
         if self._is_stop_requested():
-            yield await self._finalize_aborted_step(llm_resp_result)
+            yield await self._finalize_aborted_step()
+            return
+
+        if not llm_resp_result:
             return
 
         # 处理 LLM 响应
@@ -875,6 +928,9 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
         if llm_resp.tools_call_name:
             if self.tool_schema_mode == "skills_like":
                 requery_resp, _ = await self._resolve_tool_exec(llm_resp)
+                if self._is_stop_requested():
+                    yield await self._finalize_aborted_step()
+                    return
                 if not requery_resp.tools_call_name:
                     llm_resp = requery_resp
                     logger.warning(
@@ -934,7 +990,7 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                             data=AgentResponseData(chain=chain),
                         )
             except _ToolExecutionInterrupted:
-                yield await self._finalize_aborted_step(llm_resp)
+                yield await self._finalize_aborted_step()
                 return
 
             # 将结果添加到上下文中
@@ -1358,15 +1414,17 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
             )
             if param_subset.tools and tool_names:
                 contexts = self._build_tool_requery_context(tool_names)
-                requery_resp = await self.provider.text_chat(
-                    contexts=self._sanitize_contexts_for_provider(contexts),
-                    func_tool=param_subset,
-                    model=self.req.model,
-                    session_id=self.req.session_id,
-                    extra_user_content_parts=self.req.extra_user_content_parts,
-                    # tool_choice="required",
-                    abort_signal=self._abort_signal,
-                    request_max_retries=self.request_max_retries,
+                requery_resp = await self._await_or_stop(
+                    self.provider.text_chat(
+                        contexts=self._sanitize_contexts_for_provider(contexts),
+                        func_tool=param_subset,
+                        model=self.req.model,
+                        session_id=self.req.session_id,
+                        extra_user_content_parts=self.req.extra_user_content_parts,
+                        # tool_choice="required",
+                        abort_signal=self._abort_signal,
+                        request_max_retries=self.request_max_retries,
+                    )
                 )
                 if requery_resp:
                     llm_resp = requery_resp
@@ -1386,15 +1444,19 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                         tool_names,
                         extra_instruction=self.SKILLS_LIKE_REQUERY_REPAIR_INSTRUCTION,
                     )
-                    repair_resp = await self.provider.text_chat(
-                        contexts=self._sanitize_contexts_for_provider(repair_contexts),
-                        func_tool=param_subset,
-                        model=self.req.model,
-                        session_id=self.req.session_id,
-                        extra_user_content_parts=self.req.extra_user_content_parts,
-                        # tool_choice="required",
-                        abort_signal=self._abort_signal,
-                        request_max_retries=self.request_max_retries,
+                    repair_resp = await self._await_or_stop(
+                        self.provider.text_chat(
+                            contexts=self._sanitize_contexts_for_provider(
+                                repair_contexts
+                            ),
+                            func_tool=param_subset,
+                            model=self.req.model,
+                            session_id=self.req.session_id,
+                            extra_user_content_parts=self.req.extra_user_content_parts,
+                            # tool_choice="required",
+                            abort_signal=self._abort_signal,
+                            request_max_retries=self.request_max_retries,
+                        )
                     )
                     if repair_resp:
                         llm_resp = repair_resp
@@ -1418,38 +1480,33 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
     def get_final_llm_resp(self) -> LLMResponse | None:
         return self.final_llm_resp
 
-    async def _finalize_aborted_step(
-        self,
-        llm_resp: LLMResponse | None = None,
-    ) -> AgentResponse:
+    async def _finalize_aborted_step(self) -> AgentResponse:
         logger.info("Agent execution was requested to stop by user.")
-        if llm_resp is None:
-            llm_resp = LLMResponse(role="assistant", completion_text="")
-        if llm_resp.role != "assistant":
-            llm_resp = LLMResponse(
-                role="assistant",
-                completion_text=self.USER_INTERRUPTION_MESSAGE,
-            )
-        self.final_llm_resp = llm_resp
+
+        self.run_context.messages.extend(
+            [
+                Message(
+                    role="user",
+                    content=[TextPart(text=self.USER_INTERRUPTION_REQUEST)],
+                ),
+                Message(
+                    role="assistant",
+                    content=[TextPart(text=self.USER_INTERRUPTION_MESSAGE)],
+                ),
+            ]
+        )
+
+        interrupted_resp = LLMResponse(
+            role="assistant",
+            completion_text=self.USER_INTERRUPTION_MESSAGE,
+        )
+        self.final_llm_resp = interrupted_resp
         self._aborted = True
         self._transition_state(AgentState.DONE)
         self.stats.end_time = time.time()
 
-        parts = []
-        if llm_resp.reasoning_content is not None or llm_resp.reasoning_signature:
-            parts.append(
-                ThinkPart(
-                    think=llm_resp.reasoning_content or "",
-                    encrypted=llm_resp.reasoning_signature,
-                )
-            )
-        if llm_resp.completion_text:
-            parts.append(TextPart(text=llm_resp.completion_text))
-        if parts:
-            self.run_context.messages.append(Message(role="assistant", content=parts))
-
         try:
-            await self.agent_hooks.on_agent_done(self.run_context, llm_resp)
+            await self.agent_hooks.on_agent_done(self.run_context, interrupted_resp)
         except Exception as e:
             logger.error(f"Error in on_agent_done hook: {e}", exc_info=True)
 

astrbot/core/pipeline/process_stage/follow_up.py

@@ -7,6 +7,7 @@ from astrbot import logger
 from astrbot.core.agent.runners.tool_loop_agent_runner import FollowUpTicket
 from astrbot.core.astr_agent_run_util import AgentRunner
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
+from astrbot.core.utils.active_event_registry import active_event_registry
 
 _ACTIVE_AGENT_RUNNERS: dict[str, AgentRunner] = {}
 _FOLLOW_UP_ORDER_STATE: dict[str, dict[str, object]] = {}
@@ -37,11 +38,24 @@ def _event_follow_up_text(event: AstrMessageEvent) -> str:
 
 def register_active_runner(umo: str, runner: AgentRunner) -> None:
     _ACTIVE_AGENT_RUNNERS[umo] = runner
+    runner_event = getattr(getattr(runner.run_context, "context", None), "event", None)
+    if runner_event is not None:
+        active_event_registry.register_agent_stop_callback(
+            runner_event,
+            runner.request_stop,
+        )
 
 
 def unregister_active_runner(umo: str, runner: AgentRunner) -> None:
     if _ACTIVE_AGENT_RUNNERS.get(umo) is runner:
         _ACTIVE_AGENT_RUNNERS.pop(umo, None)
+        runner_event = getattr(
+            getattr(runner.run_context, "context", None),
+            "event",
+            None,
+        )
+        if runner_event is not None:
+            active_event_registry.unregister_agent_stop_callback(runner_event)
 
 
 def _get_follow_up_order_state(umo: str) -> dict[str, object]:

astrbot/core/utils/active_event_registry.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections import defaultdict
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -15,16 +16,39 @@ class ActiveEventRegistry:
 
     def __init__(self) -> None:
         self._events: dict[str, set[AstrMessageEvent]] = defaultdict(set)
+        self._agent_stop_callbacks: dict[AstrMessageEvent, Callable[[], None]] = {}
 
     def register(self, event: AstrMessageEvent) -> None:
         self._events[event.unified_msg_origin].add(event)
 
     def unregister(self, event: AstrMessageEvent) -> None:
         umo = event.unified_msg_origin
+        self._agent_stop_callbacks.pop(event, None)
         self._events[umo].discard(event)
         if not self._events[umo]:
             del self._events[umo]
 
+    def register_agent_stop_callback(
+        self,
+        event: AstrMessageEvent,
+        callback: Callable[[], None],
+    ) -> None:
+        """Register immediate Agent cancellation for an active event.
+
+        Args:
+            event: Event that owns the active Agent execution.
+            callback: Callback that requests cancellation of the active execution.
+        """
+        self._agent_stop_callbacks[event] = callback
+
+    def unregister_agent_stop_callback(self, event: AstrMessageEvent) -> None:
+        """Remove the Agent cancellation callback for an event.
+
+        Args:
+            event: Event whose active Agent execution has finished.
+        """
+        self._agent_stop_callbacks.pop(event, None)
+
     def stop_all(
         self,
         umo: str,
@@ -60,6 +84,9 @@ class ActiveEventRegistry:
         for event in list(self._events.get(umo, [])):
             if event is not exclude:
                 event.set_extra("agent_stop_requested", True)
+                callback = self._agent_stop_callbacks.get(event)
+                if callback:
+                    callback()
                 count += 1
         return count
 

tests/test_conversation_checkpoint.py

@@ -211,6 +211,54 @@ async def test_failed_llm_response_persists_checkpoint_for_retry():
     )
 
 
+@pytest.mark.asyncio
+async def test_aborted_response_persists_synthetic_stop_turn():
+    conversation_manager = AsyncMock()
+    stage = InternalAgentSubStage()
+    stage.conv_manager = conversation_manager
+    event = SimpleNamespace(
+        unified_msg_origin="webchat:FriendMessage:test",
+        get_extra=lambda _key: None,
+    )
+    request = ProviderRequest(
+        conversation=Conversation(
+            platform_id="webchat",
+            user_id="webchat:FriendMessage:test",
+            cid="conversation-1",
+        )
+    )
+
+    await stage._save_to_history(
+        event,
+        request,
+        LLMResponse(role="assistant", completion_text="Output stopped."),
+        [
+            Message(role="user", content="Explain the result."),
+            Message(role="user", content=[TextPart(text="Stop output.")]),
+            Message(role="assistant", content=[TextPart(text="Output stopped.")]),
+        ],
+        runner_stats=None,
+        user_aborted=True,
+    )
+
+    conversation_manager.update_conversation.assert_awaited_once_with(
+        "webchat:FriendMessage:test",
+        "conversation-1",
+        history=[
+            {"role": "user", "content": "Explain the result."},
+            {
+                "role": "user",
+                "content": [{"type": "text", "text": "Stop output."}],
+            },
+            {
+                "role": "assistant",
+                "content": [{"type": "text", "text": "Output stopped."}],
+            },
+        ],
+        token_usage=None,
+    )
+
+
 @pytest.mark.asyncio
 async def test_terminal_tool_result_persists_history_without_checkpoint():
     conversation_manager = AsyncMock()

tests/test_tool_loop_agent_runner.py

@@ -249,6 +249,33 @@ class MockAbortableStreamProvider(MockProvider):
         )
 
 
+class MockBlockingProvider(MockProvider):
+    """Provider that records cancellation while waiting for its first response."""
+
+    def __init__(self):
+        super().__init__()
+        self.started = asyncio.Event()
+        self.cancelled = asyncio.Event()
+
+    async def text_chat(self, **kwargs) -> LLMResponse:
+        self.started.set()
+        try:
+            await asyncio.Future()
+        except asyncio.CancelledError:
+            self.cancelled.set()
+            raise
+
+    async def text_chat_stream(self, **kwargs):
+        self.started.set()
+        try:
+            await asyncio.Future()
+        except asyncio.CancelledError:
+            self.cancelled.set()
+            raise
+        if False:
+            yield LLMResponse(role="assistant")
+
+
 class MockToolCallProvider(MockProvider):
     def __init__(self, tool_name: str, tool_args: dict[str, str] | None = None):
         super().__init__()
@@ -1277,7 +1304,7 @@ async def test_empty_output_retries_exhausted_then_uses_fallback_provider(
 
 
 @pytest.mark.asyncio
-async def test_stop_signal_returns_aborted_and_persists_partial_message(
+async def test_stop_signal_returns_aborted_and_discards_partial_message(
     runner, provider_request, mock_tool_executor, mock_hooks
 ):
     provider = MockAbortableStreamProvider()
@@ -1307,9 +1334,70 @@ async def test_stop_signal_returns_aborted_and_persists_partial_message(
     final_resp = runner.get_final_llm_resp()
     assert final_resp is not None
     assert final_resp.role == "assistant"
-    # When interrupted, the runner replaces completion_text with a system message
-    assert "interrupted" in final_resp.completion_text.lower()
-    assert runner.run_context.messages[-1].role == "assistant"
+    assert final_resp.completion_text == runner.USER_INTERRUPTION_MESSAGE
+    assert [message.role for message in runner.run_context.messages[-2:]] == [
+        "user",
+        "assistant",
+    ]
+    assert runner.run_context.messages[-2].content == [
+        TextPart(text=runner.USER_INTERRUPTION_REQUEST)
+    ]
+    assert runner.run_context.messages[-1].content == [
+        TextPart(text=runner.USER_INTERRUPTION_MESSAGE)
+    ]
+    assert all(
+        message.content != [TextPart(text="partial ")]
+        for message in runner.run_context.messages
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_stop_cancels_provider_before_first_response(
+    streaming,
+    runner,
+    provider_request,
+    mock_tool_executor,
+    mock_hooks,
+):
+    """Stop must cancel blocked streaming and non-streaming Provider requests."""
+    provider = MockBlockingProvider()
+    await runner.reset(
+        provider=provider,
+        request=provider_request,
+        run_context=ContextWrapper(context=None),
+        tool_executor=mock_tool_executor,
+        agent_hooks=mock_hooks,
+        streaming=streaming,
+    )
+
+    step_iter = runner.step()
+    pending_response = asyncio.create_task(anext(step_iter))
+    await asyncio.wait_for(provider.started.wait(), timeout=1)
+
+    runner.request_stop()
+
+    response = await asyncio.wait_for(pending_response, timeout=1)
+    assert response.type == "aborted"
+    await asyncio.wait_for(provider.cancelled.wait(), timeout=1)
+    assert runner.was_aborted() is True
+    assert runner.done() is True
+    final_resp = runner.get_final_llm_resp()
+    assert final_resp is not None
+    assert final_resp.completion_text == runner.USER_INTERRUPTION_MESSAGE
+    assert [message.role for message in runner.run_context.messages[-2:]] == [
+        "user",
+        "assistant",
+    ]
+    assert runner.run_context.messages[-2].content == [
+        TextPart(text=runner.USER_INTERRUPTION_REQUEST)
+    ]
+    assert runner.run_context.messages[-1].content == [
+        TextPart(text=runner.USER_INTERRUPTION_MESSAGE)
+    ]
+
+    with pytest.raises(StopAsyncIteration):
+        await step_iter.__anext__()
 
 
 @pytest.mark.asyncio
@@ -1357,6 +1445,13 @@ async def test_stop_interrupts_pending_subagent_handoff(mock_hooks):
     assert aborted_resp.type == "aborted"
     assert runner.was_aborted() is True
     assert subagent_context.cancelled is True
+    final_resp = runner.get_final_llm_resp()
+    assert final_resp is not None
+    assert final_resp.completion_text == runner.USER_INTERRUPTION_MESSAGE
+    assert [message.role for message in runner.run_context.messages[-2:]] == [
+        "user",
+        "assistant",
+    ]
 
     with pytest.raises(StopAsyncIteration):
         await step_iter.__anext__()
@@ -1409,6 +1504,13 @@ async def test_stop_interrupts_pending_regular_tool(mock_hooks):
     assert aborted_resp.type == "aborted"
     assert runner.was_aborted() is True
     assert tool_state.cancelled is True
+    final_resp = runner.get_final_llm_resp()
+    assert final_resp is not None
+    assert final_resp.completion_text == runner.USER_INTERRUPTION_MESSAGE
+    assert [message.role for message in runner.run_context.messages[-2:]] == [
+        "user",
+        "assistant",
+    ]
 
     with pytest.raises(StopAsyncIteration):
         await step_iter.__anext__()
@@ -1862,6 +1964,13 @@ async def test_follow_up_rejected_and_runner_stops_without_execution(
     # Verify runner stopped gracefully
     assert runner.done()
     assert runner.was_aborted()
+    final_resp = runner.get_final_llm_resp()
+    assert final_resp is not None
+    assert final_resp.completion_text == runner.USER_INTERRUPTION_MESSAGE
+    assert [message.role for message in runner.run_context.messages[-2:]] == [
+        "user",
+        "assistant",
+    ]
 
     # No tool execution should have occurred
     assert provider_request.tool_calls_result is None

tests/unit/test_active_event_registry.py

@@ -0,0 +1,80 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from astrbot.core.pipeline.process_stage.follow_up import (
+    register_active_runner,
+    unregister_active_runner,
+)
+from astrbot.core.utils.active_event_registry import (
+    ActiveEventRegistry,
+    active_event_registry,
+)
+
+
+class StubEvent:
+    """Minimal event implementation used by ActiveEventRegistry tests."""
+
+    def __init__(self, umo: str) -> None:
+        self.unified_msg_origin = umo
+        self.extras: dict[str, object] = {}
+
+    def set_extra(self, key: str, value: object) -> None:
+        """Store an event extra.
+
+        Args:
+            key: Extra field name.
+            value: Extra field value.
+        """
+        self.extras[key] = value
+
+
+def test_request_agent_stop_invokes_registered_callback() -> None:
+    """Agent stop requests immediately invoke the active execution callback."""
+    registry = ActiveEventRegistry()
+    event = StubEvent("webchat:FriendMessage:webchat!alice!session")
+    callback = Mock()
+    registry.register(event)
+    registry.register_agent_stop_callback(event, callback)
+
+    stopped_count = registry.request_agent_stop_all(event.unified_msg_origin)
+
+    assert stopped_count == 1
+    assert event.extras["agent_stop_requested"] is True
+    callback.assert_called_once_with()
+
+
+def test_unregister_removes_agent_stop_callback() -> None:
+    """Unregistered events cannot retain stale Agent cancellation callbacks."""
+    registry = ActiveEventRegistry()
+    event = StubEvent("webchat:FriendMessage:webchat!alice!session")
+    callback = Mock()
+    registry.register(event)
+    registry.register_agent_stop_callback(event, callback)
+
+    registry.unregister(event)
+    stopped_count = registry.request_agent_stop_all(event.unified_msg_origin)
+
+    assert stopped_count == 0
+    callback.assert_not_called()
+
+
+def test_active_runner_wires_immediate_stop_callback() -> None:
+    """Active Runner registration connects registry stop to Runner cancellation."""
+    event = StubEvent("webchat:FriendMessage:webchat!alice!runner-session")
+    runner = SimpleNamespace(
+        run_context=SimpleNamespace(context=SimpleNamespace(event=event)),
+        request_stop=Mock(),
+    )
+    active_event_registry.register(event)
+    register_active_runner(event.unified_msg_origin, runner)
+
+    try:
+        stopped_count = active_event_registry.request_agent_stop_all(
+            event.unified_msg_origin
+        )
+
+        assert stopped_count == 1
+        runner.request_stop.assert_called_once_with()
+    finally:
+        unregister_active_runner(event.unified_msg_origin, runner)
+        active_event_registry.unregister(event)