diff --git a/astrbot/core/agent/runners/tool_loop_agent_runner.py b/astrbot/core/agent/runners/tool_loop_agent_runner.py index 88038473af..f9339aef3b 100644 --- a/astrbot/core/agent/runners/tool_loop_agent_runner.py +++ b/astrbot/core/agent/runners/tool_loop_agent_runner.py @@ -112,10 +112,6 @@ 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.]" - ) 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 " @@ -176,8 +172,12 @@ def _get_persona_custom_error_message(self) -> str | None: event = getattr(self.run_context.context, "event", None) return extract_persona_custom_error_message_from_event(event) - async def _complete_with_assistant_response(self, llm_resp: LLMResponse) -> None: + async def _complete_with_assistant_response(self, llm_resp: LLMResponse) -> bool: """Finalize the current step as a plain assistant response with no tool calls.""" + if self._is_stop_requested(): + await self._finalize_aborted_step(llm_resp) + return False + self.final_llm_resp = llm_resp self._transition_state(AgentState.DONE) self.stats.end_time = time.time() @@ -196,11 +196,24 @@ async def _complete_with_assistant_response(self, llm_resp: LLMResponse) -> None logger.warning("LLM returned empty assistant message with no tool calls.") self.run_context.messages.append(Message(role="assistant", content=parts)) + if self._is_stop_requested(): + if self._is_message_from_llm_response( + self.run_context.messages[-1], llm_resp + ): + self.run_context.messages.pop() + await self._finalize_aborted_step(llm_resp) + return False + try: await self.agent_hooks.on_agent_done(self.run_context, llm_resp) except Exception as e: logger.error(f"Error in on_agent_done hook: {e}", exc_info=True) + if self._is_stop_requested(): + self.discard_late_aborted_result() + self._resolve_unconsumed_follow_ups() + return False self._resolve_unconsumed_follow_ups() + return True @override async def reset( @@ -282,6 +295,7 @@ async def reset( self._follow_up_seq = 0 self._last_tool_name: str | None = None self._same_tool_streak = 0 + self._recorded_usages: list[T.Any] = [] # These two are used for tool schema mode handling # We now have two modes: @@ -323,6 +337,16 @@ async def reset( self.stats = AgentStats() self.stats.start_time = time.time() + def _record_llm_usage(self, llm_resp: LLMResponse) -> None: + if not llm_resp.usage: + return + if any(usage is llm_resp.usage for usage in self._recorded_usages): + return + self._recorded_usages.append(llm_resp.usage) + self.stats.token_usage += llm_resp.usage + if self.req.conversation: + self.req.conversation.token_usage = llm_resp.usage.total + def _read_tool_hint(self) -> str: if self.read_tool is not None: return f"`{self.read_tool.name}`" @@ -488,6 +512,9 @@ async def _iter_llm_responses_with_fallback( 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", "") is_last_candidate = idx == total_candidates - 1 if idx > 0: @@ -526,6 +553,8 @@ async def _iter_llm_responses_with_fallback( and not has_stream_output and (not is_last_candidate) ): + if self._is_stop_requested(): + return last_err_response = resp logger.warning( "Chat Model %s returns error response, trying fallback to next provider.", @@ -734,6 +763,10 @@ async def step(self): self._simple_print_message_role("[AftCompact]", self.run_context.messages) async for llm_response in self._iter_llm_responses_with_fallback(): + if self._is_stop_requested(): + llm_resp_result = llm_response + break + if llm_response.is_chunk: if self.stats.time_to_first_token == 0: self.stats.time_to_first_token = time.time() - self.stats.start_time @@ -747,11 +780,17 @@ async def step(self): ), ), ) + if self._is_stop_requested(): + llm_resp_result = llm_response + break if llm_response.result_chain: yield AgentResponse( type="streaming_delta", data=AgentResponseData(chain=llm_response.result_chain), ) + if self._is_stop_requested(): + llm_resp_result = llm_response + break elif llm_response.completion_text: yield AgentResponse( type="streaming_delta", @@ -759,22 +798,18 @@ async def step(self): chain=MessageChain().message(llm_response.completion_text), ), ) + if self._is_stop_requested(): + llm_resp_result = llm_response + break 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, - ) + llm_resp_result = llm_response break continue llm_resp_result = llm_response if not llm_response.is_chunk and llm_response.usage: # only count the token usage of the final response for computation purpose - self.stats.token_usage += llm_response.usage - if self.req.conversation: - self.req.conversation.token_usage = llm_response.usage.total + self._record_llm_usage(llm_response) break # got final response if not llm_resp_result: @@ -809,7 +844,12 @@ async def step(self): return if not llm_resp.tools_call_name: - await self._complete_with_assistant_response(llm_resp) + if not await self._complete_with_assistant_response(llm_resp): + yield AgentResponse( + type="aborted", + data=AgentResponseData(chain=MessageChain(type="aborted")), + ) + return # 返回 LLM 结果 if llm_resp.reasoning_content: @@ -838,11 +878,21 @@ async def step(self): 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(requery_resp) + return if not requery_resp.tools_call_name: llm_resp = requery_resp logger.warning( "skills_like tool re-query returned no tool calls; fallback to assistant response." ) + if not await self._complete_with_assistant_response(llm_resp): + yield AgentResponse( + type="aborted", + data=AgentResponseData(chain=MessageChain(type="aborted")), + ) + return + if llm_resp.reasoning_content: yield AgentResponse( type="llm_result", @@ -864,8 +914,6 @@ async def step(self): chain=MessageChain().message(llm_resp.completion_text), ), ) - - await self._complete_with_assistant_response(llm_resp) return else: llm_resp.tools_call_name = requery_resp.tools_call_name @@ -1332,6 +1380,9 @@ async def _resolve_tool_exec( llm_resp = requery_resp self._sanitize_malformed_tool_calls(llm_resp) + if self._is_stop_requested(): + return llm_resp, subset + # If the re-query still returns no tool calls, and also does not have a meaningful assistant reply, # we consider it as a failure of the LLM to follow the tool-use instruction, # and we will retry once with a stronger instruction that explicitly requires the LLM to either call the tool or give an explanation. @@ -1370,7 +1421,24 @@ def request_stop(self) -> None: self._abort_signal.set() def _is_stop_requested(self) -> bool: - return self._abort_signal.is_set() + if self._abort_signal.is_set(): + return True + + event = getattr(self.run_context.context, "event", None) + if event is None: + return False + + is_stopped = getattr(event, "is_stopped", None) + if callable(is_stopped) and is_stopped(): + return True + + get_extra = getattr(event, "get_extra", None) + if callable(get_extra): + return bool(get_extra("agent_stop_requested")) or bool( + get_extra("agent_user_aborted") + ) + + return False def was_aborted(self) -> bool: return self._aborted @@ -1378,38 +1446,74 @@ def was_aborted(self) -> bool: def get_final_llm_resp(self) -> LLMResponse | None: return self.final_llm_resp + def _build_aborted_llm_response( + self, + llm_resp: LLMResponse | None = None, + ) -> LLMResponse: + """构造不会泄露迟到模型正文的空响应。""" + return LLMResponse( + role="assistant", + completion_text="", + id=getattr(llm_resp, "id", None) if llm_resp else None, + usage=getattr(llm_resp, "usage", None) if llm_resp else None, + ) + + def _is_message_from_llm_response( + self, + message: Message, + llm_resp: LLMResponse | None, + ) -> bool: + if llm_resp is None or message.role != "assistant" or message.tool_calls: + return False + content = message.content + if not isinstance(content, list): + return False + for part in content: + if isinstance(part, TextPart): + if part.text != llm_resp.completion_text: + return False + elif isinstance(part, ThinkPart): + if part.think != (llm_resp.reasoning_content or ""): + return False + else: + return False + return True + + def discard_late_aborted_result(self) -> None: + """丢弃已完成但尚未对用户可见发送的迟到模型正文。""" + original_llm_resp = self.final_llm_resp + if original_llm_resp: + self._record_llm_usage(original_llm_resp) + self.final_llm_resp = self._build_aborted_llm_response(self.final_llm_resp) + self._aborted = True + event = getattr(self.run_context.context, "event", None) + if event is not None: + event.set_extra("agent_user_aborted", True) + event.set_extra("agent_stop_requested", False) + if self.run_context.messages and self._is_message_from_llm_response( + self.run_context.messages[-1], + original_llm_resp, + ): + self.run_context.messages.pop() + async def _finalize_aborted_step( self, llm_resp: LLMResponse | None = None, ) -> 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 + safe_llm_resp = self._build_aborted_llm_response(llm_resp) + self.final_llm_resp = safe_llm_resp + self._record_llm_usage(safe_llm_resp) self._aborted = True + event = getattr(self.run_context.context, "event", None) + if event is not None: + event.set_extra("agent_user_aborted", True) + event.set_extra("agent_stop_requested", False) 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, safe_llm_resp) except Exception as e: logger.error(f"Error in on_agent_done hook: {e}", exc_info=True) diff --git a/astrbot/core/astr_agent_hooks.py b/astrbot/core/astr_agent_hooks.py index 3155257d71..b08238661e 100644 --- a/astrbot/core/astr_agent_hooks.py +++ b/astrbot/core/astr_agent_hooks.py @@ -7,9 +7,29 @@ from astrbot.core.agent.tool import FunctionTool from astrbot.core.astr_agent_context import AstrAgentContext from astrbot.core.pipeline.context_utils import call_event_hook +from astrbot.core.provider.entities import LLMResponse from astrbot.core.star.star_handler import EventType +def _event_requests_agent_stop(event) -> bool: + get_extra = getattr(event, "get_extra", None) + is_stopped = getattr(event, "is_stopped", None) + return ( + (bool(get_extra("agent_user_aborted")) if callable(get_extra) else False) + or (bool(get_extra("agent_stop_requested")) if callable(get_extra) else False) + or (bool(is_stopped()) if callable(is_stopped) else False) + ) + + +def _build_aborted_llm_response(llm_response) -> LLMResponse: + return LLMResponse( + role="assistant", + completion_text="", + id=getattr(llm_response, "id", None) if llm_response else None, + usage=getattr(llm_response, "usage", None) if llm_response else None, + ) + + class MainAgentHooks(BaseAgentRunHooks[AstrAgentContext]): async def on_agent_begin( self, run_context: ContextWrapper[AstrAgentContext] @@ -22,19 +42,34 @@ async def on_agent_begin( async def on_agent_done(self, run_context, llm_response) -> None: # 执行事件钩子 - if llm_response and llm_response.reasoning_content: + event = run_context.context.event + suppress_llm_response = _event_requests_agent_stop(event) + + if ( + not suppress_llm_response + and llm_response + and llm_response.reasoning_content + ): # we will use this in result_decorate stage to inject reasoning content to chain - run_context.context.event.set_extra( - "_llm_reasoning_content", llm_response.reasoning_content + set_extra = getattr(event, "set_extra", None) + if callable(set_extra): + set_extra("_llm_reasoning_content", llm_response.reasoning_content) + + if not suppress_llm_response: + await call_event_hook( + event, + EventType.OnLLMResponseEvent, + llm_response, ) + if _event_requests_agent_stop(event): + set_extra = getattr(event, "set_extra", None) + if callable(set_extra): + set_extra("_llm_reasoning_content", None) + llm_response = _build_aborted_llm_response(llm_response) + await call_event_hook( - run_context.context.event, - EventType.OnLLMResponseEvent, - llm_response, - ) - await call_event_hook( - run_context.context.event, + event, EventType.OnAgentDoneEvent, run_context, llm_response, diff --git a/astrbot/core/astr_agent_run_util.py b/astrbot/core/astr_agent_run_util.py index 465c5f6170..4896727119 100644 --- a/astrbot/core/astr_agent_run_util.py +++ b/astrbot/core/astr_agent_run_util.py @@ -25,7 +25,11 @@ def _should_stop_agent(astr_event) -> bool: - return astr_event.is_stopped() or bool(astr_event.get_extra("agent_stop_requested")) + return ( + astr_event.is_stopped() + or bool(astr_event.get_extra("agent_stop_requested")) + or bool(astr_event.get_extra("agent_user_aborted")) + ) def _truncate_tool_result(text: str, limit: int = 70) -> str: @@ -159,17 +163,8 @@ async def run_agent( agent_runner.request_stop() if resp.type == "aborted": - if can_buffer_llm_result: - merged_chain = _merge_buffered_llm_chains(buffered_llm_chains) - if merged_chain: - astr_event.set_result( - MessageEventResult( - chain=merged_chain.chain, - result_content_type=ResultContentType.LLM_RESULT, - ), - ) - yield merged_chain - astr_event.clear_result() + buffered_llm_chains.clear() + astr_event.clear_result() if not stop_watcher.done(): stop_watcher.cancel() try: @@ -272,16 +267,27 @@ async def run_agent( yield resp.data["chain"] # MessageChain if can_buffer_llm_result and agent_runner.done(): - merged_chain = _merge_buffered_llm_chains(buffered_llm_chains) - if merged_chain: - astr_event.set_result( - MessageEventResult( - chain=merged_chain.chain, - result_content_type=ResultContentType.LLM_RESULT, - ), - ) - yield merged_chain + if _should_stop_agent(astr_event): + buffered_llm_chains.clear() astr_event.clear_result() + discard_late_result = getattr( + agent_runner, + "discard_late_aborted_result", + None, + ) + if callable(discard_late_result): + discard_late_result() + else: + merged_chain = _merge_buffered_llm_chains(buffered_llm_chains) + if merged_chain: + astr_event.set_result( + MessageEventResult( + chain=merged_chain.chain, + result_content_type=ResultContentType.LLM_RESULT, + ), + ) + yield merged_chain + astr_event.clear_result() if not stop_watcher.done(): stop_watcher.cancel() @@ -439,6 +445,11 @@ async def run_live_agent( if queue_item is None: break + if agent_runner.was_aborted() or _should_stop_agent( + agent_runner.run_context.context.event + ): + continue + text = None if isinstance(queue_item, tuple): text, audio_data = queue_item @@ -550,8 +561,12 @@ async def _run_agent_feeder( # 更新 buffer 为剩余部分 buffer = temp_buffer + parts[-1] - # 处理剩余 buffer - if buffer.strip(): + # 处理剩余 buffer。若 stop 已到达，未送出的文本不能再进入 TTS。 + if ( + buffer.strip() + and not agent_runner.was_aborted() + and not _should_stop_agent(agent_runner.run_context.context.event) + ): await text_queue.put(buffer) except Exception as e: @@ -596,13 +611,20 @@ async def _simulated_stream_tts( text = await text_queue.get() if text is None: break + if _should_stop_agent(astr_event): + continue try: audio_path = await tts_provider.get_audio(text) + if _should_stop_agent(astr_event): + continue + if audio_path: with open(audio_path, "rb") as f: audio_data = f.read() + if _should_stop_agent(astr_event): + continue astr_event.track_temporary_local_file(audio_path) await audio_queue.put((text, audio_data)) except Exception as e: diff --git a/astrbot/core/pipeline/context_utils.py b/astrbot/core/pipeline/context_utils.py index 9402ce3e62..93104c7af9 100644 --- a/astrbot/core/pipeline/context_utils.py +++ b/astrbot/core/pipeline/context_utils.py @@ -9,6 +9,25 @@ from astrbot.core.star.star_handler import EventType, star_handlers_registry +def _event_requests_agent_stop(event: AstrMessageEvent) -> bool: + get_extra = getattr(event, "get_extra", None) + return ( + event.is_stopped() + or (bool(get_extra("agent_user_aborted")) if callable(get_extra) else False) + or (bool(get_extra("agent_stop_requested")) if callable(get_extra) else False) + ) + + +def _should_stop_hook_propagation( + event: AstrMessageEvent, hook_type: EventType +) -> bool: + if event.is_stopped(): + return True + return hook_type == EventType.OnLLMResponseEvent and _event_requests_agent_stop( + event + ) + + async def call_handler( event: AstrMessageEvent, handler: T.Callable[..., T.Awaitable[T.Any] | T.AsyncGenerator[T.Any, None]], @@ -90,6 +109,9 @@ async def call_event_hook( plugins_name=event.plugins_name, ) for handler in handlers: + if _should_stop_hook_propagation(event, hook_type): + return True + try: assert inspect.iscoroutinefunction(handler.handler) logger.debug( @@ -99,10 +121,10 @@ async def call_event_hook( except BaseException: logger.error(traceback.format_exc()) - if event.is_stopped(): + if _should_stop_hook_propagation(event, hook_type): logger.info( f"{star_map[handler.handler_module_path].name} - {handler.handler_name} 终止了事件传播。", ) return True - return event.is_stopped() + return _should_stop_hook_propagation(event, hook_type) diff --git a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py index 7b01dd2dc7..75f2bc79dc 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py +++ b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py @@ -320,6 +320,13 @@ async def process( ), ) yield + if agent_runner.was_aborted(): + event.set_result( + MessageEventResult( + chain=MessageChain().chain, + result_content_type=ResultContentType.STREAMING_FINISH, + ), + ) # 保存历史记录 if agent_runner.done() and ( @@ -351,7 +358,14 @@ async def process( ), ) yield - if agent_runner.done(): + if agent_runner.done() and agent_runner.was_aborted(): + event.set_result( + MessageEventResult( + chain=MessageChain().chain, + result_content_type=ResultContentType.STREAMING_FINISH, + ), + ) + elif agent_runner.done(): if final_llm_resp := agent_runner.get_final_llm_resp(): if final_llm_resp.completion_text: chain = ( diff --git a/tests/test_tool_loop_agent_runner.py b/tests/test_tool_loop_agent_runner.py index b4464680fb..48a736e152 100644 --- a/tests/test_tool_loop_agent_runner.py +++ b/tests/test_tool_loop_agent_runner.py @@ -16,12 +16,20 @@ from astrbot.core.agent.hooks import BaseAgentRunHooks from astrbot.core.agent.message import ImageURLPart, Message, TextPart from astrbot.core.agent.run_context import ContextWrapper +from astrbot.core.agent.runners.base import AgentState from astrbot.core.agent.runners.tool_loop_agent_runner import ToolLoopAgentRunner from astrbot.core.agent.tool import FunctionTool, ToolSet from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor from astrbot.core.exceptions import EmptyModelOutputError -from astrbot.core.provider.entities import LLMResponse, ProviderRequest, TokenUsage -from astrbot.core.provider.provider import Provider +from astrbot.core.message.message_event_result import MessageChain +from astrbot.core.provider.entities import ( + LLMResponse, + ProviderMeta, + ProviderRequest, + ProviderType, + TokenUsage, +) +from astrbot.core.provider.provider import Provider, TTSProvider class MockProvider(Provider): @@ -214,6 +222,91 @@ async def text_chat_stream(self, **kwargs): ) +class MockLeakyAbortProvider(MockProvider): + def __init__(self, event=None): + super().__init__() + self.event = event + + async def text_chat(self, **kwargs) -> LLMResponse: + self.call_count += 1 + if self.event is not None: + self.event.set_extra("agent_stop_requested", True) + return LLMResponse( + role="assistant", + completion_text="late completion text", + result_chain=MessageChain().message("late result chain"), + tools_call_name=["late_tool"], + tools_call_args=[{"query": "late"}], + tools_call_ids=["call_late"], + tools_call_extra_content={"call_late": {"extra": "late"}}, + reasoning_content="late reasoning", + reasoning_signature="late signature", + raw_completion=object(), + id="late-id", + usage=TokenUsage(input_other=10, output=5), + ) + + +class MockDelayedTextProvider(MockProvider): + def __init__(self): + super().__init__() + self.started = asyncio.Event() + self.release = asyncio.Event() + + async def text_chat(self, **kwargs) -> LLMResponse: + self.call_count += 1 + self.started.set() + await self.release.wait() + return LLMResponse( + role="assistant", + completion_text="delayed visible text", + usage=TokenUsage(input_other=10, output=5), + ) + + +class MockStopAwareStreamProvider(MockProvider): + async def text_chat_stream(self, **kwargs): + abort_signal = kwargs.get("abort_signal") + if abort_signal is None: + return + await abort_signal.wait() + yield LLMResponse( + role="assistant", + completion_text="late chunk text", + reasoning_content="late chunk reasoning", + is_chunk=True, + ) + yield LLMResponse( + role="assistant", + completion_text="late final text", + is_chunk=False, + usage=TokenUsage(input_other=10, output=5), + ) + + +class MockDelayedTTSProvider(TTSProvider): + def __init__(self, audio_path: Path): + super().__init__({"type": "test_tts", "id": "test_tts"}, {}) + self.audio_path = audio_path + self.started = asyncio.Event() + self.release = asyncio.Event() + self.call_count = 0 + + async def get_audio(self, text: str) -> str: + self.call_count += 1 + self.started.set() + await self.release.wait() + return str(self.audio_path) + + def meta(self) -> ProviderMeta: + return ProviderMeta( + id="test_tts", + model=None, + type="test_tts", + provider_type=ProviderType.TEXT_TO_SPEECH, + ) + + class MockToolCallProvider(MockProvider): def __init__(self, tool_name: str, tool_args: dict[str, str] | None = None): super().__init__() @@ -326,6 +419,7 @@ def __init__(self): self.agent_done_called = False self.tool_start_called = False self.tool_end_called = False + self.agent_done_response = None async def on_agent_begin(self, run_context): self.agent_begin_called = True @@ -338,16 +432,120 @@ async def on_tool_end(self, run_context, tool, tool_args, tool_result): async def on_agent_done(self, run_context, llm_response): self.agent_done_called = True + self.agent_done_response = llm_response + + +class MockLateStopOnDoneHooks(MockHooks): + def __init__(self, event): + super().__init__() + self.event = event + + async def on_agent_done(self, run_context, llm_response): + await super().on_agent_done(run_context, llm_response) + self.event.set_extra("agent_stop_requested", True) + + +class HookPreStopRunner(ToolLoopAgentRunner): + def __init__(self, event): + super().__init__() + self.event = event + + def _is_stop_requested(self) -> bool: + if self.final_llm_resp is not None: + self.event.set_extra("agent_stop_requested", True) + return super()._is_stop_requested() + + +class LateStopAfterYieldRunner: + def __init__( + self, + event, + response_type: str, + text: str, + streaming: bool, + stop_after_yield: bool = True, + ): + self.run_context = ContextWrapper(context=MockAgentContext(event)) + self.response_type = response_type + self.text = text + self.streaming = streaming + self.stop_after_yield = stop_after_yield + self.stats = SimpleNamespace(to_dict=lambda: {}) + self._done = False + self._aborted = False + self.stop_requested = False + self.discard_late_result_called = False + + async def step(self): + yield SimpleNamespace( + type=self.response_type, + data={"chain": MessageChain().message(self.text)}, + ) + if self.stop_after_yield: + self.run_context.context.event.set_extra("agent_stop_requested", True) + self._done = True + + def done(self): + return self._done + + def request_stop(self): + self.stop_requested = True + + def was_aborted(self): + return self._aborted + + def discard_late_aborted_result(self): + self.discard_late_result_called = True + self._aborted = True + event = self.run_context.context.event + event.set_extra("agent_user_aborted", True) + event.set_extra("agent_stop_requested", False) class MockEvent: def __init__(self, umo: str, sender_id: str): self.unified_msg_origin = umo self._sender_id = sender_id + self._extras = {} + self.result = None + self.trace = SimpleNamespace(record=lambda *args, **kwargs: None) + self._stopped = False def get_sender_id(self): return self._sender_id + def set_extra(self, key, value): + self._extras[key] = value + + def get_extra(self, key=None, default=None): + if key is None: + return self._extras + return self._extras.get(key, default) + + def set_result(self, result): + self.result = result + + def clear_result(self): + self.result = None + + def is_stopped(self): + return self._stopped + + def stop(self): + self._stopped = True + + def track_temporary_local_file(self, _path): + return None + + def get_platform_name(self): + return "test" + + def get_platform_id(self): + return "test" + + async def send(self, *args, **kwargs): + return None + class MockAgentContext: def __init__(self, event): @@ -428,10 +626,43 @@ def runner(): return ToolLoopAgentRunner() +def test_record_llm_usage_keeps_usage_reference_to_prevent_id_reuse(runner): + usage = TokenUsage(input_other=10, output=5) + runner.req = SimpleNamespace(conversation=SimpleNamespace(token_usage=0)) + runner.stats = SimpleNamespace(token_usage=TokenUsage()) + runner._recorded_usages = [] + + runner._record_llm_usage(LLMResponse(role="assistant", usage=usage)) + runner._record_llm_usage(LLMResponse(role="assistant", usage=usage)) + + assert runner.stats.token_usage.total == usage.total + assert runner.req.conversation.token_usage == usage.total + assert runner._recorded_usages == [usage] + + +def test_is_message_from_llm_response_rejects_extra_content_parts(runner): + llm_resp = LLMResponse(role="assistant", completion_text="late text") + message = Message( + role="assistant", + content=[ + TextPart(text="late text"), + ImageURLPart( + image_url=ImageURLPart.ImageURL(url="https://example.com/a.png") + ), + ], + ) + + assert runner._is_message_from_llm_response(message, llm_resp) is False + + def _make_large_tool_result_text() -> str: return "x" * 100000 +async def _collect_async_iter(async_iter): + return [item async for item in async_iter] + + @pytest.mark.asyncio async def test_max_step_limit_functionality( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks @@ -1080,7 +1311,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_delayed_response( runner, provider_request, mock_tool_executor, mock_hooks ): provider = MockAbortableStreamProvider() @@ -1110,9 +1341,451 @@ 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 == "" + assert final_resp.result_chain is None + assert final_resp.reasoning_content is None + assert final_resp.reasoning_signature is None + assert final_resp.tools_call_name == [] + assert final_resp.tools_call_args == [] + assert final_resp.tools_call_ids == [] + assert final_resp.tools_call_extra_content == {} + assert ( + not runner.run_context.messages + or runner.run_context.messages[-1].role != "assistant" + ) + assert mock_hooks.agent_done_response is final_resp + + +@pytest.mark.asyncio +async def test_aborted_final_response_sanitizes_all_model_output_fields( + runner, provider_request, mock_tool_executor, mock_hooks +): + event = MockEvent("test:FriendMessage:leaky_abort", "u1") + provider = MockLeakyAbortProvider(event) + + await runner.reset( + provider=provider, + request=provider_request, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=mock_tool_executor, + agent_hooks=mock_hooks, + streaming=False, + ) + + responses = [] + async for response in runner.step(): + responses.append(response) + + assert [response.type for response in responses] == ["aborted"] + assert runner.was_aborted() is True + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.role == "assistant" + assert final_resp.completion_text == "" + assert final_resp.result_chain is None + assert final_resp.reasoning_content is None + assert final_resp.reasoning_signature is None + assert final_resp.raw_completion is None + assert final_resp.tools_call_args == [] + assert final_resp.tools_call_name == [] + assert final_resp.tools_call_ids == [] + assert final_resp.tools_call_extra_content == {} + assert final_resp.id == "late-id" + assert final_resp.usage is not None + assert final_resp.usage.total == 15 + assert runner.stats.token_usage.total == 15 + assert mock_hooks.agent_done_called is True + assert mock_hooks.agent_done_response is final_resp + + serialized_messages = repr(runner.run_context.messages) + assert "late completion text" not in serialized_messages + assert "late result chain" not in serialized_messages + assert "late reasoning" not in serialized_messages + assert "late_tool" not in serialized_messages + + +@pytest.mark.asyncio +async def test_runner_step_treats_agent_user_aborted_as_stop( + runner, provider_request, mock_tool_executor, mock_hooks +): + event = MockEvent("test:FriendMessage:runner_user_aborted", "u1") + event.set_extra("agent_user_aborted", True) + provider = MockLeakyAbortProvider() + + await runner.reset( + provider=provider, + request=provider_request, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=mock_tool_executor, + agent_hooks=mock_hooks, + streaming=False, + ) + + responses = [response async for response in runner.step()] + + assert [response.type for response in responses] == ["aborted"] + assert provider.call_count == 0 + assert runner.was_aborted() is True + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.completion_text == "" + + +@pytest.mark.asyncio +async def test_run_agent_discards_buffered_llm_result_after_abort( + runner, provider_request, mock_tool_executor, mock_hooks +): + from astrbot.core.astr_agent_run_util import run_agent + + provider = MockDelayedTextProvider() + event = MockEvent("test:FriendMessage:buffer_abort", "u1") + + await runner.reset( + provider=provider, + request=provider_request, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=mock_tool_executor, + agent_hooks=mock_hooks, + streaming=False, + ) + + agent_task = asyncio.create_task( + _collect_async_iter( + run_agent( + runner, + buffer_intermediate_messages=True, + ) + ) + ) + await asyncio.wait_for(provider.started.wait(), timeout=5) + event.set_extra("agent_stop_requested", True) + provider.release.set() + + chains = await asyncio.wait_for(agent_task, timeout=5) + + assert chains == [] + assert runner.was_aborted() is True + assert event.result is None + + +@pytest.mark.asyncio +async def test_run_agent_discards_buffered_result_when_stop_arrives_before_flush(): + from astrbot.core.astr_agent_run_util import run_agent + + event = MockEvent("test:FriendMessage:late_buffer_abort", "u1") + runner = LateStopAfterYieldRunner( + event, + response_type="llm_result", + text="buffered late text", + streaming=False, + ) + + chains = await _collect_async_iter( + run_agent( + cast(Any, runner), + buffer_intermediate_messages=True, + ) + ) + + assert chains == [] + assert event.result is None + assert runner.discard_late_result_called is True + assert runner.was_aborted() is True + + +@pytest.mark.asyncio +async def test_run_agent_treats_agent_user_aborted_as_stop(): + from astrbot.core.astr_agent_run_util import run_agent + + event = MockEvent("test:FriendMessage:user_aborted_abort", "u1") + event.set_extra("agent_user_aborted", True) + runner = LateStopAfterYieldRunner( + event, + response_type="llm_result", + text="aborted residual text", + streaming=False, + ) + + chains = await _collect_async_iter(run_agent(cast(Any, runner))) + + assert chains == [] + assert event.result is None + + +@pytest.mark.asyncio +async def test_stop_before_agent_done_hook_skips_llm_response_event( + provider_request, + mock_tool_executor, +): + from astrbot.core import astr_agent_hooks + from astrbot.core.astr_agent_hooks import MainAgentHooks + from astrbot.core.star.star_handler import EventType + + calls = [] + + async def fake_call_event_hook(event, hook_type, *args, **kwargs): + calls.append((hook_type, args)) + return False + + event = MockEvent("test:FriendMessage:hook_race_abort", "u1") + provider = MockProvider() + provider.should_call_tools = False + runner = HookPreStopRunner(event) + monkeypatch_context = pytest.MonkeyPatch() + monkeypatch_context.setattr( + astr_agent_hooks, + "call_event_hook", + fake_call_event_hook, + ) + + try: + await runner.reset( + provider=provider, + request=provider_request, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=mock_tool_executor, + agent_hooks=MainAgentHooks(), + streaming=False, + ) + + responses = [response async for response in runner.step()] + finally: + monkeypatch_context.undo() + + assert [response.type for response in responses] == ["aborted"] + called_types = [hook_type for hook_type, _ in calls] + assert EventType.OnLLMResponseEvent not in called_types + assert EventType.OnAgentDoneEvent in called_types + assert event.get_extra("_llm_reasoning_content") is None + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.completion_text == "" + assert "这是我的最终回答" not in repr(runner.run_context.messages) + + +@pytest.mark.asyncio +async def test_stop_after_agent_done_hook_discards_final_assistant_message( + provider_request, + mock_tool_executor, +): + event = MockEvent("test:FriendMessage:post_hook_abort", "u1") + provider = MockProvider() + provider.should_call_tools = False + hooks = MockLateStopOnDoneHooks(event) + runner = ToolLoopAgentRunner() + + await runner.reset( + provider=provider, + request=provider_request, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=mock_tool_executor, + agent_hooks=hooks, + streaming=False, + ) + + responses = [response async for response in runner.step()] + + assert [response.type for response in responses] == ["aborted"] + assert runner.was_aborted() is True + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.completion_text == "" + assert "这是我的最终回答" not in repr(runner.run_context.messages) + + +@pytest.mark.asyncio +async def test_live_agent_feeder_discards_residual_buffer_after_stop(): + from astrbot.core.astr_agent_run_util import _run_agent_feeder + + event = MockEvent("test:FriendMessage:live_late_buffer_abort", "u1") + runner = LateStopAfterYieldRunner( + event, + response_type="streaming_delta", + text="partial without punctuation", + streaming=True, + ) + text_queue = asyncio.Queue() + + await _run_agent_feeder( + cast(Any, runner), + text_queue, + max_step=30, + show_tool_use=True, + show_tool_call_result=False, + show_reasoning=False, + buffer_intermediate_messages=False, + ) + + assert await text_queue.get() is None + assert text_queue.empty() + + +@pytest.mark.asyncio +async def test_live_agent_discards_queued_tts_audio_after_stop(tmp_path): + from astrbot.core.astr_agent_run_util import run_live_agent + + audio_path = tmp_path / "late.wav" + audio_path.write_bytes(b"late-audio") + event = MockEvent("test:FriendMessage:live_tts_queued_abort", "u1") + runner = LateStopAfterYieldRunner( + event, + response_type="streaming_delta", + text="complete sentence!", + streaming=True, + stop_after_yield=False, + ) + tts_provider = MockDelayedTTSProvider(audio_path) + + live_task = asyncio.create_task( + _collect_async_iter(run_live_agent(cast(Any, runner), tts_provider)) + ) + await asyncio.wait_for(tts_provider.started.wait(), timeout=5) + event.set_extra("agent_stop_requested", True) + tts_provider.release.set() + + chains = await asyncio.wait_for(live_task, timeout=5) + + assert chains == [] + + +@pytest.mark.asyncio +async def test_stop_requested_before_stream_chunk_discards_chunk( + runner, provider_request, mock_tool_executor, mock_hooks +): + provider = MockStopAwareStreamProvider() + + await runner.reset( + provider=provider, + request=provider_request, + run_context=ContextWrapper(context=None), + tool_executor=mock_tool_executor, + agent_hooks=mock_hooks, + streaming=True, + ) + + runner.request_stop() + responses = [] + async for response in runner.step(): + responses.append(response) + + assert [response.type for response in responses] == ["aborted"] + assert runner.was_aborted() is True + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.completion_text == "" + assert final_resp.reasoning_content is None + + +@pytest.mark.asyncio +async def test_main_agent_hooks_skip_llm_response_hook_for_aborted_event(monkeypatch): + from astrbot.core import astr_agent_hooks + from astrbot.core.astr_agent_hooks import MainAgentHooks + from astrbot.core.star.star_handler import EventType + + calls = [] + + async def fake_call_event_hook(event, hook_type, *args, **kwargs): + calls.append((hook_type, args)) + return False + + monkeypatch.setattr(astr_agent_hooks, "call_event_hook", fake_call_event_hook) + + event = MockEvent("test:FriendMessage:hook_abort", "u1") + event.set_extra("agent_user_aborted", True) + response = LLMResponse( + role="assistant", + completion_text="", + reasoning_content="should not be stored", + ) + hooks = MainAgentHooks() + + await hooks.on_agent_done(ContextWrapper(context=MockAgentContext(event)), response) + + called_types = [hook_type for hook_type, _ in calls] + assert EventType.OnLLMResponseEvent not in called_types + assert EventType.OnAgentDoneEvent in called_types + assert event.get_extra("_llm_reasoning_content") is None + agent_done_call = next( + args for hook_type, args in calls if hook_type == EventType.OnAgentDoneEvent + ) + assert agent_done_call[1].completion_text == "" + assert agent_done_call[1].reasoning_content is None + + +@pytest.mark.asyncio +async def test_main_agent_hooks_skip_llm_response_hook_for_pending_stop(monkeypatch): + from astrbot.core import astr_agent_hooks + from astrbot.core.astr_agent_hooks import MainAgentHooks + from astrbot.core.star.star_handler import EventType + + calls = [] + + async def fake_call_event_hook(event, hook_type, *args, **kwargs): + calls.append((hook_type, args)) + return False + + monkeypatch.setattr(astr_agent_hooks, "call_event_hook", fake_call_event_hook) + + event = MockEvent("test:FriendMessage:hook_pending_stop", "u1") + event.set_extra("agent_stop_requested", True) + response = LLMResponse( + role="assistant", + completion_text="late hook text", + reasoning_content="late hook reasoning", + ) + hooks = MainAgentHooks() + + await hooks.on_agent_done(ContextWrapper(context=MockAgentContext(event)), response) + + called_types = [hook_type for hook_type, _ in calls] + assert EventType.OnLLMResponseEvent not in called_types + assert EventType.OnAgentDoneEvent in called_types + assert event.get_extra("_llm_reasoning_content") is None + agent_done_call = next( + args for hook_type, args in calls if hook_type == EventType.OnAgentDoneEvent + ) + assert agent_done_call[1].completion_text == "" + + +def test_llm_response_hook_propagation_stops_on_pending_agent_stop(): + from astrbot.core.pipeline.context_utils import _should_stop_hook_propagation + from astrbot.core.star.star_handler import EventType + + event = MockEvent("test:FriendMessage:hook_propagation_stop", "u1") + event.set_extra("agent_stop_requested", True) + + assert _should_stop_hook_propagation(event, EventType.OnLLMResponseEvent) is True + assert _should_stop_hook_propagation(event, EventType.OnAgentDoneEvent) is False + + +@pytest.mark.asyncio +async def test_main_agent_hooks_keep_normal_llm_response_hook(monkeypatch): + from astrbot.core import astr_agent_hooks + from astrbot.core.astr_agent_hooks import MainAgentHooks + from astrbot.core.star.star_handler import EventType + + calls = [] + + async def fake_call_event_hook(event, hook_type, *args, **kwargs): + calls.append((hook_type, args)) + return False + + monkeypatch.setattr(astr_agent_hooks, "call_event_hook", fake_call_event_hook) + + event = MockEvent("test:FriendMessage:hook_normal", "u1") + response = LLMResponse( + role="assistant", + completion_text="normal text", + reasoning_content="normal reasoning", + ) + hooks = MainAgentHooks() + + await hooks.on_agent_done(ContextWrapper(context=MockAgentContext(event)), response) + + called_types = [hook_type for hook_type, _ in calls] + assert EventType.OnLLMResponseEvent in called_types + assert EventType.OnAgentDoneEvent in called_types + assert event.get_extra("_llm_reasoning_content") == "normal reasoning" @pytest.mark.asyncio @@ -1364,6 +2037,234 @@ async def text_chat(self, **kwargs) -> LLMResponse: assert parts[0].text == "一张猫的照片" +@pytest.mark.asyncio +async def test_skills_like_requery_stop_discards_late_assistant_text(): + class SkillsLikeLateStopProvider(MockProvider): + def __init__(self, event): + super().__init__() + self.event = event + + async def text_chat(self, **kwargs) -> LLMResponse: + self.call_count += 1 + if self.call_count == 1: + return LLMResponse( + role="assistant", + completion_text="选择工具", + tools_call_name=["test_tool"], + tools_call_args=[{"query": "test"}], + tools_call_ids=["call_1"], + usage=TokenUsage(input_other=10, output=5), + ) + + self.event.set_extra("agent_stop_requested", True) + return LLMResponse( + role="assistant", + completion_text="skills_like late text", + usage=TokenUsage(input_other=10, output=5), + ) + + event = MockEvent(umo="test_umo", sender_id="test_sender") + provider = SkillsLikeLateStopProvider(event) + tool = FunctionTool( + name="test_tool", + description="测试", + parameters={"type": "object", "properties": {"query": {"type": "string"}}}, + handler=AsyncMock(), + ) + req = ProviderRequest( + prompt="调用工具", + func_tool=ToolSet(tools=[tool]), + contexts=[], + ) + runner = ToolLoopAgentRunner() + hooks = MockHooks() + + await runner.reset( + provider=provider, + request=req, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=cast(Any, MockToolExecutor()), + agent_hooks=hooks, + tool_schema_mode="skills_like", + ) + + responses = [response async for response in runner.step()] + + assert [response.type for response in responses] == ["llm_result", "aborted"] + visible_text = "".join( + response.data["chain"].get_plain_text() + for response in responses + if response.type != "aborted" + ) + assert "skills_like late text" not in visible_text + assert runner.was_aborted() is True + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.completion_text == "" + assert hooks.agent_done_response is final_resp + assert "skills_like late text" not in repr(runner.run_context.messages) + + +@pytest.mark.asyncio +async def test_skills_like_requery_stop_skips_repair_request(): + class SkillsLikeRepairStopProvider(MockProvider): + def __init__(self, event): + super().__init__() + self.event = event + + async def text_chat(self, **kwargs) -> LLMResponse: + self.call_count += 1 + if self.call_count == 1: + return LLMResponse( + role="assistant", + completion_text="选择工具", + tools_call_name=["test_tool"], + tools_call_args=[{"query": "test"}], + tools_call_ids=["call_1"], + ) + + self.event.set_extra("agent_stop_requested", True) + return LLMResponse(role="assistant", completion_text="") + + event = MockEvent(umo="test_umo", sender_id="test_sender") + provider = SkillsLikeRepairStopProvider(event) + tool = FunctionTool( + name="test_tool", + description="测试", + parameters={"type": "object", "properties": {"query": {"type": "string"}}}, + handler=AsyncMock(), + ) + req = ProviderRequest( + prompt="调用工具", + func_tool=ToolSet(tools=[tool]), + contexts=[], + ) + runner = ToolLoopAgentRunner() + + await runner.reset( + provider=provider, + request=req, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=cast(Any, MockToolExecutor()), + agent_hooks=MockHooks(), + tool_schema_mode="skills_like", + ) + + responses = [response async for response in runner.step()] + + assert [response.type for response in responses] == ["llm_result", "aborted"] + assert provider.call_count == 2 + + +@pytest.mark.asyncio +async def test_skills_like_requery_fallback_checks_stop_before_yielding_text(): + class StopAfterFallbackDoneHooks(MockHooks): + async def on_agent_done(self, run_context, llm_response): + await super().on_agent_done(run_context, llm_response) + run_context.context.event.set_extra("agent_stop_requested", True) + + class SkillsLikeFallbackProvider(MockProvider): + async def text_chat(self, **kwargs) -> LLMResponse: + self.call_count += 1 + if self.call_count == 1: + return LLMResponse( + role="assistant", + completion_text="选择工具", + tools_call_name=["test_tool"], + tools_call_args=[{"query": "test"}], + tools_call_ids=["call_1"], + usage=TokenUsage(input_other=10, output=5), + ) + + return LLMResponse( + role="assistant", + completion_text="skills_like fallback text", + reasoning_content="skills_like fallback reasoning", + usage=TokenUsage(input_other=10, output=5), + ) + + event = MockEvent(umo="test_umo", sender_id="test_sender") + provider = SkillsLikeFallbackProvider() + tool = FunctionTool( + name="test_tool", + description="测试", + parameters={"type": "object", "properties": {"query": {"type": "string"}}}, + handler=AsyncMock(), + ) + req = ProviderRequest( + prompt="调用工具", + func_tool=ToolSet(tools=[tool]), + contexts=[], + ) + runner = ToolLoopAgentRunner() + hooks = StopAfterFallbackDoneHooks() + + await runner.reset( + provider=provider, + request=req, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=cast(Any, MockToolExecutor()), + agent_hooks=hooks, + tool_schema_mode="skills_like", + ) + + responses = [response async for response in runner.step()] + + assert [response.type for response in responses] == ["llm_result", "aborted"] + visible_text = "".join( + response.data["chain"].get_plain_text() + for response in responses + if response.type != "aborted" + ) + assert "skills_like fallback text" not in visible_text + assert "skills_like fallback reasoning" not in visible_text + assert runner.was_aborted() is True + final_resp = runner.get_final_llm_resp() + assert final_resp is not None + assert final_resp.completion_text == "" + assert "skills_like fallback text" not in repr(runner.run_context.messages) + assert "skills_like fallback reasoning" not in repr(runner.run_context.messages) + + +@pytest.mark.asyncio +async def test_fallback_provider_not_called_after_stop_on_primary_error_response( + provider_request, + mock_tool_executor, +): + class StopErrProvider(MockErrProvider): + def __init__(self, event): + super().__init__() + self.provider_config["id"] = "primary" + self.event = event + + async def text_chat(self, **kwargs) -> LLMResponse: + response = await super().text_chat(**kwargs) + self.event.set_extra("agent_stop_requested", True) + return response + + event = MockEvent("test:FriendMessage:fallback_stop", "u1") + primary = StopErrProvider(event) + fallback = MockProvider() + fallback.provider_config["id"] = "fallback" + fallback.should_call_tools = False + runner = ToolLoopAgentRunner() + + await runner.reset( + provider=primary, + request=provider_request, + run_context=ContextWrapper(context=MockAgentContext(event)), + tool_executor=mock_tool_executor, + agent_hooks=MockHooks(), + fallback_providers=[fallback], + ) + + responses = [response async for response in runner.step()] + + assert [response.type for response in responses] == ["aborted"] + assert primary.call_count == 1 + assert fallback.call_count == 0 + + @pytest.mark.asyncio async def test_follow_up_accepted_when_active_and_not_stopping( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks @@ -1378,6 +2279,13 @@ async def test_follow_up_accepted_when_active_and_not_stopping( agent_hooks=mock_hooks, streaming=False, ) + runner._transition_state(AgentState.RUNNING) + + ticket = runner.follow_up(message_text="follow up while active") + + assert ticket is not None + assert ticket in runner._pending_follow_ups + assert len(runner._pending_follow_ups) == 1 @pytest.mark.asyncio