From ef5775e07c7e2cea405f54fd75773dfdd03ad9b1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 24 Jun 2026 14:45:08 +0000 Subject: [PATCH] fix: address focus and streaming edge cases --- .../focus/destinations/mavvrik_destination.py | 8 ++-- .../mavvrik_focus/mavvrik_focus_logger.py | 16 +++++-- litellm/llms/anthropic/chat/handler.py | 7 ++++ litellm/llms/deepseek/chat/transformation.py | 38 +++++++++++++++-- .../chat/test_deepseek_chat_transformation.py | 21 ++++++++++ .../focus/test_mavvrik_destination.py | 42 ++++++++++++++++--- .../chat/test_anthropic_chat_handler.py | 19 +++++++-- 7 files changed, 130 insertions(+), 21 deletions(-) diff --git a/litellm/integrations/focus/destinations/mavvrik_destination.py b/litellm/integrations/focus/destinations/mavvrik_destination.py index 659f608a3e1..b490788a785 100644 --- a/litellm/integrations/focus/destinations/mavvrik_destination.py +++ b/litellm/integrations/focus/destinations/mavvrik_destination.py @@ -288,12 +288,10 @@ class FocusMavvrikDestination(FocusDestination): "Re-enable the connection in the Mavvrik dashboard." ) if resp.status_code >= 400: - verbose_logger.warning( - "Mavvrik FOCUS destination: failed to update metricsMarker (%s): %s", - resp.status_code, - resp.text[:200], + raise RuntimeError( + f"Mavvrik FOCUS destination: failed to update metricsMarker " + f"({resp.status_code}): {resp.text[:200]}" ) - return verbose_logger.debug( "Mavvrik FOCUS destination: metricsMarker advanced to %s", date_epoch ) diff --git a/litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py b/litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py index bbc9d1a6330..398c8b3a696 100644 --- a/litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py +++ b/litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py @@ -72,6 +72,16 @@ def _parse_metrics_marker( return None +def _is_empty_metrics_marker(marker: Optional[object]) -> bool: + if marker is None: + return True + if isinstance(marker, (int, float)): + return marker == 0 + if isinstance(marker, str): + return not marker.strip() + return False + + class MavvrikFocusLogger(FocusLogger): """FOCUS-based export logger that routes to the Mavvrik destination.""" @@ -177,11 +187,9 @@ class MavvrikFocusLogger(FocusLogger): last_ingested = _parse_metrics_marker(marker) - # Catch up missed dates, capped at _MAX_CATCHUP_DAYS. - # last_ingested=None means metricsMarker=0 (fresh connector, never ingested) — - # treat the same as being _MAX_CATCHUP_DAYS behind so we export all available history. + is_empty_marker = _is_empty_metrics_marker(marker) earliest_catchup = yesterday - timedelta(days=self._MAX_CATCHUP_DAYS - 1) - if last_ingested is None or last_ingested < yesterday: + if is_empty_marker or (last_ingested is not None and last_ingested < yesterday): catch_up_date = ( earliest_catchup if last_ingested is None diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py index 18741e2cd0f..82df8f6592d 100644 --- a/litellm/llms/anthropic/chat/handler.py +++ b/litellm/llms/anthropic/chat/handler.py @@ -672,6 +672,13 @@ class ModelResponseIterator: if isinstance(thinking_content, str) and thinking_content: self.reasoning_content_chunks.append(thinking_content) reasoning_content = thinking_content + thinking_blocks = [ + ChatCompletionThinkingBlock( + type="thinking", + thinking=thinking_content, + ) + ] + provider_specific_fields["thinking_blocks"] = thinking_blocks signature = content_block["delta"].get("signature") if isinstance(signature, str) and signature: diff --git a/litellm/llms/deepseek/chat/transformation.py b/litellm/llms/deepseek/chat/transformation.py index a316a3b9260..b90b1e1aa21 100644 --- a/litellm/llms/deepseek/chat/transformation.py +++ b/litellm/llms/deepseek/chat/transformation.py @@ -158,10 +158,8 @@ class DeepSeekChatConfig(OpenAIGPTConfig): through, and drop the now-dangling tool_choice/parallel_tool_calls when nothing callable survives. - Only non-`function` tools are ever dropped, so a `tool_choice` that names - a specific function still points at a surviving tool and is left intact; - `tool_choice`/`parallel_tool_calls` are cleared only when no function - tool remains. + When a specific `tool_choice` points at a dropped tool, clear it so the + sanitized request does not reference a tool DeepSeek will never receive. """ tools = optional_params.get("tools") if not isinstance(tools, list) or not tools: @@ -170,6 +168,28 @@ class DeepSeekChatConfig(OpenAIGPTConfig): def _is_function_tool(tool: object) -> bool: return isinstance(tool, dict) and tool.get("type") == "function" + def _get_function_tool_name(tool: object) -> str | None: + if not isinstance(tool, dict): + return None + function = tool.get("function") + if not isinstance(function, dict): + return None + name = function.get("name") + return name if isinstance(name, str) else None + + def _tool_choice_matches_function_tool( + tool_choice: object, function_tool_names: set[str] + ) -> bool: + if not isinstance(tool_choice, dict): + return True + if tool_choice.get("type") != "function": + return False + function = tool_choice.get("function") + if not isinstance(function, dict): + return False + name = function.get("name") + return isinstance(name, str) and name in function_tool_names + function_tools = [tool for tool in tools if _is_function_tool(tool)] if len(function_tools) == len(tools): return optional_params @@ -189,6 +209,16 @@ class DeepSeekChatConfig(OpenAIGPTConfig): cleaned = {k: v for k, v in optional_params.items() if k != "tools"} if function_tools: + function_tool_names = { + name + for tool in function_tools + for name in (_get_function_tool_name(tool),) + if name is not None + } + if not _tool_choice_matches_function_tool( + cleaned.get("tool_choice"), function_tool_names + ): + cleaned = {k: v for k, v in cleaned.items() if k != "tool_choice"} return {**cleaned, "tools": function_tools} return { k: v diff --git a/tests/litellm/llms/deepseek/chat/test_deepseek_chat_transformation.py b/tests/litellm/llms/deepseek/chat/test_deepseek_chat_transformation.py index a2f45e7188b..66d7e0bcbf9 100644 --- a/tests/litellm/llms/deepseek/chat/test_deepseek_chat_transformation.py +++ b/tests/litellm/llms/deepseek/chat/test_deepseek_chat_transformation.py @@ -166,3 +166,24 @@ class TestDeepSeekThinkingParams: ) assert "thinking" not in result + + def test_drop_unsupported_tools_removes_dangling_tool_choice(self): + optional_params = { + "tools": [ + {"type": "namespace", "name": "local_shell"}, + {"type": "function", "function": {"name": "get_weather"}}, + ], + "tool_choice": { + "type": "function", + "function": {"name": "local_shell"}, + }, + "parallel_tool_calls": True, + } + + result = self.config._drop_unsupported_tools(optional_params) + + assert result["tools"] == [ + {"type": "function", "function": {"name": "get_weather"}} + ] + assert "tool_choice" not in result + assert result["parallel_tool_calls"] is True diff --git a/tests/test_litellm/integrations/focus/test_mavvrik_destination.py b/tests/test_litellm/integrations/focus/test_mavvrik_destination.py index 3a23dc4ffb2..9f3863127ee 100644 --- a/tests/test_litellm/integrations/focus/test_mavvrik_destination.py +++ b/tests/test_litellm/integrations/focus/test_mavvrik_destination.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -574,6 +574,39 @@ async def test_run_scheduled_export_no_catchup_when_marker_is_current(): ) +@pytest.mark.asyncio +async def test_run_scheduled_export_skips_catchup_when_marker_is_unparseable(): + import polars as pl + from litellm.integrations.mavvrik_focus.mavvrik_focus_logger import ( + MavvrikFocusLogger, + ) + from litellm.integrations.focus.destinations.mavvrik_destination import ( + FocusMavvrikDestination, + ) + + logger = MavvrikFocusLogger() + now = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) + yesterday = now - timedelta(days=1) + + dest_mock = MagicMock(spec=FocusMavvrikDestination) + dest_mock.get_metrics_marker = AsyncMock(return_value="not-a-date") + + db_mock = MagicMock() + db_mock.get_usage_data = AsyncMock(return_value=pl.DataFrame()) + engine_mock = MagicMock() + engine_mock._database = db_mock + engine_mock._destination = dest_mock + logger._engine = engine_mock + + await logger._run_scheduled_export() + + assert db_mock.get_usage_data.call_count == 1 + assert ( + db_mock.get_usage_data.call_args.kwargs["start_time_utc"].date() + == yesterday.date() + ) + + @pytest.mark.asyncio async def test_metrics_marker_always_calls_api(): """get_metrics_marker must call the register API every time to get a fresh marker. @@ -774,8 +807,7 @@ async def test_gcs_session_cancelled_on_chunk_failure(): @pytest.mark.asyncio -async def test_update_metrics_marker_warns_on_non_410_error(): - """_update_metrics_marker must log a warning on any >=400 (non-410) status but not raise.""" +async def test_update_metrics_marker_raises_on_non_410_error(): dest = _dest() fail_resp = MagicMock() @@ -787,8 +819,8 @@ async def test_update_metrics_marker_warns_on_non_410_error(): mock_http.client.request = AsyncMock(return_value=fail_resp) dest._http = mock_http - # Must not raise — warning only - await dest._update_metrics_marker(1234567890) + with pytest.raises(RuntimeError, match="failed to update metricsMarker"): + await dest._update_metrics_marker(1234567890) assert mock_http.client.request.call_count == 1 diff --git a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py index 0b1aaf87516..f934c7184f8 100644 --- a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py +++ b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py @@ -113,6 +113,10 @@ def test_streaming_thinking_blocks_are_replayable_after_signature_delta(): for chunk in parsed_chunks for block in (getattr(chunk.choices[0].delta, "thinking_blocks", None) or []) ) + expected_delta_blocks = ( + {"type": "thinking", "thinking": "Step 1. "}, + {"type": "thinking", "thinking": "Step 2."}, + ) expected_thinking_block = { "type": "thinking", "thinking": "Step 1. Step 2.", @@ -120,7 +124,10 @@ def test_streaming_thinking_blocks_are_replayable_after_signature_delta(): } assert reasoning_content == "Step 1. Step 2." - assert thinking_blocks == (expected_thinking_block,) + assert thinking_blocks == (*expected_delta_blocks, expected_thinking_block) + assert parsed_chunks[1].choices[0].delta.provider_specific_fields == { + "thinking_blocks": [expected_delta_blocks[0]] + } assert parsed_chunks[-1].choices[0].delta.provider_specific_fields == { "thinking_blocks": [expected_thinking_block] } @@ -163,7 +170,10 @@ def test_streaming_unsigned_thinking_deltas_keep_reasoning_content(): ) assert reasoning_content == "Step 1. Step 2." - assert thinking_blocks == () + assert thinking_blocks == ( + {"type": "thinking", "thinking": "Step 1. "}, + {"type": "thinking", "thinking": "Step 2."}, + ) def test_streaming_truncated_thinking_deltas_keep_reasoning_content(): @@ -202,7 +212,10 @@ def test_streaming_truncated_thinking_deltas_keep_reasoning_content(): ) assert reasoning_content == "Step 1. Step 2." - assert thinking_blocks == () + assert thinking_blocks == ( + {"type": "thinking", "thinking": "Step 1. "}, + {"type": "thinking", "thinking": "Step 2."}, + ) def test_handle_json_mode_chunk_response_format_tool():