fix: sanitize incomplete tool history before provider fallback

astrbot/core/agent/context/manager.py

@@ -53,8 +53,15 @@ async def process(
         Returns:
             The processed message list.
         """
+        result = messages
         try:
-            result = messages
+            result = self.truncator.fix_messages(messages)
+            history_was_sanitized = result != messages
+            if history_was_sanitized:
+                logger.debug(
+                    f"Removed {len(messages) - len(result)} invalid tool history "
+                    "message(s) before context processing."
+                )
 
             # 1. 基于轮次的截断 (Enforce max turns)
             if self.config.enforce_max_turns != -1:
@@ -67,7 +74,7 @@ async def process(
             # 2. 基于 token 的压缩
             if self.config.max_context_tokens > 0:
                 total_tokens = self.token_counter.count_tokens(
-                    result, trusted_token_usage
+                    result, 0 if history_was_sanitized else trusted_token_usage
                 )
 
                 if self.compressor.should_compress(
@@ -78,7 +85,7 @@ async def process(
             return result
         except Exception as e:
             logger.error(f"Error during context processing: {e}", exc_info=True)
-            return messages
+            return result
 
     async def _run_compression(
         self, messages: list[Message], prev_tokens: int

astrbot/core/agent/context/truncator.py

@@ -53,7 +53,8 @@ def fix_messages(self, messages: list[Message]) -> list[Message]:
 
         This method ensures that:
         1. Each `tool` message is preceded by an `assistant` message containing `tool_calls`.
-        2. Each `assistant` message containing `tool_calls` is followed by corresponding `
+        2. Each `assistant` message containing `tool_calls` is followed by exactly one
+           `tool` message for every tool call ID.
 
         This is a requirement of the OpenAI Chat Completions API specification (Gemini enforces this strictly).
         """
@@ -66,9 +67,32 @@ def fix_messages(self, messages: list[Message]) -> list[Message]:
 
         def flush_pending_if_valid() -> None:
             nonlocal pending_assistant, pending_tools
-            if pending_assistant is not None and pending_tools:
-                fixed_messages.append(pending_assistant)
-                fixed_messages.extend(pending_tools)
+            if pending_assistant is not None:
+                expected_ids = []
+                for tool_call in pending_assistant.tool_calls or []:
+                    if isinstance(tool_call, dict):
+                        tool_call_id = tool_call.get("id")
+                    else:
+                        tool_call_id = tool_call.id
+                    if not isinstance(tool_call_id, str) or not tool_call_id:
+                        expected_ids = []
+                        break
+                    expected_ids.append(tool_call_id)
+                result_ids = [tool.tool_call_id for tool in pending_tools]
+                has_valid_expected_ids = bool(expected_ids) and len(
+                    expected_ids
+                ) == len(set(expected_ids))
+                has_valid_result_ids = (
+                    len(result_ids) == len(expected_ids)
+                    and all(
+                        isinstance(tool_id, str) and tool_id for tool_id in result_ids
+                    )
+                    and len(result_ids) == len(set(result_ids))
+                )
+                ids_match = set(result_ids) == set(expected_ids)
+                if has_valid_expected_ids and has_valid_result_ids and ids_match:
+                    fixed_messages.append(pending_assistant)
+                    fixed_messages.extend(pending_tools)
             pending_assistant = None
             pending_tools = []
 

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -1096,11 +1096,19 @@ async def _handle_function_tools(
         logger.info(f"Agent 使用工具: {llm_response.tools_call_name}")
 
         def _append_tool_call_result(tool_call_id: str, content: str) -> None:
+            content = self._merge_follow_up_notice(content)
+            if (
+                len(tool_call_result_blocks) > tool_result_blocks_start
+                and tool_call_result_blocks[-1].tool_call_id == tool_call_id
+            ):
+                previous = tool_call_result_blocks[-1]
+                previous.content = f"{previous.content}\n\n{content}"
+                return
             tool_call_result_blocks.append(
                 ToolCallMessageSegment(
                     role="tool",
                     tool_call_id=tool_call_id,
-                    content=self._merge_follow_up_notice(content),
+                    content=content,
                 ),
             )
 

astrbot/core/provider/sources/gemini_source.py

@@ -322,6 +322,7 @@ def append_or_extend(
                 contents.append(content_cls(parts=part))
 
         gemini_contents: list[types.Content] = []
+        tool_name_by_call_id: dict[str, str] = {}
         for message in payloads["messages"]:
             role, content = message["role"], message.get("content")
 
@@ -392,6 +393,11 @@ def append_or_extend(
 
                 if "tool_calls" in message:
                     for tool in message["tool_calls"]:
+                        tool_call_id = tool.get("id")
+                        if isinstance(tool_call_id, str) and tool_call_id:
+                            tool_name_by_call_id[tool_call_id] = tool["function"][
+                                "name"
+                            ]
                         part = types.Part.from_function_call(
                             name=tool["function"]["name"],
                             args=json.loads(tool["function"]["arguments"]),
@@ -415,7 +421,12 @@ def append_or_extend(
                 append_or_extend(gemini_contents, parts, types.ModelContent)
 
             elif role == "tool":
-                func_name = message.get("name", message["tool_call_id"])
+                tool_call_id = message["tool_call_id"]
+                func_name = (
+                    message.get("name")
+                    or tool_name_by_call_id.get(tool_call_id)
+                    or tool_call_id
+                )
                 part = types.Part.from_function_response(
                     name=func_name,
                     response={
@@ -547,8 +558,13 @@ def _process_content_parts(
                 llm_response.role = "tool"
                 llm_response.tools_call_name.append(part.function_call.name)
                 llm_response.tools_call_args.append(part.function_call.args)
-                # function_call.id might be None, use name as fallback
-                tool_call_id = part.function_call.id or part.function_call.name
+                # function_call.id might be None, use a unique name-based fallback.
+                base_tool_call_id = part.function_call.id or part.function_call.name
+                tool_call_id = base_tool_call_id
+                duplicate_index = 2
+                while tool_call_id in llm_response.tools_call_ids:
+                    tool_call_id = f"{base_tool_call_id}__astrbot_{duplicate_index}"
+                    duplicate_index += 1
                 llm_response.tools_call_ids.append(tool_call_id)
                 # extra_content
                 if part.thought_signature:

tests/agent/test_context_manager.py

@@ -412,6 +412,37 @@ async def test_process_with_no_limits(self):
         assert len(result) == 20
         assert result == messages
 
+    @pytest.mark.asyncio
+    async def test_process_fixes_incomplete_tool_history_without_limits(self):
+        """Provider-facing history is valid even when no size limit is enabled."""
+        config = ContextConfig(max_context_tokens=0, enforce_max_turns=-1)
+        manager = ContextManager(config)
+        messages = [
+            self.create_message("user", "Run both tools"),
+            Message(
+                role="assistant",
+                content="Calling tools",
+                tool_calls=[
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "first", "arguments": "{}"},
+                    },
+                    {
+                        "id": "call_2",
+                        "type": "function",
+                        "function": {"name": "second", "arguments": "{}"},
+                    },
+                ],
+            ),
+            Message(role="tool", content="first result", tool_call_id="call_1"),
+            self.create_message("user", "Continue"),
+        ]
+
+        result = await manager.process(messages)
+
+        assert result == [messages[0], messages[-1]]
+
     # ==================== Enforce Max Turns Tests ====================
 
     @pytest.mark.asyncio
@@ -546,6 +577,44 @@ async def test_trusted_usage_triggers_compression_before_provider_call(self):
         mock_compressor.assert_awaited_once_with(messages)
         assert result == compressed
 
+    @pytest.mark.asyncio
+    async def test_sanitized_history_does_not_reuse_stale_trusted_usage(self):
+        config = ContextConfig(max_context_tokens=100, truncate_turns=1)
+        manager = ContextManager(config)
+        messages = [
+            self.create_message("user", "old request"),
+            Message(
+                role="assistant",
+                content="Calling tools",
+                tool_calls=[
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "first", "arguments": "{}"},
+                    },
+                    {
+                        "id": "call_2",
+                        "type": "function",
+                        "function": {"name": "second", "arguments": "{}"},
+                    },
+                ],
+            ),
+            Message(role="tool", content="first result", tool_call_id="call_1"),
+            self.create_message("user", "current request"),
+        ]
+        sanitized = [messages[0], messages[-1]]
+        mock_compressor = AsyncMock()
+        mock_compressor.should_compress = MagicMock(return_value=False)
+        manager.compressor = mock_compressor
+
+        result = await manager.process(messages, trusted_token_usage=83)
+
+        first_check = mock_compressor.should_compress.call_args_list[0]
+        expected_tokens = manager.token_counter.count_tokens(sanitized)
+        assert first_check.args == (sanitized, expected_tokens, 100)
+        mock_compressor.assert_not_awaited()
+        assert result == sanitized
+
     @pytest.mark.asyncio
     async def test_token_compression_with_zero_max_tokens(self):
         """Test that compression is skipped when max_context_tokens is 0."""
@@ -655,13 +724,48 @@ async def test_error_handling_returns_original_messages(self):
 
         # Make compressor raise an exception
         with patch.object(
-            manager.compressor, "__call__", side_effect=Exception("Test error")
+            manager, "_run_compression", side_effect=Exception("Test error")
         ):
             result = await manager.process(messages)
 
             # Should return original messages despite error
             assert result == messages
 
+    @pytest.mark.asyncio
+    async def test_error_handling_keeps_tool_history_sanitized(self):
+        """Compression errors must not restore an invalid tool history block."""
+        config = ContextConfig(max_context_tokens=1)
+        manager = ContextManager(config)
+        manager.compressor.should_compress = MagicMock(return_value=True)
+        messages = [
+            self.create_message("user", "Run both tools"),
+            Message(
+                role="assistant",
+                content="Calling tools",
+                tool_calls=[
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "first", "arguments": "{}"},
+                    },
+                    {
+                        "id": "call_2",
+                        "type": "function",
+                        "function": {"name": "second", "arguments": "{}"},
+                    },
+                ],
+            ),
+            Message(role="tool", content="first result", tool_call_id="call_1"),
+            self.create_message("user", "Continue"),
+        ]
+
+        with patch.object(
+            manager, "_run_compression", side_effect=Exception("Test error")
+        ):
+            result = await manager.process(messages)
+
+        assert result == [messages[0], messages[-1]]
+
     @pytest.mark.asyncio
     async def test_error_handling_logs_exception(self):
         """Test that errors are logged."""

tests/agent/test_truncator.py

@@ -62,6 +62,65 @@ def test_fix_messages_tool_without_context(self):
         # Tool message without context should be removed
         assert len(result) == 0
 
+    def test_fix_messages_keeps_complete_multi_tool_block(self):
+        """Keep a tool block when every call has exactly one matching result."""
+        truncator = ContextTruncator()
+        messages = [
+            self.create_message("user", "Run both tools"),
+            Message(
+                role="assistant",
+                content="Calling tools",
+                tool_calls=[
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "first", "arguments": "{}"},
+                    },
+                    {
+                        "id": "call_2",
+                        "type": "function",
+                        "function": {"name": "second", "arguments": "{}"},
+                    },
+                ],
+            ),
+            Message(role="tool", content="second result", tool_call_id="call_2"),
+            Message(role="tool", content="first result", tool_call_id="call_1"),
+            self.create_message("assistant", "Done"),
+        ]
+
+        result = truncator.fix_messages(messages)
+
+        assert result == messages
+
+    def test_fix_messages_drops_incomplete_multi_tool_block(self):
+        """Drop the whole block when one of multiple tool results is missing."""
+        truncator = ContextTruncator()
+        messages = [
+            self.create_message("user", "Run both tools"),
+            Message(
+                role="assistant",
+                content="Calling tools",
+                tool_calls=[
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "first", "arguments": "{}"},
+                    },
+                    {
+                        "id": "call_2",
+                        "type": "function",
+                        "function": {"name": "second", "arguments": "{}"},
+                    },
+                ],
+            ),
+            Message(role="tool", content="first result", tool_call_id="call_1"),
+            self.create_message("user", "Continue"),
+        ]
+
+        result = truncator.fix_messages(messages)
+
+        assert result == [messages[0], messages[-1]]
+
     # ==================== truncate_by_turns Tests ====================
 
     def test_truncate_by_turns_no_limit(self):