From d181176fb7220d42ff38106bc63947e1710e889b Mon Sep 17 00:00:00 2001 From: Thibault Serot Date: Thu, 9 Jul 2026 15:09:21 +1000 Subject: [PATCH] fix(responses): monotonic sequence number for terminal error event; drop orphaned tool events on batch failure Review feedback (Greptile on #32579): - The terminal error event was numbered sequence_number=1, out of order after tool-execution events. __anext__ now tracks the highest sequence_number that passed through the stream and the error event is numbered after it. - A batch tool-execution failure queued mcp_call.in_progress events that never received a terminal per-item event. Those queued events are now dropped; the terminal error event carries the failure instead. Co-Authored-By: Claude Fable 5 --- .../responses/mcp/mcp_streaming_iterator.py | 18 +++++++++++++++++- .../mcp/test_mcp_streaming_iterator.py | 10 ++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index dae7bcbd085..5759960fa13 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/litellm/responses/mcp/mcp_streaming_iterator.py @@ -313,6 +313,10 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): self._initial_creation_error: Optional[Exception] = None self._stream_error: Optional[Exception] = None self._error_event_emitted = False + # Highest sequence_number emitted so far; the terminal `error` event + # must be numbered after it to keep the stream monotonic for strict + # clients. + self._last_sequence_number = 0 def _extract_mcp_headers_from_params(self) -> None: """Extract MCP headers from original request params to pass to tool calls""" @@ -386,7 +390,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): status_code = getattr(err, "status_code", None) return ErrorEvent( type=ResponsesAPIStreamEvents.ERROR, - sequence_number=1, + sequence_number=self._last_sequence_number + 1, error=ErrorEventError( type="mcp_gateway_error", code=str(status_code) if status_code is not None else "internal_error", @@ -399,6 +403,13 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): return self async def __anext__(self) -> ResponsesAPIStreamingResponse: + chunk = await self._anext_impl() + sequence_number = getattr(chunk, "sequence_number", None) + if isinstance(sequence_number, int) and sequence_number > self._last_sequence_number: + self._last_sequence_number = sequence_number + return chunk + + async def _anext_impl(self) -> ResponsesAPIStreamingResponse: """ Phase-based streaming: 1. initial_response - Stream the first LLM response (includes response.created, response.in_progress, response.output_item.added) @@ -733,6 +744,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): traceback.print_exc() self.tool_results = [] + # Drop the queued per-tool events: emitting mcp_call.in_progress + # items that never receive a completed/failed terminal event is a + # protocol deviation. The terminal `error` event carries the + # failure instead. + self.tool_execution_events = [] # Remember the failure. Without this, the follow-up call is made # with function_call items but no function_call_output items and # the provider rejects it with "No tool output found for function diff --git a/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py b/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py index d6cefaed1b6..b1c4b2d2844 100644 --- a/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py +++ b/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py @@ -229,6 +229,9 @@ async def test_tool_execution_failure_emits_error_event_and_skips_follow_up(monk assert "mcp server exploded" in error_events[0].error.message # The doomed follow-up call was never made. aresponses_mock.assert_not_called() + # No orphaned per-tool events: a batch failure must not emit + # mcp_call.in_progress items that never receive a terminal event. + assert all(getattr(c, "type", None) != ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS for c in chunks) @pytest.mark.asyncio @@ -257,6 +260,13 @@ async def test_follow_up_failure_emits_error_event(monkeypatch): assert "No tool output found" in error_events[0].error.message # Tool-execution events were still streamed before the error surfaced. assert any(getattr(c, "type", None) == ResponsesAPIStreamEvents.MCP_CALL_COMPLETED for c in chunks) + # The terminal error event keeps sequence numbers monotonic for strict clients. + prior_sequence_numbers = [ + c.sequence_number + for c in chunks + if isinstance(getattr(c, "sequence_number", None), int) and c is not error_events[0] + ] + assert error_events[0].sequence_number > max(prior_sequence_numbers) @pytest.mark.asyncio