import asyncio import os import sys from pathlib import Path from types import SimpleNamespace from typing import Any, cast from unittest.mock import AsyncMock import pytest # 将项目根目录添加到 sys.path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from astrbot.core.agent.agent import Agent from astrbot.core.agent.handoff import HandoffTool 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.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 class MockProvider(Provider): """模拟Provider用于测试""" def __init__(self): super().__init__({}, {}) self.call_count = 0 self.should_call_tools = True self.max_calls_before_normal_response = 10 def get_current_key(self) -> str: return "test_key" def set_key(self, key: str): pass async def get_models(self) -> list[str]: return ["test_model"] async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 # 检查工具是否被禁用 func_tool = kwargs.get("func_tool") # 如果工具被禁用或超过最大调用次数，返回正常响应 if func_tool is None or self.call_count > self.max_calls_before_normal_response: return LLMResponse( role="assistant", completion_text="这是我的最终回答", usage=TokenUsage(input_other=10, output=5), ) # 模拟工具调用响应 if self.should_call_tools: return LLMResponse( role="assistant", completion_text="我需要使用工具来帮助您", tools_call_name=["test_tool"], tools_call_args=[{"query": "test"}], tools_call_ids=["call_123"], usage=TokenUsage(input_other=10, output=5), ) # 默认返回正常响应 return LLMResponse( role="assistant", completion_text="这是我的最终回答", usage=TokenUsage(input_other=10, output=5), ) async def text_chat_stream(self, **kwargs): response = await self.text_chat(**kwargs) response.is_chunk = True yield response response.is_chunk = False yield response class MockToolExecutor: """模拟工具执行器""" @classmethod def execute(cls, tool, run_context, **tool_args): async def generator(): # 模拟工具返回结果，使用正确的类型 from mcp.types import CallToolResult, TextContent result = CallToolResult( content=[TextContent(type="text", text="工具执行结果")] ) yield result return generator() class LargeTextToolExecutor: """模拟返回超长文本的工具执行器""" def __init__(self, text: str): self.text = text @classmethod def from_text(cls, text: str) -> "LargeTextToolExecutor": return cls(text) def execute(self, tool, run_context, **tool_args): async def generator(): from mcp.types import CallToolResult, TextContent result = CallToolResult(content=[TextContent(type="text", text=self.text)]) yield result return generator() class MockMixedContentToolExecutor: """模拟返回图片 + 文本的工具执行器""" @classmethod def execute(cls, tool, run_context, **tool_args): async def generator(): from mcp.types import CallToolResult, ImageContent, TextContent result = CallToolResult( content=[ ImageContent( type="image", data="dGVzdA==", mimeType="image/png", ), TextContent(type="text", text="直播间标题：新游首发：零~红蝶~"), ] ) yield result return generator() class VaryingUsageProvider(MockProvider): """Return distinct token usage values for each tool-loop request.""" async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 usage = TokenUsage( input_other=self.call_count * 100, input_cached=self.call_count * 10, output=self.call_count, ) if self.call_count == 1: return LLMResponse( role="assistant", tools_call_name=["test_tool"], tools_call_args=[{"query": "test"}], tools_call_ids=["call_varying_usage"], usage=usage, ) return LLMResponse( role="assistant", completion_text="final", usage=usage, ) class MissingFinalUsageProvider(VaryingUsageProvider): """Omit usage from the final response after reporting an earlier request.""" async def text_chat(self, **kwargs) -> LLMResponse: response = await super().text_chat(**kwargs) if self.call_count == 2: response.usage = None return response class MockFailingProvider(MockProvider): async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 raise RuntimeError("primary provider failed") class MockErrProvider(MockProvider): async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 return LLMResponse( role="err", completion_text="primary provider returned error", ) class CapturingProvider(MockProvider): def __init__(self, modalities: list[str]): super().__init__() self.provider_config["modalities"] = modalities self.received_contexts = [] self.received_func_tools = [] self.should_call_tools = False async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 self.received_contexts.append(kwargs.get("contexts")) self.received_func_tools.append(kwargs.get("func_tool")) return LLMResponse( role="assistant", completion_text="final", usage=TokenUsage(input_other=10, output=5), ) class MockEmptyOutputThenSuccessProvider(MockProvider): def __init__(self, failures_before_success: int = 1): super().__init__() self.failures_before_success = failures_before_success async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 if self.call_count <= self.failures_before_success: raise EmptyModelOutputError("model returned no usable output") return LLMResponse( role="assistant", completion_text="这是重试后的最终回答", usage=TokenUsage(input_other=10, output=5), ) class MockAbortableStreamProvider(MockProvider): async def text_chat_stream(self, **kwargs): abort_signal = kwargs.get("abort_signal") yield LLMResponse( role="assistant", completion_text="partial ", is_chunk=True, ) if abort_signal and abort_signal.is_set(): yield LLMResponse( role="assistant", completion_text="partial ", is_chunk=False, ) return yield LLMResponse( role="assistant", completion_text="partial final", is_chunk=False, ) class MockToolCallProvider(MockProvider): def __init__(self, tool_name: str, tool_args: dict[str, str] | None = None): super().__init__() self.tool_name = tool_name self.tool_args = tool_args or {} self.abort_signal = None async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 self.abort_signal = kwargs.get("abort_signal") return LLMResponse( role="assistant", completion_text="", tools_call_name=[self.tool_name], tools_call_args=[self.tool_args], tools_call_ids=[f"call_{self.tool_name}"], usage=TokenUsage(input_other=10, output=5), ) class SingleToolThenFinalProvider(MockProvider): def __init__(self, tool_name: str, tool_args: dict[str, str] | None = None): super().__init__() self.tool_name = tool_name self.tool_args = tool_args or {} async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 func_tool = kwargs.get("func_tool") if func_tool is None or self.call_count > 1: return LLMResponse( role="assistant", completion_text="最终回复", usage=TokenUsage(input_other=10, output=5), ) return LLMResponse( role="assistant", completion_text="", tools_call_name=[self.tool_name], tools_call_args=[self.tool_args], tools_call_ids=["call_large_result"], usage=TokenUsage(input_other=10, output=5), ) class CapturingToolLoopProvider(MockProvider): def __init__(self, tool_name: str): super().__init__() self.tool_name = tool_name self.received_contexts = [] async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 self.received_contexts.append(list(kwargs.get("contexts") or [])) func_tool = kwargs.get("func_tool") if func_tool is None or self.call_count > 1: return LLMResponse( role="assistant", completion_text="最终回复", usage=TokenUsage(input_other=10, output=5), ) return LLMResponse( role="assistant", completion_text="", tools_call_name=[self.tool_name], tools_call_args=[{"query": "test"}], tools_call_ids=["call_context_refresh"], usage=TokenUsage(input_other=10, output=5), ) class SequentialToolProvider(MockProvider): def __init__(self, tool_sequence: list[str]): super().__init__() self.tool_sequence = tool_sequence async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 func_tool = kwargs.get("func_tool") if func_tool is None or self.call_count > len(self.tool_sequence): return LLMResponse( role="assistant", completion_text="这是我的最终回答", usage=TokenUsage(input_other=10, output=5), ) tool_name = self.tool_sequence[self.call_count - 1] return LLMResponse( role="assistant", completion_text="", tools_call_name=[tool_name], tools_call_args=[{"query": f"step-{self.call_count}"}], tools_call_ids=[f"call_{self.call_count}"], usage=TokenUsage(input_other=10, output=5), ) class MockHandoffProvider(MockToolCallProvider): def __init__(self, handoff_tool_name: str): super().__init__(handoff_tool_name, {"input": "delegate this task"}) class MockHooks(BaseAgentRunHooks): """模拟钩子函数""" def __init__(self): self.agent_begin_called = False self.agent_done_called = False self.tool_start_called = False self.tool_end_called = False async def on_agent_begin(self, run_context): self.agent_begin_called = True async def on_tool_start(self, run_context, tool, tool_args): self.tool_start_called = True async def on_tool_end(self, run_context, tool, tool_args, tool_result): self.tool_end_called = True async def on_agent_done(self, run_context, llm_response): self.agent_done_called = True class MockEvent: def __init__(self, umo: str, sender_id: str): self.unified_msg_origin = umo self._sender_id = sender_id def get_sender_id(self): return self._sender_id class MockAgentContext: def __init__(self, event): self.event = event class BlockingSubagentContext: def __init__(self): self.started = asyncio.Event() self.cancelled = False async def get_current_chat_provider_id(self, _umo: str) -> str: return "provider-id" def get_config(self, **_kwargs): return {"provider_settings": {}} async def tool_loop_agent(self, **_kwargs): self.started.set() try: await asyncio.Future() except asyncio.CancelledError: self.cancelled = True raise class BlockingToolState: def __init__(self): self.started = asyncio.Event() self.cancelled = False async def handler(self, event, query: str = ""): del event, query self.started.set() try: await asyncio.Future() except asyncio.CancelledError: self.cancelled = True raise @pytest.fixture def mock_provider(): return MockProvider() @pytest.fixture def mock_tool_executor(): return MockToolExecutor() @pytest.fixture def mock_hooks(): return MockHooks() @pytest.fixture def tool_set(): """创建测试用的工具集""" tool = FunctionTool( name="test_tool", description="测试工具", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) return ToolSet(tools=[tool]) @pytest.fixture def provider_request(tool_set): """创建测试用的ProviderRequest""" return ProviderRequest(prompt="请帮我查询信息", func_tool=tool_set, contexts=[]) @pytest.fixture def runner(): """创建ToolLoopAgentRunner实例""" return ToolLoopAgentRunner() def _make_large_tool_result_text() -> str: return "x" * 100000 @pytest.mark.asyncio async def test_max_step_limit_functionality( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """测试最大步数限制功能""" # 设置模拟provider，让它总是返回工具调用 mock_provider.should_call_tools = True mock_provider.max_calls_before_normal_response = ( 100 # 设置一个很大的值，确保不会自然结束 ) # 初始化runner await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # 设置较小的最大步数来测试限制功能 max_steps = 3 # 收集所有响应 responses = [] async for response in runner.step_until_done(max_steps): responses.append(response) # 验证结果 assert runner.done(), "代理应该在达到最大步数后完成" # 验证工具被禁用（这是最重要的验证点） assert runner.req.func_tool is None, "达到最大步数后工具应该被禁用" # 验证有最终响应 final_responses = [r for r in responses if r.type == "llm_result"] assert len(final_responses) > 0, "应该有最终的LLM响应" # 验证最后一条消息是assistant的最终回答 last_message = runner.run_context.messages[-1] assert last_message.role == "assistant", "最后一条消息应该是assistant的最终回答" @pytest.mark.asyncio async def test_max_step_final_request_includes_limit_prompt( runner, provider_request, mock_tool_executor, mock_hooks ): """The forced final step must use contexts recomputed after max-step prompt.""" provider = CapturingToolLoopProvider("test_tool") await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) async def snapshot_context_manager(messages, trusted_token_usage=0): return list(messages) runner.request_context_manager.process = snapshot_context_manager async for _ in runner.step_until_done(1): pass assert provider.call_count == 2 final_contexts = provider.received_contexts[-1] assert final_contexts[-1].role == "user" assert final_contexts[-1].content == runner.MAX_STEPS_REACHED_PROMPT @pytest.mark.asyncio async def test_tool_loop_next_request_includes_tool_result( runner, provider_request, mock_tool_executor, mock_hooks ): """Tool-loop provider contexts must be recomputed after tool results append.""" provider = CapturingToolLoopProvider("test_tool") await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) async def snapshot_context_manager(messages, trusted_token_usage=0): return list(messages) runner.request_context_manager.process = snapshot_context_manager async for _ in runner.step_until_done(3): pass assert provider.call_count == 2 second_contexts = provider.received_contexts[1] tool_messages = [msg for msg in second_contexts if msg.role == "tool"] assert len(tool_messages) == 1 assert tool_messages[0].tool_call_id == "call_context_refresh" assert "工具执行结果" in tool_messages[0].content @pytest.mark.asyncio async def test_normal_completion_without_max_step( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """测试正常完成（不触发最大步数限制）""" # 设置模拟provider，让它在第2次调用时返回正常响应 mock_provider.should_call_tools = True mock_provider.max_calls_before_normal_response = 2 # 初始化runner await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # 设置足够大的最大步数 max_steps = 10 # 收集所有响应 responses = [] async for response in runner.step_until_done(max_steps): responses.append(response) # 验证结果 assert runner.done(), "代理应该正常完成" # 验证没有触发最大步数限制 - 通过检查provider调用次数 # mock_provider在第2次调用后返回正常响应，所以不应该达到max_steps(10) assert mock_provider.call_count < max_steps, ( f"正常完成时调用次数({mock_provider.call_count})应该小于最大步数({max_steps})" ) # 验证没有最大步数警告消息（注意：实际注入的是user角色的消息） user_messages = [m for m in runner.run_context.messages if m.role == "user"] max_step_messages = [ m for m in user_messages if "工具调用次数已达到上限" in m.content ] assert len(max_step_messages) == 0, "正常完成时不应该有步数限制消息" # 验证工具仍然可用（没有被禁用） assert runner.req.func_tool is not None, "正常完成时工具不应该被禁用" @pytest.mark.asyncio @pytest.mark.parametrize("streaming", [False, True]) async def test_stats_separate_latest_context_from_cumulative_usage( runner, provider_request, mock_tool_executor, mock_hooks, streaming ): """Context occupancy uses the latest input while usage remains cumulative.""" provider = VaryingUsageProvider() await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=streaming, ) async for _ in runner.step_until_done(3): pass assert provider.call_count == 2 assert runner.stats.token_usage == TokenUsage( input_other=300, input_cached=30, output=3, ) assert runner.stats.current_context_tokens == 220 assert runner.stats.to_dict()["current_context_tokens"] == 220 @pytest.mark.asyncio @pytest.mark.parametrize("streaming", [False, True]) async def test_stats_emit_update_after_each_completed_llm_request( runner, provider_request, mock_tool_executor, mock_hooks, streaming ): """Emit one stats update for every completed LLM request.""" provider = VaryingUsageProvider() await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=streaming, ) responses = [response async for response in runner.step_until_done(3)] stats_responses = [ response for response in responses if response.type == "agent_stats" ] assert provider.call_count == 2 assert len(stats_responses) == 2 assert [response.data["chain"].type for response in stats_responses] == [ "agent_stats", "agent_stats", ] stats_snapshots = [ response.data["chain"].chain[0].data for response in stats_responses ] assert [snapshot["current_context_tokens"] for snapshot in stats_snapshots] == [ 110, 220, ] assert stats_snapshots[0]["token_usage"]["input_other"] == 100 assert stats_snapshots[0]["token_usage"]["input_cached"] == 10 assert stats_snapshots[1]["token_usage"]["input_other"] == 300 assert stats_snapshots[1]["token_usage"]["input_cached"] == 30 # Emitted events keep their own snapshots even if live stats mutate later. runner.stats.token_usage.input_other = 999 assert stats_snapshots[0]["token_usage"]["input_other"] == 100 assert stats_snapshots[1]["token_usage"]["input_other"] == 300 assert responses.index(stats_responses[0]) < next( index for index, response in enumerate(responses) if response.type == "tool_call" ) @pytest.mark.asyncio @pytest.mark.parametrize("streaming", [False, True]) async def test_stats_clear_current_context_when_latest_usage_is_missing( runner, provider_request, mock_tool_executor, mock_hooks, streaming ): """Do not expose stale context occupancy when the latest usage is unknown.""" provider = MissingFinalUsageProvider() await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=streaming, ) async for _ in runner.step_until_done(3): pass assert runner.stats.token_usage == TokenUsage( input_other=100, input_cached=10, output=1, ) assert runner.stats.current_context_tokens == 0 assert runner.stats.to_dict()["current_context_tokens"] == 0 @pytest.mark.asyncio async def test_max_step_with_streaming( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """测试流式响应下的最大步数限制""" # 设置模拟provider mock_provider.should_call_tools = True mock_provider.max_calls_before_normal_response = 100 # 初始化runner，启用流式响应 await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=True, ) # 设置较小的最大步数 max_steps = 2 # 收集所有响应 responses = [] async for response in runner.step_until_done(max_steps): responses.append(response) # 验证结果 assert runner.done(), "代理应该在达到最大步数后完成" # 验证有流式响应 streaming_responses = [r for r in responses if r.type == "streaming_delta"] assert len(streaming_responses) > 0, "应该有流式响应" # 验证工具被禁用 assert runner.req.func_tool is None, "达到最大步数后工具应该被禁用" # 验证最后一条消息是assistant的最终回答 last_message = runner.run_context.messages[-1] assert last_message.role == "assistant", "最后一条消息应该是assistant的最终回答" @pytest.mark.asyncio async def test_hooks_called_with_max_step( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """测试达到最大步数时钩子函数是否被正确调用""" # 设置模拟provider mock_provider.should_call_tools = True mock_provider.max_calls_before_normal_response = 100 # 初始化runner await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # 设置较小的最大步数 max_steps = 2 # 执行步骤 async for response in runner.step_until_done(max_steps): pass # 验证钩子函数被调用 assert mock_hooks.agent_begin_called, "on_agent_begin应该被调用" assert mock_hooks.agent_done_called, "on_agent_done应该被调用" assert mock_hooks.tool_start_called, "on_tool_start应该被调用" assert mock_hooks.tool_end_called, "on_tool_end应该被调用" @pytest.mark.asyncio async def test_tool_result_includes_all_calltoolresult_content( runner, mock_provider, provider_request, mock_hooks, monkeypatch ): """工具返回多个 content 项时，tool result 应包含全部内容。""" from astrbot.core.agent.tool_image_cache import tool_image_cache mock_provider.should_call_tools = True mock_provider.max_calls_before_normal_response = 1 saved_images = [] def fake_save_image( base64_data, tool_call_id, tool_name, index=0, mime_type="image/png" ): saved_images.append( { "base64_data": base64_data, "tool_call_id": tool_call_id, "tool_name": tool_name, "index": index, "mime_type": mime_type, } ) return SimpleNamespace( file_path=f"/tmp/{tool_call_id}_{index}.png", mime_type=mime_type ) monkeypatch.setattr(tool_image_cache, "save_image", fake_save_image) await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=MockMixedContentToolExecutor, agent_hooks=mock_hooks, streaming=False, ) async for _ in runner.step_until_done(3): pass tool_messages = [ m for m in runner.run_context.messages if getattr(m, "role", None) == "tool" ] assert len(tool_messages) == 1 content = str(tool_messages[0].content) assert "Image returned and cached at path='/tmp/call_123_0.png'." in content assert "直播间标题：新游首发：零~红蝶~" in content assert saved_images == [ { "base64_data": "dGVzdA==", "tool_call_id": "call_123", "tool_name": "test_tool", "index": 0, "mime_type": "image/png", } ] @pytest.mark.asyncio async def test_runner_replaces_runtime_image_context_before_provider_call( runner, provider_request, mock_hooks ): provider = CapturingProvider(modalities=["tool_use"]) await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=MockToolExecutor, agent_hooks=mock_hooks, streaming=False, ) runner.run_context.messages.append( Message( role="user", content=[ TextPart(text="Review this image"), ImageURLPart( image_url=ImageURLPart.ImageURL( url="data:image/png;base64,dGVzdA==" ) ), ], ) ) async for _ in runner.step_until_done(1): pass assert provider.received_contexts sent_context = provider.received_contexts[0] assert sent_context[-1]["content"] == [ {"type": "text", "text": "Review this image"}, {"type": "text", "text": "[Image]"}, ] assert len(runner.run_context.messages[-2].content) == 2 @pytest.mark.asyncio async def test_runner_builds_placeholder_for_unsupported_request_image( runner, mock_hooks, tool_set ): provider = CapturingProvider(modalities=["tool_use"]) request = ProviderRequest( prompt="Describe it", image_urls=["/path/that/should/not/be/read.jpg"], func_tool=tool_set, contexts=[], ) await runner.reset( provider=provider, request=request, run_context=ContextWrapper(context=None), tool_executor=MockToolExecutor, agent_hooks=mock_hooks, streaming=False, ) async for _ in runner.step_until_done(1): pass sent_context = provider.received_contexts[0] assert sent_context[-1]["content"] == [ {"type": "text", "text": "Describe it"}, {"type": "text", "text": "[Image]"}, ] @pytest.mark.asyncio async def test_runner_clears_tools_for_provider_without_tool_use( runner, provider_request, mock_hooks, mock_tool_executor ): provider = CapturingProvider(modalities=["text"]) await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) async for _ in runner.step_until_done(1): pass assert provider.received_func_tools == [None] @pytest.mark.asyncio async def test_same_tool_consecutive_results_include_escalating_guidance( runner, mock_tool_executor, mock_hooks ): runner_cls = type(runner) total_calls = runner_cls.REPEATED_TOOL_NOTICE_L3_THRESHOLD provider = SequentialToolProvider(["test_tool"] * total_calls) tool = FunctionTool( name="test_tool", description="测试工具", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) request = ProviderRequest( prompt="请连续执行工具", func_tool=ToolSet(tools=[tool]), contexts=[], ) await runner.reset( provider=provider, request=request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) async for _ in runner.step_until_done(total_calls + 1): pass tool_messages = [ m for m in runner.run_context.messages if getattr(m, "role", None) == "tool" ] assert len(tool_messages) == total_calls tool_contents = [str(message.content) for message in tool_messages] level_1_notice = runner_cls.REPEATED_TOOL_NOTICE_L1_TEMPLATE.format( tool_name="test_tool", streak=runner_cls.REPEATED_TOOL_NOTICE_L1_THRESHOLD, ) level_2_notice = runner_cls.REPEATED_TOOL_NOTICE_L2_TEMPLATE.format( tool_name="test_tool", streak=runner_cls.REPEATED_TOOL_NOTICE_L2_THRESHOLD, ) level_3_notice = runner_cls.REPEATED_TOOL_NOTICE_L3_TEMPLATE.format( tool_name="test_tool", streak=runner_cls.REPEATED_TOOL_NOTICE_L3_THRESHOLD, ) for streak, content in enumerate(tool_contents, start=1): if streak < runner_cls.REPEATED_TOOL_NOTICE_L1_THRESHOLD: assert level_1_notice not in content assert level_2_notice not in content assert level_3_notice not in content elif streak < runner_cls.REPEATED_TOOL_NOTICE_L2_THRESHOLD: assert level_1_notice in content assert level_2_notice not in content assert level_3_notice not in content elif streak < runner_cls.REPEATED_TOOL_NOTICE_L3_THRESHOLD: assert level_1_notice not in content assert level_2_notice in content assert level_3_notice not in content else: assert level_1_notice not in content assert level_2_notice not in content assert level_3_notice in content @pytest.mark.asyncio async def test_same_tool_streak_resets_after_switching_tools( runner, mock_tool_executor, mock_hooks ): runner_cls = type(runner) repeated_after_reset = runner_cls.REPEATED_TOOL_NOTICE_L1_THRESHOLD provider = SequentialToolProvider( ["test_tool", "other_tool", *(["test_tool"] * repeated_after_reset)] ) tool_a = FunctionTool( name="test_tool", description="测试工具 A", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) tool_b = FunctionTool( name="other_tool", description="测试工具 B", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) request = ProviderRequest( prompt="切换工具后再重复", func_tool=ToolSet(tools=[tool_a, tool_b]), contexts=[], ) await runner.reset( provider=provider, request=request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) async for _ in runner.step_until_done(repeated_after_reset + 3): pass tool_messages = [ m for m in runner.run_context.messages if getattr(m, "role", None) == "tool" ] assert len(tool_messages) == repeated_after_reset + 2 tool_contents = [str(message.content) for message in tool_messages] level_1_notice = runner_cls.REPEATED_TOOL_NOTICE_L1_TEMPLATE.format( tool_name="test_tool", streak=runner_cls.REPEATED_TOOL_NOTICE_L1_THRESHOLD, ) level_2_notice = runner_cls.REPEATED_TOOL_NOTICE_L2_TEMPLATE.format( tool_name="test_tool", streak=runner_cls.REPEATED_TOOL_NOTICE_L2_THRESHOLD, ) assert level_1_notice not in tool_contents[0] assert level_1_notice not in tool_contents[1] assert level_2_notice not in tool_contents[0] assert level_2_notice not in tool_contents[1] repeated_contents = tool_contents[2:] for streak_after_reset, content in enumerate(repeated_contents, start=1): if streak_after_reset < runner_cls.REPEATED_TOOL_NOTICE_L1_THRESHOLD: assert level_1_notice not in content assert level_2_notice not in content elif streak_after_reset < runner_cls.REPEATED_TOOL_NOTICE_L2_THRESHOLD: assert level_1_notice in content assert level_2_notice not in content else: assert level_1_notice not in content assert level_2_notice in content @pytest.mark.asyncio async def test_fallback_provider_used_when_primary_raises( runner, provider_request, mock_tool_executor, mock_hooks ): primary_provider = MockFailingProvider() fallback_provider = MockProvider() fallback_provider.should_call_tools = False await runner.reset( provider=primary_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, fallback_providers=[fallback_provider], ) async for _ in runner.step_until_done(5): pass final_resp = runner.get_final_llm_resp() assert final_resp is not None assert final_resp.role == "assistant" assert final_resp.completion_text == "这是我的最终回答" assert primary_provider.call_count == 1 assert fallback_provider.call_count == 1 @pytest.mark.asyncio async def test_fallback_provider_used_when_primary_returns_err( runner, provider_request, mock_tool_executor, mock_hooks ): primary_provider = MockErrProvider() fallback_provider = MockProvider() fallback_provider.should_call_tools = False await runner.reset( provider=primary_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, fallback_providers=[fallback_provider], ) async for _ in runner.step_until_done(5): pass final_resp = runner.get_final_llm_resp() assert final_resp is not None assert final_resp.role == "assistant" assert final_resp.completion_text == "这是我的最终回答" assert primary_provider.call_count == 1 assert fallback_provider.call_count == 1 @pytest.mark.asyncio async def test_empty_output_is_retried_before_succeeding( runner, provider_request, mock_tool_executor, mock_hooks, monkeypatch ): monkeypatch.setattr(runner, "EMPTY_OUTPUT_RETRY_WAIT_MIN_S", 0) monkeypatch.setattr(runner, "EMPTY_OUTPUT_RETRY_WAIT_MAX_S", 0) provider = MockEmptyOutputThenSuccessProvider(failures_before_success=1) await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) async for _ in runner.step_until_done(5): pass final_resp = runner.get_final_llm_resp() assert final_resp is not None assert final_resp.role == "assistant" assert final_resp.completion_text == "这是重试后的最终回答" assert provider.call_count == 2 @pytest.mark.asyncio async def test_empty_output_retries_exhausted_then_uses_fallback_provider( runner, provider_request, mock_tool_executor, mock_hooks, monkeypatch ): monkeypatch.setattr(runner, "EMPTY_OUTPUT_RETRY_WAIT_MIN_S", 0) monkeypatch.setattr(runner, "EMPTY_OUTPUT_RETRY_WAIT_MAX_S", 0) primary_provider = MockEmptyOutputThenSuccessProvider( failures_before_success=runner.EMPTY_OUTPUT_RETRY_ATTEMPTS ) fallback_provider = MockProvider() fallback_provider.should_call_tools = False await runner.reset( provider=primary_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, fallback_providers=[fallback_provider], ) async for _ in runner.step_until_done(5): pass final_resp = runner.get_final_llm_resp() assert final_resp is not None assert final_resp.role == "assistant" assert final_resp.completion_text == "这是我的最终回答" assert primary_provider.call_count == runner.EMPTY_OUTPUT_RETRY_ATTEMPTS assert fallback_provider.call_count == 1 @pytest.mark.asyncio async def test_stop_signal_returns_aborted_and_persists_partial_message( runner, provider_request, mock_tool_executor, mock_hooks ): provider = MockAbortableStreamProvider() await runner.reset( provider=provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=True, ) step_iter = runner.step() first_resp = await step_iter.__anext__() assert first_resp.type == "streaming_delta" runner.request_stop() rest_responses = [] async for response in step_iter: rest_responses.append(response) assert any(resp.type == "aborted" for resp in rest_responses) assert runner.was_aborted() is True 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" @pytest.mark.asyncio async def test_stop_interrupts_pending_subagent_handoff(mock_hooks): subagent_context = BlockingSubagentContext() event = MockEvent("webchat:FriendMessage:webchat!user!session", "user") handoff_tool = HandoffTool( Agent(name="subagent", instructions="subagent-instructions", tools=[]), tool_description="Delegate tasks to the subagent.", ) provider = MockHandoffProvider(handoff_tool.name) request = ProviderRequest( prompt="delegate", func_tool=ToolSet(tools=[handoff_tool]), contexts=[], ) runner = ToolLoopAgentRunner() await runner.reset( provider=provider, request=request, run_context=ContextWrapper( context=SimpleNamespace(event=event, context=subagent_context) ), tool_executor=FunctionToolExecutor(), agent_hooks=mock_hooks, streaming=False, ) step_iter = runner.step() first_resp = await step_iter.__anext__() if first_resp.type == "agent_stats": first_resp = await step_iter.__anext__() assert first_resp.type == "tool_call" assert provider.abort_signal is not None assert provider.abort_signal.is_set() is False pending_resp = asyncio.create_task(step_iter.__anext__()) await asyncio.wait_for(subagent_context.started.wait(), timeout=5) runner.request_stop() assert provider.abort_signal.is_set() is True aborted_resp = await asyncio.wait_for(pending_resp, timeout=1) assert aborted_resp.type == "aborted" assert runner.was_aborted() is True assert subagent_context.cancelled is True with pytest.raises(StopAsyncIteration): await step_iter.__anext__() @pytest.mark.asyncio async def test_stop_interrupts_pending_regular_tool(mock_hooks): tool_state = BlockingToolState() event = MockEvent("webchat:FriendMessage:webchat!user!session", "user") tool = FunctionTool( name="long_tool", description="A long-running test tool", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=tool_state.handler, ) provider = MockToolCallProvider(tool.name, {"query": "slow"}) request = ProviderRequest( prompt="run a slow tool", func_tool=ToolSet(tools=[tool]), contexts=[], ) runner = ToolLoopAgentRunner() await runner.reset( provider=provider, request=request, run_context=ContextWrapper( context=SimpleNamespace(event=event, context=SimpleNamespace()) ), tool_executor=FunctionToolExecutor(), agent_hooks=mock_hooks, streaming=False, ) step_iter = runner.step() first_resp = await step_iter.__anext__() if first_resp.type == "agent_stats": first_resp = await step_iter.__anext__() assert first_resp.type == "tool_call" assert provider.abort_signal is not None assert provider.abort_signal.is_set() is False pending_resp = asyncio.create_task(step_iter.__anext__()) await asyncio.wait_for(tool_state.started.wait(), timeout=5) runner.request_stop() assert provider.abort_signal.is_set() is True aborted_resp = await asyncio.wait_for(pending_resp, timeout=5) assert aborted_resp.type == "aborted" assert runner.was_aborted() is True assert tool_state.cancelled is True with pytest.raises(StopAsyncIteration): await step_iter.__anext__() @pytest.mark.asyncio async def test_tool_result_injects_follow_up_notice( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): mock_event = MockEvent("test:FriendMessage:follow_up", "u1") run_context = ContextWrapper(context=MockAgentContext(mock_event)) await runner.reset( provider=mock_provider, request=provider_request, run_context=run_context, tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) ticket1 = runner.follow_up( message_text="follow up 1", ) ticket2 = runner.follow_up( message_text="follow up 2", ) assert ticket1 is not None assert ticket2 is not None async for _ in runner.step(): pass assert provider_request.tool_calls_result is not None assert isinstance(provider_request.tool_calls_result, list) assert provider_request.tool_calls_result tool_result = str( provider_request.tool_calls_result[0].tool_calls_result[0].content ) assert "SYSTEM NOTICE" in tool_result assert "1. follow up 1" in tool_result assert "2. follow up 2" in tool_result assert ticket1.resolved.is_set() is True assert ticket2.resolved.is_set() is True assert ticket1.consumed is True assert ticket2.consumed is True @pytest.mark.asyncio async def test_follow_up_ticket_not_consumed_when_no_next_tool_call( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): mock_provider.should_call_tools = False mock_event = MockEvent("test:FriendMessage:follow_up_no_tool", "u1") run_context = ContextWrapper(context=MockAgentContext(mock_event)) await runner.reset( provider=mock_provider, request=provider_request, run_context=run_context, tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) ticket = runner.follow_up(message_text="follow up without tool") assert ticket is not None async for _ in runner.step(): pass assert ticket.resolved.is_set() is True assert ticket.consumed is False @pytest.mark.asyncio async def test_skills_like_requery_passes_extra_user_content_parts(): """skills-like 模式 re-query 时应传递 extra_user_content_parts（如 image_caption）""" from astrbot.core.agent.message import TextPart captured_kwargs = {} class SkillsLikeProvider(MockProvider): async def text_chat(self, **kwargs) -> LLMResponse: self.call_count += 1 if self.call_count == 1: # 第一次调用：返回工具选择（light schema） 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), ) if self.call_count == 2: # 第二次调用：re-query with param schema captured_kwargs.update(kwargs) return LLMResponse( role="assistant", completion_text="调用工具", tools_call_name=["test_tool"], tools_call_args=[{"query": "actual"}], tools_call_ids=["call_2"], usage=TokenUsage(input_other=10, output=5), ) # 后续调用：正常回复 return LLMResponse( role="assistant", completion_text="最终回复", usage=TokenUsage(input_other=10, output=5), ) provider = SkillsLikeProvider() tool = FunctionTool( name="test_tool", description="测试", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) tool_set = ToolSet(tools=[tool]) caption_part = TextPart(text="一张猫的照片") req = ProviderRequest( prompt="看看这张图", func_tool=tool_set, contexts=[], extra_user_content_parts=[caption_part], ) event = MockEvent(umo="test_umo", sender_id="test_sender") ctx = MockAgentContext(event) run_context = ContextWrapper(context=ctx) runner = ToolLoopAgentRunner() await runner.reset( provider=provider, request=req, run_context=run_context, tool_executor=cast(Any, MockToolExecutor()), agent_hooks=MockHooks(), tool_schema_mode="skills_like", ) async for _ in runner.step(): pass # 验证 re-query 调用包含了 extra_user_content_parts assert "extra_user_content_parts" in captured_kwargs, ( "re-query 应该传递 extra_user_content_parts" ) parts = captured_kwargs["extra_user_content_parts"] assert len(parts) == 1 assert parts[0].text == "一张猫的照片" @pytest.mark.asyncio async def test_follow_up_accepted_when_active_and_not_stopping( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Test that follow-up is accepted when runner is active and stop is not requested.""" await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) @pytest.mark.asyncio async def test_large_tool_result_is_spilled_to_file_and_replaced_with_read_notice( tmp_path, ): tool = FunctionTool( name="test_tool", description="测试工具", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) read_tool = FunctionTool( name="astrbot_file_read_tool", description="read file", parameters={"type": "object", "properties": {"path": {"type": "string"}}}, handler=AsyncMock(), ) tool_set = ToolSet(tools=[tool, read_tool]) provider = SingleToolThenFinalProvider(tool.name, {"query": "large"}) request = ProviderRequest(prompt="run tool", func_tool=tool_set, contexts=[]) runner = ToolLoopAgentRunner() await runner.reset( provider=provider, request=request, run_context=ContextWrapper(context=None), tool_executor=cast( Any, LargeTextToolExecutor.from_text(_make_large_tool_result_text()), ), agent_hooks=MockHooks(), streaming=False, tool_result_overflow_dir=str(tmp_path), read_tool=read_tool, ) responses = [] async for response in runner.step_until_done(3): responses.append(response) tool_messages = [m for m in runner.run_context.messages if m.role == "tool"] assert len(tool_messages) == 1 tool_message_content = str(tool_messages[0].content) assert "xxxxxxxxxx" in tool_message_content assert "Truncated tool output preview shown above." in tool_message_content assert "The tool output was too large to include directly" in tool_message_content assert "`astrbot_file_read_tool`" in tool_message_content assert "Use `astrbot_file_read_tool` to inspect it." in tool_message_content overflow_files = list(Path(tmp_path).glob("call_large_result_*.txt")) assert len(overflow_files) == 1 assert ( overflow_files[0].read_text(encoding="utf-8") == _make_large_tool_result_text() ) assert str(overflow_files[0]) in tool_message_content llm_results = [resp for resp in responses if resp.type == "llm_result"] assert llm_results @pytest.mark.asyncio async def test_large_tool_result_keeps_preview_when_spill_fails( tmp_path, monkeypatch: pytest.MonkeyPatch, ): tool = FunctionTool( name="test_tool", description="测试工具", parameters={"type": "object", "properties": {"query": {"type": "string"}}}, handler=AsyncMock(), ) read_tool = FunctionTool( name="astrbot_file_read_tool", description="read file", parameters={"type": "object", "properties": {"path": {"type": "string"}}}, handler=AsyncMock(), ) tool_set = ToolSet(tools=[tool, read_tool]) provider = SingleToolThenFinalProvider(tool.name, {"query": "large"}) request = ProviderRequest(prompt="run tool", func_tool=tool_set, contexts=[]) runner = ToolLoopAgentRunner() async def _raise_spill_error(*, tool_call_id: str, content: str) -> str: raise OSError("disk full") monkeypatch.setattr(runner, "_write_tool_result_overflow_file", _raise_spill_error) await runner.reset( provider=provider, request=request, run_context=ContextWrapper(context=None), tool_executor=cast( Any, LargeTextToolExecutor.from_text(_make_large_tool_result_text()), ), agent_hooks=MockHooks(), streaming=False, tool_result_overflow_dir=str(tmp_path), read_tool=read_tool, ) async for _ in runner.step_until_done(3): pass tool_messages = [m for m in runner.run_context.messages if m.role == "tool"] assert len(tool_messages) == 1 tool_message_content = str(tool_messages[0].content) assert "xxxxxxxxxx" in tool_message_content assert "Tool output exceeded the inline result limit" in tool_message_content assert "disk full" in tool_message_content @pytest.mark.asyncio async def test_follow_up_rejected_when_stop_requested( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Test that follow-up is rejected when stop has been requested.""" await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # Request stop runner.request_stop() assert runner._is_stop_requested() is True ticket = runner.follow_up(message_text="follow-up after stop") assert ticket is None, "Follow-up should be rejected after stop is requested" assert len(runner._pending_follow_ups) == 0 @pytest.mark.asyncio async def test_follow_up_rejected_when_runner_done( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Test that follow-up is rejected when runner is done.""" await runner.reset( provider=mock_provider, request=provider_request, run_context=ContextWrapper(context=None), tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # Run to completion async for _ in runner.step_until_done(10): pass # Runner should be done assert runner.done() ticket = runner.follow_up(message_text="follow-up after done") assert ticket is None, "Follow-up should be rejected when runner is done" @pytest.mark.asyncio async def test_follow_up_rejected_after_stop_before_tool_call( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Test that follow-ups submitted after stop are not merged into tool results.""" mock_event = MockEvent("test:FriendMessage:stop_race", "u1") run_context = ContextWrapper(context=MockAgentContext(mock_event)) await runner.reset( provider=mock_provider, request=provider_request, run_context=run_context, tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # Add a follow-up before stop ticket_before_stop = runner.follow_up(message_text="before stop") assert ticket_before_stop is not None # Request stop runner.request_stop() # Try to add a follow-up after stop ticket_after_stop = runner.follow_up(message_text="after stop") assert ticket_after_stop is None, "Follow-up after stop should be rejected" # Verify only the pre-stop follow-up is in the queue assert len(runner._pending_follow_ups) == 1 assert runner._pending_follow_ups[0].text == "before stop" @pytest.mark.asyncio async def test_follow_up_merged_into_tool_result_before_stop( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Test that follow-ups queued before stop are merged into tool results.""" mock_event = MockEvent("test:FriendMessage:merge_before_stop", "u1") run_context = ContextWrapper(context=MockAgentContext(mock_event)) await runner.reset( provider=mock_provider, request=provider_request, run_context=run_context, tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # Queue follow-ups before stop ticket1 = runner.follow_up(message_text="follow up 1 before stop") ticket2 = runner.follow_up(message_text="follow up 2 before stop") assert ticket1 is not None assert ticket2 is not None # Run the agent step (should execute tool and merge follow-ups) async for _ in runner.step(): pass # Verify follow-ups were merged into tool result assert provider_request.tool_calls_result is not None assert isinstance(provider_request.tool_calls_result, list) assert provider_request.tool_calls_result tool_result = str( provider_request.tool_calls_result[0].tool_calls_result[0].content ) # Should contain the follow-up notice assert "SYSTEM NOTICE" in tool_result assert "follow up 1 before stop" in tool_result assert "follow up 2 before stop" in tool_result # Tickets should be marked as consumed assert ticket1.consumed is True assert ticket2.consumed is True @pytest.mark.asyncio async def test_follow_up_rejected_and_runner_stops_without_execution( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Test that when stop is requested before execution, follow-ups are rejected and runner stops gracefully.""" mock_event = MockEvent("test:FriendMessage:stop_before_execution", "u1") run_context = ContextWrapper(context=MockAgentContext(mock_event)) await runner.reset( provider=mock_provider, request=provider_request, run_context=run_context, tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # Request stop before any execution (simulates /stop command received at start) runner.request_stop() assert runner._is_stop_requested() is True # Try to add follow-up after stop (should be rejected) ticket_after = runner.follow_up(message_text="follow-up after stop") assert ticket_after is None, "Post-stop follow-up should be rejected" # Verify queue is empty assert len(runner._pending_follow_ups) == 0 # Run the agent step - should stop immediately without executing tools async for response in runner.step(): # Should yield an aborted response if response.type == "aborted": break # Verify runner stopped gracefully assert runner.done() assert runner.was_aborted() # No tool execution should have occurred assert provider_request.tool_calls_result is None @pytest.mark.asyncio async def test_follow_up_after_stop_not_merged_into_tool_result( runner, mock_provider, provider_request, mock_tool_executor, mock_hooks ): """Regression test for issue #6626: verify post-stop follow-ups are not injected into tool results. This test simulates the race condition where: 1. Runner is active and executing tools 2. A follow-up is queued (should be included in tool result) 3. Stop is requested 4. Another follow-up is attempted (should be rejected) 5. Tool execution completes and merges follow-ups into result The key assertion is that only pre-stop follow-ups are merged into the tool result. """ mock_event = MockEvent("test:FriendMessage:regression_6626", "u1") run_context = ContextWrapper(context=MockAgentContext(mock_event)) await runner.reset( provider=mock_provider, request=provider_request, run_context=run_context, tool_executor=mock_tool_executor, agent_hooks=mock_hooks, streaming=False, ) # Add a follow-up before stop (should be included in tool result) ticket_before = runner.follow_up(message_text="valid before stop") assert ticket_before is not None assert ticket_before in runner._pending_follow_ups # Request stop (simulates /stop command during active execution) runner.request_stop() assert runner._is_stop_requested() is True # Try to add follow-up after stop (should be rejected) ticket_after = runner.follow_up(message_text="invalid after stop") assert ticket_after is None, "Post-stop follow-up should be rejected" # Verify queue only contains pre-stop follow-up assert len(runner._pending_follow_ups) == 1 assert runner._pending_follow_ups[0].text == "valid before stop" # Run the agent step - this will execute tool and merge follow-ups into result async for response in runner.step(): # The runner should execute tools and then stop pass # Verify tool result was created with follow-up merged # Note: When stop is requested, the tool may or may not execute depending on timing. # The key assertion is that IF tool_calls_result exists, it only contains pre-stop follow-ups. if provider_request.tool_calls_result is not None: assert isinstance(provider_request.tool_calls_result, list) assert provider_request.tool_calls_result tool_result = str( provider_request.tool_calls_result[0].tool_calls_result[0].content ) # Should contain the pre-stop follow-up assert "valid before stop" in tool_result # Should NOT contain the post-stop follow-up assert "invalid after stop" not in tool_result assert "after stop" not in tool_result or "after stop" in "valid before stop" # Ticket should be marked as consumed (merged into tool result) assert ticket_before.consumed is True else: # If tool execution was aborted by stop, the ticket should still be resolved # but not consumed (since there was no tool call to merge into) assert ticket_before.resolved.is_set() if __name__ == "__main__": # 运行测试 pytest.main([__file__, "-v"])