fix: sanitize incomplete tool history before provider fallback

astrbot/core/agent/context/manager.py

@@ -53,8 +53,14 @@ async def process(
         Returns:
             The processed message list.
         """
+        result = messages
         try:
-            result = messages
+            result = self.truncator.fix_messages(messages)
+            if len(result) != len(messages):
+                logger.warning(
+                    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:
@@ -78,7 +84,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,30 @@ 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]
+                if (
+                    expected_ids
+                    and len(expected_ids) == len(set(expected_ids))
+                    and 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))
+                    and set(result_ids) == set(expected_ids)
+                ):
+                    fixed_messages.append(pending_assistant)
+                    fixed_messages.extend(pending_tools)
             pending_assistant = None
             pending_tools = []
 

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
@@ -655,13 +686,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):

tests/test_tool_loop_agent_runner.py

@@ -183,6 +183,16 @@ async def text_chat(self, **kwargs) -> LLMResponse:
         raise RuntimeError("primary provider failed")
 
 
+class CapturingFailingProvider(MockFailingProvider):
+    def __init__(self):
+        super().__init__()
+        self.received_contexts = []
+
+    async def text_chat(self, **kwargs) -> LLMResponse:
+        self.received_contexts.append(list(kwargs.get("contexts") or []))
+        return await super().text_chat(**kwargs)
+
+
 class MockErrProvider(MockProvider):
     async def text_chat(self, **kwargs) -> LLMResponse:
         self.call_count += 1
@@ -1213,6 +1223,61 @@ async def test_fallback_provider_used_when_primary_raises(
     assert fallback_provider.call_count == 1
 
 
+@pytest.mark.asyncio
+async def test_fallback_providers_receive_only_complete_tool_history(
+    runner, mock_tool_executor, mock_hooks
+):
+    primary_provider = CapturingFailingProvider()
+    fallback_provider = CapturingProvider(modalities=[])
+    request = ProviderRequest(
+        prompt="Continue",
+        contexts=[
+            {"role": "user", "content": "Run both tools"},
+            {
+                "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": "{}"},
+                    },
+                ],
+            },
+            {"role": "tool", "content": "first result", "tool_call_id": "call_1"},
+        ],
+    )
+
+    await runner.reset(
+        provider=primary_provider,
+        request=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
+
+    for contexts in [
+        primary_provider.received_contexts[0],
+        fallback_provider.received_contexts[0],
+    ]:
+        assert all(message.role != "tool" for message in contexts)
+        assert all(not message.tool_calls for message in contexts)
+
+    final_resp = runner.get_final_llm_resp()
+    assert final_resp is not None
+    assert final_resp.completion_text == "final"
+
+
 @pytest.mark.asyncio
 async def test_fallback_provider_used_when_primary_returns_err(
     runner, provider_request, mock_tool_executor, mock_hooks