diff --git a/.github/workflows/_test-unit-base.yml b/.github/workflows/_test-unit-base.yml index 6f6822a975b..75b0f93fd77 100644 --- a/.github/workflows/_test-unit-base.yml +++ b/.github/workflows/_test-unit-base.yml @@ -55,17 +55,14 @@ on: permissions: contents: read +env: + UV_PYTHON: "3.12" + jobs: run: - name: ${{ matrix.python-version == '3.12' && 'Run tests' || format('Run tests (Python {0})', matrix.python-version) }} + name: Run tests runs-on: ubuntu-latest timeout-minutes: ${{ inputs.job-timeout-minutes }} - strategy: - fail-fast: false - matrix: - python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] - env: - UV_PYTHON: ${{ matrix.python-version }} permissions: contents: read pull-requests: read @@ -88,7 +85,7 @@ jobs: timeout-minutes: 3 uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: - python-version: ${{ matrix.python-version }} + python-version: ${{ env.UV_PYTHON }} - name: Set up uv if: steps.changes.outputs.decision != 'skip' @@ -103,9 +100,9 @@ jobs: uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0 with: path: ${{ env.UV_CACHE_DIR }} - key: ${{ runner.os }}-uv-downloads-py${{ matrix.python-version }}-${{ hashFiles('uv.lock') }} + key: ${{ runner.os }}-uv-downloads-py${{ env.UV_PYTHON }}-${{ hashFiles('uv.lock') }} restore-keys: | - ${{ runner.os }}-uv-downloads-py${{ matrix.python-version }}- + ${{ runner.os }}-uv-downloads-py${{ env.UV_PYTHON }}- - name: Cache the Rust build if: steps.changes.outputs.decision != 'skip' @@ -139,7 +136,7 @@ jobs: WORKERS: ${{ inputs.workers }} RERUNS: ${{ inputs.reruns }} DIST: ${{ inputs.dist }} - COVERAGE_CORE: ${{ contains(fromJSON('["3.10", "3.11"]'), matrix.python-version) && 'ctrace' || 'sysmon' }} + COVERAGE_CORE: sysmon run: | if [ "${WORKERS}" = "0" ]; then uv run --no-sync pytest ${TEST_PATH:?} \ @@ -166,7 +163,7 @@ jobs: fi - name: Save coverage report - if: always() && matrix.python-version == '3.12' && steps.changes.outputs.decision != 'skip' + if: always() && steps.changes.outputs.decision != 'skip' uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1 with: name: coverage-${{ inputs.artifact-name }}-${{ github.run_id }}-${{ github.run_attempt }} diff --git a/litellm/batches/batch_utils.py b/litellm/batches/batch_utils.py index 97be5f77d79..959c7498479 100644 --- a/litellm/batches/batch_utils.py +++ b/litellm/batches/batch_utils.py @@ -25,6 +25,27 @@ class BatchCostUsageResult: failed_requests: int +_COMPLETED_BATCH_STATUSES: Final = frozenset({"completed", "complete"}) +_TERMINAL_BATCH_STATUSES: Final = _COMPLETED_BATCH_STATUSES | frozenset({"failed", "cancelled", "expired"}) + + +def batch_cost_is_final(batch: Batch) -> bool: + """Whether this retrieve of the batch is the one to account its cost from. + + A batch still in flight has nothing to price, and a "completed" batch can report + no output_file_id for a moment before the output populates; pricing either records + $0 under the batch's single spend row and pins it there. Final means a completed + batch whose output file has arrived or whose counts prove no line succeeded, or + any other terminal status (failed, cancelled, expired). + """ + if batch.status not in _TERMINAL_BATCH_STATUSES: + return False + if batch.status not in _COMPLETED_BATCH_STATUSES or batch.output_file_id is not None: + return True + request_counts: Final = batch.request_counts + return request_counts is not None and request_counts.total > 0 and request_counts.completed == 0 + + async def calculate_batch_cost_and_usage( file_content_dictionary: list[dict], custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"], diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 989777e9412..0f0392ebb48 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -36,7 +36,7 @@ from litellm._logging import ( verbose_logger, ) from litellm._uuid import uuid -from litellm.batches.batch_utils import _handle_completed_batch +from litellm.batches.batch_utils import _handle_completed_batch, batch_cost_is_final from litellm.caching.caching import DualCache, InMemoryCache from litellm.caching.caching_handler import LLMCachingHandler from litellm.constants import ( @@ -2899,13 +2899,6 @@ class Logging(LiteLLMLoggingBaseClass): ): # polling job will query these frequently, don't spam db logs return - from litellm.proxy.openai_files_endpoints.common_utils import ( - _is_base64_encoded_unified_file_id, - ) - - # check if file id is a unified file id - is_base64_unified_file_id: Final = _is_base64_encoded_unified_file_id(result.id) - batch_cost: Final = kwargs.get("batch_cost", None) batch_usage = kwargs.get("batch_usage", None) batch_models = kwargs.get("batch_models", None) @@ -2913,9 +2906,7 @@ class Logging(LiteLLMLoggingBaseClass): batch_failed_requests: Final = kwargs.get("batch_failed_requests", None) has_explicit_batch_data: Final = all(x is not None for x in (batch_cost, batch_usage, batch_models)) - should_compute_batch_data: Final = ( - not is_base64_unified_file_id or not has_explicit_batch_data and result.status == "completed" - ) + should_compute_batch_data: Final = not has_explicit_batch_data and batch_cost_is_final(result) if has_explicit_batch_data: result._hidden_params["response_cost"] = batch_cost result._hidden_params["batch_models"] = batch_models diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index 32b8cbe6343..9158ff4569f 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -11,6 +11,7 @@ from typing import ( Final, Literal, Protocol, + cast, # noqa: TID251 # rebuilt message_delta dict spans the ContentBlockDelta/MessageBlockDelta union get_args, ) @@ -100,6 +101,10 @@ class _CombinedChunkSplitter: @staticmethod def _is_combined(chunk: "ModelResponseStream") -> bool: """True if ``chunk`` carries response content AND a finish_reason.""" + from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + openai_chat_refusal_text, + ) + choices: Final = _optional_attr_sequence(chunk, "choices") if not choices: return False @@ -114,6 +119,7 @@ class _CombinedChunkSplitter: or _optional_attr(delta, "tool_calls") or _optional_attr(delta, "reasoning_content") or _optional_attr(delta, "thinking_blocks") + or openai_chat_refusal_text(delta) ) _PAYLOAD_FIELD_GROUPS: "tuple[tuple[str, ...], ...]" = ( @@ -305,6 +311,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): # Synthesized compaction block from compact_20260112 polyfill (streaming). self.compaction_block = compaction_block self.iterations_usage = iterations_usage + self._refusal_text: str = "" self.sent_compaction_block: bool = False # Per-phase flags so the compaction block's start/delta/stop events # are emitted (and the public state machine is advanced) in @@ -572,6 +579,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): current_content_block_index=self.current_content_block_index, applied_edits=(self.applied_edits if is_final_chunk and not will_merge_into_held else None), ) + processed_chunk = self._with_refusal_stop_details(processed_chunk) # Check if this is a usage chunk and we have a held stop_reason chunk if will_merge_into_held: @@ -806,6 +814,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): current_content_block_index=self.current_content_block_index, applied_edits=(self.applied_edits if is_final_chunk and not will_merge_into_held else None), ) + processed_chunk = self._with_refusal_stop_details(processed_chunk) # Check if this is a usage chunk and we have a held stop_reason chunk if will_merge_into_held: @@ -993,6 +1002,31 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): def _increment_content_block_index(self): self.current_content_block_index += 1 + def _with_refusal_stop_details( + self, + processed_chunk: ContentBlockDelta | MessageBlockDelta, + ) -> ContentBlockDelta | MessageBlockDelta: + if processed_chunk.get("type") != "message_delta" or not self._refusal_text: + return processed_chunk + delta: Final = cast(Mapping[str, object], processed_chunk["delta"]) # cast-ok: keys checked before use + if delta.get("stop_reason") == "max_tokens": + return processed_chunk + from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + refusal_stop_details, + ) + + return cast( # cast-ok: rebuilt dict matches the message_delta TypedDict shape for this branch + ContentBlockDelta | MessageBlockDelta, + { # mutable-ok: fresh translation payload; never mutated after construction + **processed_chunk, + "delta": { # mutable-ok: fresh message_delta payload; never mutated after construction + **delta, + "stop_reason": "refusal", + "stop_details": refusal_stop_details(self._refusal_text), + }, + }, + ) + @staticmethod def _delta_has_content(processed_chunk: Mapping[str, object]) -> bool: """Return True if a translated chunk carries a non-empty @@ -1035,6 +1069,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): @staticmethod def _is_blank_delta(chunk: "ModelResponseStream") -> bool: from litellm.llms.anthropic.common_utils import is_empty_unsigned_thinking_block + from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + openai_chat_refusal_text, + ) choice: Final = chunk.choices[0] if choice.finish_reason is not None: @@ -1044,6 +1081,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): return False if getattr(delta, "content", None): return False + if openai_chat_refusal_text(delta): + return False if getattr(delta, "reasoning_content", None): return False # thinking_blocks whose entries are all empty AND unsigned must not @@ -1067,13 +1106,19 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): - Different content types in the response - Specific markers in the content """ + from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + openai_chat_refusal_text, + ) + from .transformation import LiteLLMAnthropicMessagesAdapter - # Example logic - customize based on your needs: - # If chunk indicates a tool call if chunk.choices[0].finish_reason is not None: return False + refusal_text: Final = openai_chat_refusal_text(chunk.choices[0].delta) + if refusal_text is not None: + self._refusal_text = self._refusal_text + refusal_text + ( block_type, content_block_start, diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py index f450f3899f4..db890662132 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py @@ -117,6 +117,10 @@ from litellm.llms.anthropic.common_utils import ( from litellm.llms.anthropic.experimental_pass_through.context_management import ( PolyfillResult, ) +from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + openai_chat_refusal_text, + refusal_stop_details, +) from litellm.types.llms.anthropic import ( ANTHROPIC_HOSTED_TOOLS, AllAnthropicPassThroughMessageValues, @@ -1314,6 +1318,8 @@ class LiteLLMAnthropicMessagesAdapter: new_content.append( AnthropicResponseContentBlockText(type="text", text=choice.message.content).model_dump() ) + if (refusal_text := openai_chat_refusal_text(choice.message)) is not None: + new_content.append(AnthropicResponseContentBlockText(type="text", text=refusal_text).model_dump()) # Handle tool calls (in parallel to text content) if choice.message.tool_calls is not None and len(choice.message.tool_calls) > 0: for tool_call in choice.message.tool_calls: @@ -1472,14 +1478,23 @@ class LiteLLMAnthropicMessagesAdapter: choices=response.choices, tool_name_mapping=tool_name_mapping, ) + refusal_text: Final = next( + (text for choice in response.choices if (text := openai_chat_refusal_text(choice.message)) is not None), + None, + ) if polyfill_result is not None and polyfill_result.compaction_block is not None: anthropic_content.insert(0, polyfill_result.compaction_block) ## extract finish reason - anthropic_finish_reason: Final = self._translate_openai_finish_reason_to_anthropic( + translated_finish_reason: Final = self._translate_openai_finish_reason_to_anthropic( openai_finish_reason=response.choices[0].finish_reason ) + anthropic_finish_reason: Final = ( + "refusal" + if refusal_text is not None and translated_finish_reason != "max_tokens" + else translated_finish_reason + ) # extract usage usage: Final[Usage] = getattr(response, "usage") anthropic_usage: Final = self._translate_openai_usage_to_anthropic_usage(usage) @@ -1501,6 +1516,7 @@ class LiteLLMAnthropicMessagesAdapter: usage=anthropic_usage, content=anthropic_content, stop_reason=anthropic_finish_reason, + stop_details=(refusal_stop_details(refusal_text) if anthropic_finish_reason == "refusal" else None), ) applied_edits: Final = polyfill_result.applied_edits_for_response() if polyfill_result else None @@ -1541,7 +1557,9 @@ class LiteLLMAnthropicMessagesAdapter: "signature": thought_sig, } return "tool_use", cast("ContentBlockContentBlockDict", tool_block) - elif choice.delta.content is not None and len(choice.delta.content) > 0: + elif (choice.delta.content is not None and len(choice.delta.content) > 0) or openai_chat_refusal_text( + choice.delta + ) is not None: return "text", TextBlock(type="text", text="") elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"): thinking_blocks = choice.delta.thinking_blocks or [] @@ -1613,7 +1631,10 @@ class LiteLLMAnthropicMessagesAdapter: elif reasoning_content: return "thinking_delta", ContentThinkingBlockDelta(type="thinking_delta", thinking=reasoning_content) else: - return "text_delta", ContentTextBlockDelta(type="text_delta", text=text) + refusal_text: Final = "".join( + refusal for choice in choices if (refusal := openai_chat_refusal_text(choice.delta)) is not None + ) + return "text_delta", ContentTextBlockDelta(type="text_delta", text=text + refusal_text) def translate_streaming_openai_response_to_anthropic( self, diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/utils.py b/litellm/llms/anthropic/experimental_pass_through/messages/utils.py index 242300c7b6d..7545dff1408 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/utils.py +++ b/litellm/llms/anthropic/experimental_pass_through/messages/utils.py @@ -1,8 +1,11 @@ -from collections.abc import Mapping +from collections.abc import Iterable, Mapping, Sequence from functools import lru_cache from typing import TYPE_CHECKING, Any, Final, cast, get_type_hints -from litellm.types.llms.anthropic import AnthropicMessagesRequestOptionalParams +from litellm.types.llms.anthropic import ( + AnthropicMessagesRequestOptionalParams, + AnthropicStopDetails, +) from litellm.types.llms.anthropic_messages.anthropic_response import ( AnthropicMessagesResponse, ) @@ -25,6 +28,69 @@ def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, Any] | return stop_details if isinstance(stop_details, dict) else None +def refusal_stop_details(explanation: str | None) -> AnthropicStopDetails: + """The ``stop_details`` object accompanying a translated ``stop_reason: "refusal"``.""" + return AnthropicStopDetails(type="refusal", category=None, explanation=explanation) + + +def _mapping_field(container: object, key: str) -> object | None: + """One key of a raw provider payload, or None when the payload is not a mapping.""" + if not isinstance(container, Mapping): + return None + return cast(Mapping[str, object], container).get(key) # cast-ok: raw payload, callers re-check every value + + +def _mapping_str_field(container: object, key: str) -> str | None: + value: Final = _mapping_field(container, key) + return value if isinstance(value, str) and value else None + + +def openai_chat_refusal_text(message_or_delta: object) -> str | None: + """ + Refusal text carried by an OpenAI Chat Completions message or streaming delta, + read from ``refusal`` or from the ``provider_specific_fields`` LiteLLM parks it + in, or None when the turn is not a refusal. + """ + refusal: Final = getattr(message_or_delta, "refusal", None) + if isinstance(refusal, str) and refusal: + return refusal + return _mapping_str_field(getattr(message_or_delta, "provider_specific_fields", None), "refusal") + + +def _responses_message_refusal_text(item: object) -> str | None: + from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal + + if isinstance(item, ResponseOutputMessage): + return next( + (part.refusal for part in item.content if isinstance(part, ResponseOutputRefusal) and part.refusal), + None, + ) + raw_parts: Final = _mapping_field(item, "content") + if _mapping_str_field(item, "type") != "message" or not isinstance(raw_parts, Sequence): + return None + return next( + ( + refusal + for part in cast(Sequence[object], raw_parts) # cast-ok: members re-validated below + if _mapping_str_field(part, "type") == "refusal" + and isinstance(refusal := _mapping_str_field(part, "refusal"), str) + ), + None, + ) + + +def responses_output_refusal_text(output: Iterable[object]) -> str | None: + """ + Refusal text carried by an OpenAI Responses ``output`` list, in typed + (``ResponseOutputRefusal``) or raw-dictionary shape, or None when none of the + output messages refused. + """ + return next( + (text for item in output if (text := _responses_message_refusal_text(item)) is not None), + None, + ) + + def safeguard_refusal_error(model: str, stop_details: Mapping[str, object]) -> "ContentPolicyViolationError": """The exception a safeguard-refused Anthropic response converts into so the content-policy fallback chain can re-dispatch it.""" diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py index b0aba386753..2e0a6a9df8f 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py @@ -1,13 +1,18 @@ # What is this? ## Translates OpenAI call to Anthropic `/v1/messages` format +import asyncio import json import traceback from collections import deque -from collections.abc import AsyncIterator, Mapping +from collections.abc import AsyncIterator, Iterator, Mapping from typing import TYPE_CHECKING, Any, Final from litellm import verbose_logger from litellm._uuid import uuid +from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + refusal_stop_details, + responses_output_refusal_text, +) from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage from .transformation import LiteLLMAnthropicToResponsesAPIAdapter @@ -49,6 +54,8 @@ class AnthropicResponsesStreamWrapper: self._sent_message_start = False self._sent_message_stop = False self._chunk_queue: deque[dict[str, object]] = deque() + self._refusal_text: str = "" + self._sync_responses_iterator: Iterator[object] | None = None def _make_message_start(self) -> dict[str, object]: return { @@ -131,6 +138,24 @@ class AnthropicResponsesStreamWrapper: ) return + if event_type == "response.refusal.delta": + delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "") + if not isinstance(delta, str) or not delta: + return + self._refusal_text = self._refusal_text + delta + item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None) + block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index + if block_idx < 0: + block_idx = self._open_block(item_id, {"type": "text", "text": ""}) + self._chunk_queue.append( + { + "type": "content_block_delta", + "index": block_idx, + "delta": {"type": "text_delta", "text": delta}, + } + ) + return + # ---- text delta ---- if event_type == "response.output_text.delta": item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None) @@ -215,34 +240,47 @@ class AnthropicResponsesStreamWrapper: response_obj: Final = getattr(event, "response", None) or ( event.get("response") if isinstance(event, dict) else None ) - stop_reason = "end_turn" - anthropic_usage: AnthropicUsage = AnthropicUsage(input_tokens=0, output_tokens=0) - - if response_obj is not None: - status: Final = getattr(response_obj, "status", None) - if status == "incomplete": - stop_reason = "max_tokens" - anthropic_usage = ( - LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage( - getattr(response_obj, "usage", None) - ) + output: Final = (getattr(response_obj, "output", None) or ()) if response_obj is not None else () + refusal_text: Final = responses_output_refusal_text(output) or (self._refusal_text or None) + status: Final = getattr(response_obj, "status", None) if response_obj is not None else None + has_tool_call: Final = any( + getattr(item, "type", None) == "function_call" + or (isinstance(item, dict) and item.get("type") == "function_call") + for item in output + ) + stop_reason: Final = ( + "max_tokens" + if status == "incomplete" + else "refusal" + if refusal_text is not None + else "tool_use" + if has_tool_call + else "end_turn" + ) + anthropic_usage: Final[AnthropicUsage] = ( + LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage( + getattr(response_obj, "usage", None) ) + if response_obj is not None + else AnthropicUsage(input_tokens=0, output_tokens=0) + ) - # Check if tool_use was in the output to override stop_reason - if response_obj is not None: - output: Final = getattr(response_obj, "output", []) or [] - for out_item in output: - out_type = getattr(out_item, "type", None) or ( - out_item.get("type") if isinstance(out_item, dict) else None - ) - if out_type == "function_call": - stop_reason = "tool_use" - break + message_delta_payload: Final = { # mutable-ok: fresh message_delta payload built per chunk + "stop_reason": stop_reason, + "stop_sequence": None, + **( + { # mutable-ok: fresh message_delta stop_details entry built per chunk + "stop_details": refusal_stop_details(refusal_text) + } + if stop_reason == "refusal" + else {} # mutable-ok: empty spread placeholder for non-refusal stop + ), + } self._chunk_queue.append( { "type": "message_delta", - "delta": {"stop_reason": stop_reason, "stop_sequence": None}, + "delta": message_delta_payload, "usage": dict(anthropic_usage), } ) @@ -266,10 +304,20 @@ class AnthropicResponsesStreamWrapper: # Consume the upstream stream try: - async for event in self.responses_stream: - self._process_event(event) - if self._chunk_queue: - return self._chunk_queue.popleft() + if hasattr(self.responses_stream, "__aiter__"): + async for event in self.responses_stream: + self._process_event(event) + if self._chunk_queue: + return self._chunk_queue.popleft() + else: + if self._sync_responses_iterator is None: + self._sync_responses_iterator = iter(self.responses_stream) + sync_iterator: Final = self._sync_responses_iterator + missing: Final = object() + while (event := await asyncio.to_thread(next, sync_iterator, missing)) is not missing: + self._process_event(event) + if self._chunk_queue: + return self._chunk_queue.popleft() except StopAsyncIteration: pass except Exception as e: diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py index 62fc06a7c30..f1daf2be42a 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py @@ -19,6 +19,10 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import ( from litellm.litellm_core_utils.reasoning_effort_utils import ( reasoning_effort_from_thinking_budget, ) +from litellm.llms.anthropic.experimental_pass_through.messages.utils import ( + refusal_stop_details, + responses_output_refusal_text, +) from litellm.llms.anthropic.experimental_pass_through.utils import ( is_reasoning_auto_summary_enabled, prompt_cache_key_from_user_id, @@ -624,6 +628,9 @@ class LiteLLMAnthropicToResponsesAPIAdapter: content: Final[list[dict[str, object]]] = [] stop_reason: AnthropicFinishReason = "end_turn" + refusal_text: Final = responses_output_refusal_text( + cast(Iterable[object], response.output) # cast-ok: output items re-validated per item + ) for item in response.output: if isinstance(item, ResponseReasoningItem): @@ -631,10 +638,17 @@ class LiteLLMAnthropicToResponsesAPIAdapter: elif isinstance(item, ResponseOutputMessage): for part in item.content: - if getattr(part, "type", None) == "output_text": + part_type = getattr(part, "type", None) + if part_type == "output_text": content.append( AnthropicResponseContentBlockText(type="text", text=getattr(part, "text", "")).model_dump() ) + elif part_type == "refusal": + content.append( + AnthropicResponseContentBlockText( + type="text", text=getattr(part, "refusal", "") or "" + ).model_dump() + ) elif isinstance(item, ResponseFunctionToolCall): try: @@ -654,11 +668,21 @@ class LiteLLMAnthropicToResponsesAPIAdapter: elif isinstance(item, dict): item_type = item.get("type") if item_type == "message": - for part in item.get("content", []): - if isinstance(part, dict) and part.get("type") == "output_text": - content.append( - AnthropicResponseContentBlockText(type="text", text=part.get("text", "")).model_dump() - ) + for part in item.get("content", ()): + if isinstance(part, dict): + part_type = part.get("type") + if part_type == "output_text": + content.append( + AnthropicResponseContentBlockText( + type="text", text=part.get("text", "") + ).model_dump() + ) + elif part_type == "refusal": + content.append( + AnthropicResponseContentBlockText( + type="text", text=part.get("refusal", "") or "" + ).model_dump() + ) elif item_type == "reasoning": content.extend( self._thinking_blocks_from_reasoning_item( @@ -679,10 +703,10 @@ class LiteLLMAnthropicToResponsesAPIAdapter: ).model_dump(exclude_none=True) ) stop_reason = "tool_use" - - # status -> stop_reason override if response.status == "incomplete": stop_reason = "max_tokens" + elif refusal_text is not None: + stop_reason = "refusal" anthropic_usage: Final = self.translate_responses_api_usage_to_anthropic_usage(response.usage) @@ -695,4 +719,5 @@ class LiteLLMAnthropicToResponsesAPIAdapter: usage=anthropic_usage, content=content, stop_reason=stop_reason, + stop_details=(refusal_stop_details(refusal_text) if stop_reason == "refusal" else None), ) diff --git a/litellm/llms/azure_ai/chat/transformation.py b/litellm/llms/azure_ai/chat/transformation.py index f2d405e9a17..039c462b38a 100644 --- a/litellm/llms/azure_ai/chat/transformation.py +++ b/litellm/llms/azure_ai/chat/transformation.py @@ -17,6 +17,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import ( from litellm.llms.azure.common_utils import BaseAzureLLM from litellm.llms.azure_ai.common_utils import is_foundry_model_inference_base from litellm.llms.base_llm.chat.transformation import LiteLLMLoggingObj +from litellm.llms.openai.chat.gpt_5_transformation import OpenAIGPT5Config from litellm.llms.openai.common_utils import drop_params_from_unprocessable_entity_error from litellm.llms.openai.openai import OpenAIConfig from litellm.llms.xai.chat.transformation import XAIChatConfig @@ -42,12 +43,37 @@ NON_OPENAI_SPEC_MESSAGE_FIELDS: Final = ( ) +class AzureAIGPT5Config(OpenAIGPT5Config): + @classmethod + def _model_map_lookup_name(cls, model: str) -> str: + """Normalise a Foundry routing name to its cost-map key, when the map has one. + + A Foundry deployment and its OpenAI-hosted namesake are different products with + different capabilities, so ``azure_ai/`` is the entry to read whenever the map + carries it. Most gpt-5-family names have no ``azure_ai/`` row, though, and prefixing + those anyway costs them every flag: ``get_llm_provider`` re-resolves an ``azure_ai/`` + name to the azure provider when a global AZURE_AI_API_BASE points at an + openai.azure.com host, ``azure/`` is not a key either, so the lookup lands + nowhere and every effort answer degrades to False. A missing key defers to the base + resolver instead. + """ + prefixed: Final = model if model.startswith("azure_ai/") else f"azure_ai/{model}" + return prefixed if prefixed in litellm.model_cost else super()._model_map_lookup_name(model) + + +azureAIGPT5Config: Final = AzureAIGPT5Config() + + class AzureAIStudioConfig(OpenAIConfig): def get_supported_openai_params(self, model: str) -> list: model_supports_tool_choice = True # azure ai supports this by default if not supports_tool_choice(model=f"azure_ai/{model}"): model_supports_tool_choice = False - supported_params = super().get_supported_openai_params(model) + supported_params = ( + azureAIGPT5Config.get_supported_openai_params(model) + if azureAIGPT5Config.is_model_gpt_5_model(model) + else super().get_supported_openai_params(model) + ) if not model_supports_tool_choice: filtered_supported_params: Final = [] for param in supported_params: @@ -61,6 +87,27 @@ class AzureAIStudioConfig(OpenAIConfig): return supported_params + def map_openai_params( + self, + non_default_params: dict[str, object], # mutable-ok: OpenAIConfig.map_openai_params signature + optional_params: dict[str, object], # mutable-ok: OpenAIConfig.map_openai_params signature + model: str, + drop_params: bool, + ) -> dict[str, object]: # mutable-ok: OpenAIConfig.map_openai_params signature + if not azureAIGPT5Config.is_model_gpt_5_model(model): + return super().map_openai_params( + non_default_params=non_default_params, + optional_params=optional_params, + model=model, + drop_params=drop_params, + ) + return azureAIGPT5Config.map_openai_params( + non_default_params=non_default_params, + optional_params=optional_params, + model=model, + drop_params=drop_params, + ) + def _supports_stop_reason(self, model: str) -> bool: """ Check if the model supports stop tokens. diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 2c55828c1e2..6c6e65927f9 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -3485,6 +3485,55 @@ "supports_response_schema": true, "supports_tool_choice": true }, + "azure_ai/gpt-6-astra": { + "cache_creation_input_token_cost": 1.25e-05, + "cache_creation_input_token_cost_above_272k_tokens": 2.5e-05, + "cache_read_input_token_cost": 1e-06, + "cache_read_input_token_cost_above_272k_tokens": 2e-06, + "input_cost_per_token": 1e-05, + "input_cost_per_token_above_272k_tokens": 2e-05, + "litellm_provider": "azure_ai", + "max_input_tokens": 922000, + "max_output_tokens": 128000, + "max_tokens": 128000, + "mode": "chat", + "output_cost_per_token": 5e-05, + "output_cost_per_token_above_272k_tokens": 7.5e-05, + "search_context_cost_per_query": { + "search_context_size_high": 0.01, + "search_context_size_low": 0.01, + "search_context_size_medium": 0.01 + }, + "source": "https://ai.azure.com/catalog/models/gpt-6-astra", + "supported_endpoints": [ + "/v1/chat/completions", + "/v1/responses" + ], + "supported_modalities": [ + "text", + "image" + ], + "supported_output_modalities": [ + "text" + ], + "supports_computer_use": true, + "supports_function_calling": true, + "supports_max_reasoning_effort": false, + "supports_minimal_reasoning_effort": false, + "supports_native_streaming": true, + "supports_none_reasoning_effort": true, + "supports_parallel_function_calling": true, + "supports_pdf_input": true, + "supports_prompt_cache_breakpoint": true, + "supports_prompt_caching": true, + "supports_reasoning": true, + "supports_response_schema": true, + "supports_system_messages": true, + "supports_tool_choice": true, + "supports_vision": true, + "supports_web_search": true, + "supports_xhigh_reasoning_effort": true + }, "azure_ai/gpt-5.5": { "deprecation_date": "2027-10-26", "cache_read_input_token_cost": 5e-07, @@ -7189,7 +7238,7 @@ ], "supports_computer_use": true, "supports_function_calling": true, - "supports_max_reasoning_effort": true, + "supports_max_reasoning_effort": false, "supports_minimal_reasoning_effort": false, "supports_native_streaming": true, "supports_none_reasoning_effort": true, @@ -7455,7 +7504,7 @@ ], "supports_computer_use": true, "supports_function_calling": true, - "supports_max_reasoning_effort": true, + "supports_max_reasoning_effort": false, "supports_minimal_reasoning_effort": false, "supports_native_streaming": true, "supports_none_reasoning_effort": true, diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index e6880d521f1..9230be8055e 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -14,6 +14,7 @@ import time import traceback from collections.abc import Mapping, Sequence from datetime import datetime, timedelta, timezone +from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, cast, overload import litellm @@ -84,6 +85,25 @@ else: RESPONSES_SESSION_CALL_TYPES: Final = frozenset({CallTypes.responses.value, CallTypes.aresponses.value}) +def _is_batch_cost_row(payload: SpendLogsPayload) -> bool: + return payload.get("call_type") == CallTypes.aretrieve_batch.value and payload.get("status") == "success" + + +_BATCH_COST_CLAIM_FIELDS: Final = frozenset({"request_id", "call_type", "spend", "startTime", "endTime", "status"}) + + +def _batch_cost_row_to_write(payload: SpendLogsPayload, disable_spend_logs: bool) -> Mapping[str, object]: + """Reduce a batch's cost row to what tells the retrieves apart when logging is off. + + A proxy run with spend logs disabled still needs one row per batch to charge it once, + so the row is written either way, but it carries no request of its own: no metadata, + no requester IP, no key, model, or token counts (LIT-7048). + """ + if disable_spend_logs is False: + return payload + return MappingProxyType({field: value for field, value in payload.items() if field in _BATCH_COST_CLAIM_FIELDS}) + + class _SpendBatch(Protocol): litellm_usertable: BatchTable litellm_verificationtoken: BatchTable @@ -215,7 +235,12 @@ class DBSpendUpdateWriter: start_time: datetime | None, end_time: datetime | None, response_cost: float | None, - ) -> None: + ) -> bool: + """Record the request's spend, answering whether its cost still needs charging. + + False only for a batch retrieve whose cost row another retrieve already wrote, + so the caller leaves the key, team, and user counters alone (LIT-7048). + """ from litellm.proxy.proxy_server import ( disable_spend_logs, litellm_proxy_budget_name, @@ -232,7 +257,7 @@ class DBSpendUpdateWriter: team_id, ) if ProxyUpdateSpend.disable_spend_updates() is True: - return + return True if token is not None and isinstance(token, str) and token.startswith("sk-"): hashed_token = hash_token(token=token) else: @@ -262,11 +287,12 @@ class DBSpendUpdateWriter: if team_id is not None and team_id != "": payload["team_id"] = team_id + if not await self._record_spend_log( + payload=payload, prisma_client=prisma_client, disable_spend_logs=disable_spend_logs + ): + return False + if disable_spend_logs is False: - await self._insert_spend_log_to_db( - payload=payload, - prisma_client=prisma_client, - ) await self._enqueue_tool_usage_transaction( payload=payload, completion_response=completion_response, @@ -306,6 +332,7 @@ class DBSpendUpdateWriter: ) verbose_proxy_logger.debug("Runs spend update on all tables") + return True except Exception: spend_log_error( "Spend tracking - update_database failed. Spend log insertion or daily transaction enqueue " @@ -318,7 +345,102 @@ class DBSpendUpdateWriter: org_id, end_user_id, ) - return + return True + + async def _record_spend_log( + self, payload: SpendLogsPayload, prisma_client: "PrismaClient | None", disable_spend_logs: bool + ) -> bool: + if prisma_client is not None and _is_batch_cost_row(payload): + return await self._claim_batch_cost_spend_log( + payload=payload, prisma_client=prisma_client, disable_spend_logs=disable_spend_logs + ) + if disable_spend_logs is False: + await self._insert_spend_log_to_db(payload=payload, prisma_client=prisma_client) + return True + + async def _claim_batch_cost_spend_log( + self, payload: SpendLogsPayload, prisma_client: "PrismaClient", disable_spend_logs: bool + ) -> bool: + """Write the batch's cost row now, or learn that another retrieve already did. + + Every retrieve of one batch shares this row, so the insert that lands first owns + the charge and every later one finds the row and charges nothing (LIT-7048). Only + a row that recorded a charge counts: a failed retrieve, a request whose client + picked the batch id as its call id, and the $0 row an older proxy left behind + while the batch was still running all leave the charge to be made. + """ + from litellm.repositories.table_repositories import SpendLogsRepository + + request_id: Final = payload["request_id"] + row: Final = _batch_cost_row_to_write(payload, disable_spend_logs) + spend_logs: Final = SpendLogsRepository(prisma_client).table + try: + claimed: Final = await spend_logs.create_many( + data=[prisma_client.jsonify_object(row)], # mutable-ok: prisma create_many takes a list + skip_duplicates=True, + ) + if claimed == 1: + return True + existing: Final = await spend_logs.find_unique( + where={"request_id": request_id} # mutable-ok: prisma where clause + ) + except Exception as e: # noqa: BLE001 # prisma raises its own hierarchy; an unreachable DB queues the row like any other spend log + verbose_proxy_logger.warning( + "Could not claim spend row %s for a batch's cost, queueing it: %s", request_id, e + ) + await self._insert_spend_log_to_db(payload=prisma_client.jsonify_object(row), prisma_client=prisma_client) + return True + if existing is None or existing.call_type != CallTypes.aretrieve_batch.value or existing.status != "success": + verbose_proxy_logger.warning( + "Spend row %s belongs to a %s request, so this batch's cost is charged without a row of its own", + request_id, + getattr(existing, "call_type", None), + ) + return True + if existing.spend > 0: + verbose_proxy_logger.debug("Cost tracking skipped: spend row %s already charged this batch", request_id) + return False + return await self._take_over_uncharged_batch_cost_row(payload=payload, prisma_client=prisma_client, row=row) + + async def _take_over_uncharged_batch_cost_row( + self, payload: SpendLogsPayload, prisma_client: "PrismaClient", row: Mapping[str, object] + ) -> bool: + """Take the batch's cost row over from the poll that left it charging nothing. + + A pre-upgrade proxy wrote that row every time it polled the batch while it was still + running, so the charge is still to be made and the row still has to end up carrying + it. The row stops matching the moment it carries a charge, so it is one retrieve that + takes it over and charges, and every later one reads the charge and charges nothing. + """ + from litellm.repositories.table_repositories import SpendLogsRepository + + request_id: Final = payload["request_id"] + if payload["spend"] <= 0: + verbose_proxy_logger.debug( + "Cost tracking skipped: this batch costs nothing and spend row %s says so", request_id + ) + return False + try: + taken_over: Final = await SpendLogsRepository(prisma_client).table.update_many( + data=prisma_client.jsonify_object( + MappingProxyType({field: value for field, value in row.items() if field != "request_id"}) + ), + where={ # mutable-ok: prisma where clause + "request_id": request_id, + "call_type": CallTypes.aretrieve_batch.value, + "status": "success", + "spend": 0.0, + }, + ) + except Exception as e: # noqa: BLE001 # prisma raises its own hierarchy; the next retrieve takes the row over + verbose_proxy_logger.warning( + "Could not take over spend row %s, leaving this batch's cost to the next retrieve: %s", request_id, e + ) + return False + if taken_over == 0: + verbose_proxy_logger.debug("Cost tracking skipped: spend row %s already charged this batch", request_id) + return False + return True async def _enqueue_tool_usage_transaction( self, diff --git a/litellm/proxy/hooks/proxy_track_cost_callback.py b/litellm/proxy/hooks/proxy_track_cost_callback.py index d6276c03155..c4fba8ecf9e 100644 --- a/litellm/proxy/hooks/proxy_track_cost_callback.py +++ b/litellm/proxy/hooks/proxy_track_cost_callback.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Any, Final, cast import litellm from litellm._logging import verbose_proxy_logger +from litellm.batches.batch_utils import batch_cost_is_final from litellm.constants import BACKGROUND_INTERACTION_COST_POLLING_ENABLED from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.core_helpers import ( @@ -37,6 +38,7 @@ from litellm.proxy.spend_tracking.spend_tracking_utils import ( from litellm.proxy.utils import ProxyUpdateSpend from litellm.types.utils import ( CallTypes, + LiteLLMBatch, StandardLoggingPayload, StandardLoggingPayloadErrorInformation, ) @@ -248,6 +250,18 @@ class _ProxyDBLogger(CustomLogger): ) _write_spend_metadata_to_kwargs(kwargs=kwargs, metadata=metadata) budget_reservation: Final = _get_budget_reservation_from_metadata(metadata=metadata) + if ( + isinstance(completion_response, LiteLLMBatch) + and kwargs.get("call_type") == CallTypes.aretrieve_batch.value + and not batch_cost_is_final(completion_response) + ): + verbose_proxy_logger.debug( + "Cost tracking deferred for batch %s still in status %s", + completion_response.id, + completion_response.status, + ) + await _release_budget_reservation(budget_reservation=budget_reservation) + return user_id: Final = cast(str | None, metadata.get("user_api_key_user_id", None)) team_id: Final = cast(str | None, metadata.get("user_api_key_team_id", None)) org_id: Final = cast(str | None, metadata.get("user_api_key_org_id", None)) @@ -285,7 +299,7 @@ class _ProxyDBLogger(CustomLogger): call_type=call_type, ): ## UPDATE DATABASE - await _update_database_and_spend_counters( + charged: Final = await _update_database_and_spend_counters( proxy_logging_obj=proxy_logging_obj, increment_spend_counters=increment_spend_counters, user_api_key=user_api_key, @@ -302,6 +316,8 @@ class _ProxyDBLogger(CustomLogger): request_tags=tags, model_access_groups=model_access_groups, ) + if not charged: + return # update cache (fire-and-forget for backward compat: # cached object fields, soft budget alerts, etc.) @@ -578,9 +594,9 @@ async def _update_database_and_spend_counters( budget_reservation: dict | None, request_tags: list[str] | None = None, model_access_groups: Sequence[str] | None = None, -) -> None: +) -> bool: try: - await proxy_logging_obj.db_spend_update_writer.update_database( + charged: Final = await proxy_logging_obj.db_spend_update_writer.update_database( token=user_api_key, response_cost=response_cost, user_id=user_id, @@ -605,6 +621,9 @@ async def _update_database_and_spend_counters( "Failed to invalidate budget reservation counters after release failed" ) raise + if not charged: + await _release_budget_reservation(budget_reservation=budget_reservation) + return False try: await increment_spend_counters( @@ -630,6 +649,7 @@ async def _update_database_and_spend_counters( finally: budget_reservation["finalized"] = True raise + return True async def _release_budget_reservation(budget_reservation: dict | None) -> None: diff --git a/litellm/proxy/openai_files_endpoints/common_utils.py b/litellm/proxy/openai_files_endpoints/common_utils.py index 15eeddbc489..b1f282a0978 100644 --- a/litellm/proxy/openai_files_endpoints/common_utils.py +++ b/litellm/proxy/openai_files_endpoints/common_utils.py @@ -15,6 +15,7 @@ from typing import ( runtime_checkable, ) +from litellm.batches.batch_utils import batch_cost_is_final from litellm.proxy._types import ProxyException from litellm.repositories.table_repositories import ( ManagedFileRepository, @@ -1357,12 +1358,7 @@ def _completed_batch_safe_to_retire(response: "LiteLLMBatch") -> bool: enumerated the batch and none succeeded. A zero or unknown total means counts are unreported, so stay eligible and let the next poller pass revisit it. (#37713) """ - if response.output_file_id is not None: - return True - request_counts = response.request_counts - if request_counts is None: - return False - return request_counts.total > 0 and request_counts.completed == 0 + return batch_cost_is_final(response) async def update_batch_in_database( diff --git a/litellm/router_utils/reasoning_effort_capability.py b/litellm/router_utils/reasoning_effort_capability.py index 9185d901a28..7b145c15a07 100644 --- a/litellm/router_utils/reasoning_effort_capability.py +++ b/litellm/router_utils/reasoning_effort_capability.py @@ -10,8 +10,8 @@ opt-in. none is opt-out everywhere except the azure gpt-5 family, whose config r UnsupportedParamsError without an explicit true. xhigh is gated on the request path by the openai and azure gpt-5 configs. max is not gated there at -all: every entry carrying supports_max_reasoning_effort is Claude-family, and -anthropic/chat/transformation.py gates max on the output_config path while its reasoning_effort +all: outside the gpt-6-astra rows every entry carrying supports_max_reasoning_effort is Claude-family, +and anthropic/chat/transformation.py gates max on the output_config path while its reasoning_effort path maps any level to a thinking budget. Making max opt-in is a deliberate trade, then, since an explicit flag is the only signal that the tier is a real one rather than litellm rounding the level to a budget, and a missing flag costs advisory metadata rather than a rejected request. diff --git a/litellm/types/llms/anthropic.py b/litellm/types/llms/anthropic.py index b3462203c4b..365d59a179b 100644 --- a/litellm/types/llms/anthropic.py +++ b/litellm/types/llms/anthropic.py @@ -520,8 +520,15 @@ ContentBlockContentBlockDict = ToolUseBlock | TextBlock | ChatCompletionThinking ContentBlockStart = ContentBlockStartToolUse | ContentBlockStartText +class AnthropicStopDetails(TypedDict, total=False): + type: ReadOnly[Literal["refusal"]] + category: ReadOnly[str | None] + explanation: ReadOnly[str | None] + + class MessageDelta(TypedDict, total=False): stop_reason: str | None + stop_details: ReadOnly[AnthropicStopDetails] class ServerToolUsage(TypedDict, total=False): @@ -658,7 +665,7 @@ class AnthropicOutputTokensDetails(BaseModel): thinking_tokens: int | None = None -AnthropicFinishReason = Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"] +AnthropicFinishReason = Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal"] class AnthropicResponse(BaseModel): diff --git a/litellm/types/llms/anthropic_messages/anthropic_response.py b/litellm/types/llms/anthropic_messages/anthropic_response.py index 4fe1dafc73b..038a23a3ca2 100644 --- a/litellm/types/llms/anthropic_messages/anthropic_response.py +++ b/litellm/types/llms/anthropic_messages/anthropic_response.py @@ -5,6 +5,7 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict from litellm.types.llms.anthropic import ( AnthropicResponseContentBlockText, AnthropicResponseContentBlockToolUse, + AnthropicStopDetails, ContextManagementResponse, ServerToolUsage, ) @@ -78,16 +79,6 @@ class AnthropicUsage(TypedDict, total=False): server_tool_use: NotRequired[ReadOnly[ServerToolUsage]] -class AnthropicStopDetails(TypedDict, total=False): - """ - Safeguard verdict accompanying a `stop_reason: "refusal"` response: - https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback - """ - - category: ReadOnly[str | None] - explanation: ReadOnly[str | None] - - class AnthropicMessagesResponse(TypedDict, total=False): """ Anthropic Messages API Response: https://docs.anthropic.com/en/api/messages diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 2c55828c1e2..6c6e65927f9 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -3485,6 +3485,55 @@ "supports_response_schema": true, "supports_tool_choice": true }, + "azure_ai/gpt-6-astra": { + "cache_creation_input_token_cost": 1.25e-05, + "cache_creation_input_token_cost_above_272k_tokens": 2.5e-05, + "cache_read_input_token_cost": 1e-06, + "cache_read_input_token_cost_above_272k_tokens": 2e-06, + "input_cost_per_token": 1e-05, + "input_cost_per_token_above_272k_tokens": 2e-05, + "litellm_provider": "azure_ai", + "max_input_tokens": 922000, + "max_output_tokens": 128000, + "max_tokens": 128000, + "mode": "chat", + "output_cost_per_token": 5e-05, + "output_cost_per_token_above_272k_tokens": 7.5e-05, + "search_context_cost_per_query": { + "search_context_size_high": 0.01, + "search_context_size_low": 0.01, + "search_context_size_medium": 0.01 + }, + "source": "https://ai.azure.com/catalog/models/gpt-6-astra", + "supported_endpoints": [ + "/v1/chat/completions", + "/v1/responses" + ], + "supported_modalities": [ + "text", + "image" + ], + "supported_output_modalities": [ + "text" + ], + "supports_computer_use": true, + "supports_function_calling": true, + "supports_max_reasoning_effort": false, + "supports_minimal_reasoning_effort": false, + "supports_native_streaming": true, + "supports_none_reasoning_effort": true, + "supports_parallel_function_calling": true, + "supports_pdf_input": true, + "supports_prompt_cache_breakpoint": true, + "supports_prompt_caching": true, + "supports_reasoning": true, + "supports_response_schema": true, + "supports_system_messages": true, + "supports_tool_choice": true, + "supports_vision": true, + "supports_web_search": true, + "supports_xhigh_reasoning_effort": true + }, "azure_ai/gpt-5.5": { "deprecation_date": "2027-10-26", "cache_read_input_token_cost": 5e-07, @@ -7189,7 +7238,7 @@ ], "supports_computer_use": true, "supports_function_calling": true, - "supports_max_reasoning_effort": true, + "supports_max_reasoning_effort": false, "supports_minimal_reasoning_effort": false, "supports_native_streaming": true, "supports_none_reasoning_effort": true, @@ -7455,7 +7504,7 @@ ], "supports_computer_use": true, "supports_function_calling": true, - "supports_max_reasoning_effort": true, + "supports_max_reasoning_effort": false, "supports_minimal_reasoning_effort": false, "supports_native_streaming": true, "supports_none_reasoning_effort": true, diff --git a/tests/test_litellm/batches/test_batch_utils.py b/tests/test_litellm/batches/test_batch_utils.py index c86c7c4df03..976a96f2db1 100644 --- a/tests/test_litellm/batches/test_batch_utils.py +++ b/tests/test_litellm/batches/test_batch_utils.py @@ -21,11 +21,12 @@ from types import MappingProxyType import httpx import pytest import respx +from openai.types.batch import BatchRequestCounts import litellm import litellm.batches.batch_utils as bu -from litellm.types.utils import Usage +from litellm.types.utils import LiteLLMBatch, Usage # --------------------------------------------------------------------------- # # Builders for batch OUTPUT file rows. @@ -1718,3 +1719,57 @@ def test_unparsable_bedrock_batch_usage_warns(caplog): assert usage.total_tokens == 0 assert "does not understand" in caplog.text assert "inputTextTokenCount" in caplog.text + + +# --------------------------------------------------------------------------- # +# batch_cost_is_final +# --------------------------------------------------------------------------- # + +def _retrieved_batch( + status: str, output_file_id: str | None = None, counts: BatchRequestCounts | None = None +) -> LiteLLMBatch: + return LiteLLMBatch( + id="batch_abc", + completion_window="24h", + created_at=1, + endpoint="/v1/chat/completions", + input_file_id="file-in", + object="batch", + status="validating", + output_file_id=output_file_id, + request_counts=counts, + ).model_copy(update={"status": status}) + + +class TestBatchCostIsFinal: + """Every retrieve of one batch writes the same spend row, so the first retrieve + that prices it decides the row for good. A poll before the output exists must + therefore not count as final: pricing it recorded $0 and pinned it (LIT-7048).""" + + @pytest.mark.parametrize("status", ["validating", "in_progress", "finalizing", "cancelling"]) + def test_in_flight_batch_is_not_final(self, status): + assert bu.batch_cost_is_final(_retrieved_batch(status)) is False + + @pytest.mark.parametrize("status", ["completed", "complete"]) + def test_completed_with_output_is_final(self, status): + assert bu.batch_cost_is_final(_retrieved_batch(status, output_file_id="file-out")) is True + + def test_completed_without_output_and_unknown_counts_is_not_final(self): + assert bu.batch_cost_is_final(_retrieved_batch("completed")) is False + + def test_completed_without_output_and_zero_counts_is_not_final(self): + counts = BatchRequestCounts(total=0, completed=0, failed=0) + assert bu.batch_cost_is_final(_retrieved_batch("completed", counts=counts)) is False + + def test_completed_without_output_but_successful_lines_is_not_final(self): + counts = BatchRequestCounts(total=2, completed=2, failed=0) + assert bu.batch_cost_is_final(_retrieved_batch("completed", counts=counts)) is False + + @pytest.mark.parametrize("status", ["completed", "complete"]) + def test_completed_without_output_and_every_line_failed_is_final(self, status): + counts = BatchRequestCounts(total=2, completed=0, failed=2) + assert bu.batch_cost_is_final(_retrieved_batch(status, counts=counts)) is True + + @pytest.mark.parametrize("status", ["failed", "expired", "cancelled"]) + def test_other_terminal_statuses_are_final(self, status): + assert bu.batch_cost_is_final(_retrieved_batch(status)) is True diff --git a/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py b/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py index df680b7cb0e..40abb5bfca3 100644 --- a/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py +++ b/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py @@ -2008,7 +2008,14 @@ def test_generic_cost_per_token_azure_gpt56(_local_model_cost_map, assert round(completion_cost, 10) == round(output_cost * completion_tokens, 10) -@pytest.mark.parametrize("model,zone_multiplier", [("azure/gpt-6-astra", 1.0), ("azure/us/gpt-6-astra", 1.1)]) +@pytest.mark.parametrize( + "model,custom_llm_provider,zone_multiplier", + [ + ("azure/gpt-6-astra", "azure", 1.0), + ("azure/us/gpt-6-astra", "azure", 1.1), + ("azure_ai/gpt-6-astra", "azure_ai", 1.0), + ], +) @pytest.mark.parametrize( "prompt_tokens,input_side_multiplier,output_multiplier", [(100000, 1.0, 1.0), (300000, 2.0, 1.5)], @@ -2016,6 +2023,7 @@ def test_generic_cost_per_token_azure_gpt56(_local_model_cost_map, def test_generic_cost_per_token_azure_gpt_6_astra_foundry_price_sheet( _local_model_cost_map, model, + custom_llm_provider, zone_multiplier, prompt_tokens, input_side_multiplier, @@ -2023,7 +2031,8 @@ def test_generic_cost_per_token_azure_gpt_6_astra_foundry_price_sheet( ): """Microsoft Foundry sells gpt-6-astra at the OpenAI rates: $10 input, $1 cache read, $12.50 cache write, $50 output per 1M tokens on Standard Global, with the input side doubling and output 1.5x above 272K - prompt tokens. Standard US Data Zone carries the usual 10% uplift on every rate. + prompt tokens. Standard US Data Zone carries the usual 10% uplift on every rate. A Foundry + deployment reached through the azure_ai route bills the same Standard Global sheet. """ cached_tokens = 50000 cache_write_tokens = 40000 @@ -2041,7 +2050,7 @@ def test_generic_cost_per_token_azure_gpt_6_astra_foundry_price_sheet( prompt_cost, completion_cost = generic_cost_per_token( model=model, usage=usage, - custom_llm_provider="azure", + custom_llm_provider=custom_llm_provider, ) input_side = zone_multiplier * input_side_multiplier @@ -2051,6 +2060,18 @@ def test_generic_cost_per_token_azure_gpt_6_astra_foundry_price_sheet( assert completion_cost == pytest.approx(zone_multiplier * output_multiplier * completion_tokens * 5e-5) +def test_generic_cost_per_token_azure_ai_gpt_6_astra_flex_bills_the_standard_rate(_local_model_cost_map): + usage = Usage(prompt_tokens=1000, completion_tokens=100, total_tokens=1100) + + standard = generic_cost_per_token(model="azure_ai/gpt-6-astra", usage=usage, custom_llm_provider="azure_ai") + flex = generic_cost_per_token( + model="azure_ai/gpt-6-astra", usage=usage, custom_llm_provider="azure_ai", service_tier="flex" + ) + + assert flex == standard + assert standard == pytest.approx((1000 * 1e-05, 100 * 5e-05)) + + @pytest.mark.parametrize( "model,expected_none,expected_xhigh,expected_minimal", [ diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index 0f44cbbb2a7..0568f258dba 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -632,6 +632,86 @@ class TestRetrieveBatchCostPassesModelIdentity: assert captured["model_info"]["input_cost_per_token"] == 0.0 +class TestRetrieveBatchPricesOnlyFinalBatches: + """Regression (LIT-7048): retrieving a provider-id batch priced it on every poll. + + Every retrieve of one batch logs under the same spend row, so pricing a poll + that landed before the output existed wrote that row at $0 and pinned it there. + Only a final batch gets priced; an in-flight poll carries no cost at all. + """ + + @staticmethod + def _logging_obj() -> LitellmLogging: + obj = LitellmLogging( + model="gpt-5.6-luna", + messages=[{"role": "user", "content": "Hey"}], + stream=False, + call_type="aretrieve_batch", + start_time=time.time(), + litellm_call_id="batch-call-2", + function_id="f", + ) + obj.custom_llm_provider = "openai" + return obj + + @staticmethod + def _batch(status: str, output_file_id: str | None): + from litellm.types.utils import LiteLLMBatch + + return LiteLLMBatch( + id="batch_6a9c99e185588190877d391f8b9d7f8a", + completion_window="24h", + created_at=1, + endpoint="/v1/chat/completions", + input_file_id="file-in", + object="batch", + status="validating", + output_file_id=output_file_id, + ).model_copy(update={"status": status}) + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("status", "output_file_id"), + [("validating", None), ("in_progress", None), ("finalizing", None), ("completed", None), ("complete", None)], + ) + async def test_non_final_batch_is_not_priced(self, monkeypatch, status, output_file_id) -> None: + from litellm.litellm_core_utils import litellm_logging as logging_module + + handle_completed_batch = AsyncMock() + monkeypatch.setattr(logging_module, "_handle_completed_batch", handle_completed_batch) + batch = self._batch(status, output_file_id) + + await self._logging_obj()._async_success_handler_body(result=batch, start_time=None, end_time=None) + + handle_completed_batch.assert_not_awaited() + assert "response_cost" not in batch._hidden_params + + @pytest.mark.asyncio + async def test_completed_batch_with_output_is_priced(self, monkeypatch) -> None: + from litellm.batches.batch_utils import BatchCostUsageResult + from litellm.litellm_core_utils import litellm_logging as logging_module + from litellm.types.utils import Usage + + handle_completed_batch = AsyncMock( + return_value=BatchCostUsageResult( + cost=8e-06, + usage=Usage(prompt_tokens=26, completion_tokens=9, total_tokens=35), + models=["gpt-5.6-luna"], + successful_requests=2, + failed_requests=0, + ) + ) + monkeypatch.setattr(logging_module, "_handle_completed_batch", handle_completed_batch) + batch = self._batch("completed", "file-out") + + await self._logging_obj()._async_success_handler_body(result=batch, start_time=None, end_time=None) + + handle_completed_batch.assert_awaited_once() + assert batch._hidden_params["response_cost"] == 8e-06 + assert batch.usage is not None + assert batch.usage.total_tokens == 35 + + class TestAnthropicPassthroughCustomPricing: """Verify the Anthropic pass-through handler forwards custom pricing.""" diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py index 0478a251057..30465ca25ba 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py @@ -40,6 +40,51 @@ from litellm.types.utils import ( ) +def test_translate_chat_refusal_to_anthropic_response(): + response = ModelResponse( + id="chatcmpl-refusal", + model="openai-model", + choices=[ + Choices( + index=0, + finish_reason="stop", + message=Message(content=None, role="assistant", refusal="I cannot fulfill this request."), + ) + ], + usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2), + ) + + result = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response) + + assert result["content"] == [{"type": "text", "text": "I cannot fulfill this request."}] + assert result["stop_reason"] == "refusal" + assert result.get("stop_details") == { + "type": "refusal", + "category": None, + "explanation": "I cannot fulfill this request.", + } + + +def test_translate_chat_length_takes_precedence_over_refusal(): + response = ModelResponse( + id="chatcmpl-partial-refusal", + model="openai-model", + choices=[ + Choices( + index=0, + finish_reason="length", + message=Message(content=None, role="assistant", refusal="Partial refusal"), + ) + ], + usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2), + ) + + result = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response) + + assert result["stop_reason"] == "max_tokens" + assert result.get("stop_details") is None + + def test_translate_streaming_openai_chunk_to_anthropic_content_block(): choices = [ StreamingChoices( diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_handler_reasoning_effort_normalization.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_handler_reasoning_effort_normalization.py index 56b754c3476..af7befecc33 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_handler_reasoning_effort_normalization.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_handler_reasoning_effort_normalization.py @@ -82,3 +82,22 @@ class TestTheNormalizedTierIsTheTierSent: self, local_model_cost_map, model, provider, effort, expected ): assert _reasoning_effort_sent(model, provider, effort) == expected + + @pytest.mark.parametrize( + "model, provider", + [ + ("gpt-6-astra", "azure_ai"), + ("azure_ai/gpt-6-astra", "azure_ai"), + ("gpt-6-astra", "azure"), + ("us/gpt-6-astra", "azure"), + ], + ) + def test_an_azure_hosted_astra_deployment_drops_to_the_tier_it_accepts( + self, local_model_cost_map, model, provider + ): + """The deployment answers ``max`` with a 400 naming ``none`` through ``xhigh``, so the rows + say so and the adapter sends the tier below instead of the rejected one.""" + assert _reasoning_effort_sent(model, provider, "max") == "xhigh" + + def test_the_openai_hosted_twin_still_sends_max(self, local_model_cost_map): + assert _reasoning_effort_sent("gpt-6-astra", "openai", "max") == "max" diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py index 17d42f55ae0..fdd08eaa182 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py @@ -108,6 +108,136 @@ def _text_deltas(events: List[dict]) -> List[str]: ] +def test_streaming_chat_refusal_emits_refusal_text_and_stop_details(): + chunks = [ + _make_chunk(Delta(content=None, refusal="I cannot fulfill this request.")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="openai-model") + + events = _drain_sync(wrapper) + + assert _text_deltas(events) == ["I cannot fulfill this request."] + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"] == { + "stop_reason": "refusal", + "stop_details": { + "type": "refusal", + "category": None, + "explanation": "I cannot fulfill this request.", + }, + } + + +@pytest.mark.asyncio +async def test_streaming_chat_refusal_emits_refusal_text_and_stop_details_async(): + chunks = [ + _make_chunk(Delta(content=None, refusal="I cannot fulfill this request.")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="openai-model") + + events = await _drain_async(wrapper) + + assert _text_deltas(events) == ["I cannot fulfill this request."] + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "refusal" + assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request." + + +def test_streaming_chat_refusal_parked_in_provider_specific_fields_is_emitted(): + """Providers that do not populate ``delta.refusal`` (Azure o-series among + them) hand LiteLLM the refusal as an unrecognized field, which lands in + ``provider_specific_fields``. That first delta still has to stream as text, + otherwise the client gets ``stop_reason: refusal`` over an empty content + array and shows the user nothing. + """ + chunks = [ + _make_chunk(Delta(content=None, provider_specific_fields={"refusal": "I cannot fulfill this request."})), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="openai-model") + + events = _drain_sync(wrapper) + + assert _text_deltas(events) == ["I cannot fulfill this request."] + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "refusal" + assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request." + + +@pytest.mark.asyncio +async def test_streaming_chat_refusal_parked_in_provider_specific_fields_is_emitted_async(): + chunks = [ + _make_chunk(Delta(content=None, provider_specific_fields={"refusal": "I cannot fulfill this request."})), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="openai-model") + + events = await _drain_async(wrapper) + + assert _text_deltas(events) == ["I cannot fulfill this request."] + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "refusal" + assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request." + + +def test_streaming_chat_combined_refusal_and_finish_reason_is_preserved(): + """Fake-streamed responses arrive as one chunk carrying both the delta and the + finish_reason. The refusal has to be split off and streamed as text, or the + client gets ``stop_reason: refusal`` over an empty content array. + """ + chunks = [ + _make_chunk( + Delta(content=None, refusal="I cannot fulfill this request."), + finish_reason="stop", + ) + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="openai-model") + + events = _drain_sync(wrapper) + + assert _text_deltas(events) == ["I cannot fulfill this request."] + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "refusal" + assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request." + + +@pytest.mark.asyncio +async def test_streaming_chat_combined_refusal_and_finish_reason_is_preserved_async(): + chunks = [ + _make_chunk( + Delta(content=None, provider_specific_fields={"refusal": "I cannot fulfill this request."}), + finish_reason="stop", + ) + ] + wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="openai-model") + + events = await _drain_async(wrapper) + + assert _text_deltas(events) == ["I cannot fulfill this request."] + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "refusal" + assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request." + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_streaming_chat_length_takes_precedence_over_refusal(async_mode: bool): + chunks = [ + _make_chunk(Delta(content=None, refusal="Partial refusal")), + _make_chunk(Delta(content=None), finish_reason="length"), + ] + stream = _AsyncStream(chunks) if async_mode else iter(chunks) + wrapper = AnthropicStreamWrapper(completion_stream=stream, model="openai-model") + + events = await _drain_async(wrapper) if async_mode else _drain_sync(wrapper) + + message_delta = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "max_tokens" + assert "stop_details" not in message_delta["delta"] + + def _input_json_deltas(events: List[dict]) -> List[str]: return [ e["delta"]["partial_json"] diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index aebbed88c70..d9df5df426a 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py @@ -34,6 +34,14 @@ def _drain_async(events: list) -> list: return asyncio.run(_run()) +def _drain_sync_upstream(events: list) -> list: + async def _run() -> list: + wrapper = AnthropicResponsesStreamWrapper(responses_stream=iter(events), model="m") + return [chunk async for chunk in wrapper] + + return asyncio.run(_run()) + + class TestMessageStartEmittedExactlyOnce: """The ``__anext__`` fallback emits ``message_start`` before consuming the stream, so ``_process_event`` must not emit a second one when @@ -55,6 +63,15 @@ class TestMessageStartEmittedExactlyOnce: chunks = _drain_async([{"type": "response.created"}]) assert chunks[0]["type"] == "message_start" + def test_sync_upstream_iterator_is_consumed(self): + chunks = _drain_sync_upstream( + [ + {"type": "response.created"}, + {"type": "response.output_text.delta", "item_id": "m1", "delta": "hi"}, + ] + ) + assert any(chunk.get("delta", {}).get("text") == "hi" for chunk in chunks) + class TestProcessEventResponseCreatedGuard: """``_process_event`` must emit ``message_start`` exactly once even if @@ -308,3 +325,60 @@ class TestResponseCompletedUsage: "cache_creation_input_tokens": 10, "cache_read_input_tokens": 4004, } + + +class TestRefusalStreamEvents: + def test_refusal_event_sequence_emits_refusal_text_and_stop_details(self): + response = SimpleNamespace( + status="completed", + output=[{"type": "message", "content": [{"type": "refusal", "refusal": "I cannot fulfill this."}]}], + usage=None, + ) + chunks = _process_all( + [ + {"type": "response.created"}, + {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}}, + {"type": "response.refusal.delta", "item_id": "msg_1", "delta": "I cannot fulfill this."}, + {"type": "response.output_item.done", "item": {"type": "message", "id": "msg_1"}}, + {"type": "response.completed", "response": response}, + ] + ) + assert [chunk["type"] for chunk in chunks] == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + assert chunks[2]["delta"] == {"type": "text_delta", "text": "I cannot fulfill this."} + assert chunks[4]["delta"] == { + "stop_reason": "refusal", + "stop_sequence": None, + "stop_details": { + "type": "refusal", + "category": None, + "explanation": "I cannot fulfill this.", + }, + } + + def test_response_completed_with_refusal_sets_stop_reason_refusal(self): + response = SimpleNamespace( + status="completed", + output=[{"type": "message", "content": [{"type": "refusal", "refusal": "Policy violation"}]}], + usage=None, + ) + chunks = _process_all([{"type": "response.completed", "response": response}]) + message_delta = next(c for c in chunks if c["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "refusal" + + def test_incomplete_status_takes_precedence_over_refusal(self): + response = SimpleNamespace( + status="incomplete", + output=[{"type": "message", "content": [{"type": "refusal", "refusal": "Partial refusal"}]}], + usage=None, + ) + chunks = _process_all([{"type": "response.incomplete", "response": response}]) + message_delta = next(c for c in chunks if c["type"] == "message_delta") + assert message_delta["delta"]["stop_reason"] == "max_tokens" + assert "stop_details" not in message_delta["delta"] diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py index 0b2a9e655fd..9f8414afa38 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py @@ -147,9 +147,7 @@ class TestOutputConfigStructuredOutput: def test_output_config_format_explicit_strict_true_is_preserved(self): """Nested output_config.format with explicit strict=True is preserved.""" - req = _make_request( - output_config={"format": {"type": "json_schema", "schema": self._SCHEMA, "strict": True}} - ) + req = _make_request(output_config={"format": {"type": "json_schema", "schema": self._SCHEMA, "strict": True}}) kwargs = _ADAPTER.translate_request(req) assert kwargs["text"]["format"]["strict"] is True @@ -1207,6 +1205,18 @@ def _make_output_message(texts: List[str]) -> MagicMock: return msg +def _make_refusal_message(refusal_text: str): + from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal + + return ResponseOutputMessage( + id="msg_refusal", + content=[ResponseOutputRefusal(type="refusal", refusal=refusal_text)], + role="assistant", + status="completed", + type="message", + ) + + def _make_function_call_item(call_id: str, name: str, arguments: str) -> MagicMock: """Build a mock ResponseFunctionToolCall.""" from openai.types.responses import ResponseFunctionToolCall # type: ignore[import] @@ -1280,6 +1290,41 @@ class TestTranslateResponse: result: Any = _ADAPTER.translate_response(response) assert result["stop_reason"] == "end_turn" + def test_refusal_part_becomes_text_block_and_sets_stop_reason_refusal(self): + response = _make_mock_response(output=[_make_refusal_message("I cannot fulfill this request.")]) + result: Any = _ADAPTER.translate_response(response) + assert len(result["content"]) == 1 + assert result["content"][0]["type"] == "text" + assert result["content"][0]["text"] == "I cannot fulfill this request." + assert result["stop_reason"] == "refusal" + assert result.get("stop_details") == { + "type": "refusal", + "category": None, + "explanation": "I cannot fulfill this request.", + } + + def test_dict_refusal_part_in_message_becomes_text_block(self): + output_item = { + "type": "message", + "content": [{"type": "refusal", "refusal": "Refused by policy"}], + } + response = _make_mock_response(output=[output_item]) + result: Any = _ADAPTER.translate_response(response) + assert len(result["content"]) == 1 + assert result["content"][0]["type"] == "text" + assert result["content"][0]["text"] == "Refused by policy" + assert result["stop_reason"] == "refusal" + assert result.get("stop_details", {}).get("explanation") == "Refused by policy" + + def test_incomplete_status_takes_precedence_over_refusal(self): + response = _make_mock_response( + output=[_make_refusal_message("Partial refusal")], + status="incomplete", + ) + result: Any = _ADAPTER.translate_response(response) + assert result["stop_reason"] == "max_tokens" + assert result.get("stop_details") is None + def test_incomplete_status_sets_max_tokens(self): """status='incomplete' overrides stop_reason to 'max_tokens'.""" response = _make_mock_response( @@ -1338,9 +1383,7 @@ class TestTranslateResponse: ] ) result: Any = _ADAPTER.translate_response(response) - assert result["content"] == [ - {"type": "thinking", "thinking": "Weighing the options.", "signature": None} - ] + assert result["content"] == [{"type": "thinking", "thinking": "Weighing the options.", "signature": None}] def test_thinking_blocks_are_dropped_when_replayed_to_anthropic(self): """Replaying this turn to an Anthropic model must not send a signature it cannot verify.""" @@ -1483,9 +1526,7 @@ class TestToolResultImages: }, { "role": "user", - "content": [ - {"type": "tool_result", "tool_use_id": "toolu_01", "content": tool_result_content} - ], + "content": [{"type": "tool_result", "tool_use_id": "toolu_01", "content": tool_result_content}], }, ] @@ -1632,9 +1673,7 @@ class TestToolResultDocuments: }, { "role": "user", - "content": [ - {"type": "tool_result", "tool_use_id": "toolu_01", "content": tool_result_content} - ], + "content": [{"type": "tool_result", "tool_use_id": "toolu_01", "content": tool_result_content}], }, ] @@ -1667,9 +1706,7 @@ class TestToolResultDocuments: def test_document_title_becomes_filename(self): output = self._tool_output(self._translate([self._base64_document(title="quarterly-report.pdf")])) - assert output == [ - {"type": "input_file", "filename": "quarterly-report.pdf", "file_data": self.PDF_DATA_URI} - ] + assert output == [{"type": "input_file", "filename": "quarterly-report.pdf", "file_data": self.PDF_DATA_URI}] def test_url_document_becomes_file_url_part(self): output = self._tool_output( @@ -1778,9 +1815,7 @@ class TestUserContentDocuments: def test_document_title_becomes_filename(self): content = self._user_content(self._translate([self._base64_document(title="quarterly-report.pdf")])) - assert content == [ - {"type": "input_file", "filename": "quarterly-report.pdf", "file_data": self.PDF_DATA_URI} - ] + assert content == [{"type": "input_file", "filename": "quarterly-report.pdf", "file_data": self.PDF_DATA_URI}] def test_url_document_becomes_file_url_part(self): content = self._user_content( @@ -1810,9 +1845,7 @@ class TestUserContentDocuments: assert content == [{"type": "input_text", "text": "still here"}] def test_document_breakpoint_rides_on_the_file_part(self): - content = self._user_content( - self._translate([self._base64_document(prompt_cache_breakpoint=self.EXPLICIT)]) - ) + content = self._user_content(self._translate([self._base64_document(prompt_cache_breakpoint=self.EXPLICIT)])) assert content == [ { "type": "input_file", @@ -1859,7 +1892,9 @@ class TestPromptCacheBreakpointToResponses: ] def test_system_without_breakpoint_still_becomes_instructions(self): - request = _make_request(system=[{"type": "text", "text": "Be concise."}, {"type": "text", "text": "Be helpful."}]) + request = _make_request( + system=[{"type": "text", "text": "Be concise."}, {"type": "text", "text": "Be helpful."}] + ) kwargs = _ADAPTER.translate_request(request) assert kwargs["instructions"] == "Be concise.\nBe helpful." assert kwargs["input"] == [ diff --git a/tests/test_litellm/llms/azure_ai/chat/test_azure_ai_transformation.py b/tests/test_litellm/llms/azure_ai/chat/test_azure_ai_transformation.py index 33fbb4e8fc7..f8cc0b5071e 100644 --- a/tests/test_litellm/llms/azure_ai/chat/test_azure_ai_transformation.py +++ b/tests/test_litellm/llms/azure_ai/chat/test_azure_ai_transformation.py @@ -3,6 +3,8 @@ from unittest.mock import MagicMock, patch import pytest +import litellm +from litellm.litellm_core_utils.get_model_cost_map import get_model_cost_map from litellm.llms.azure_ai.azure_model_router.transformation import ( AzureModelRouterConfig, ) @@ -138,6 +140,46 @@ def test_azure_ai_validate_environment_with_azure_ad_token(): assert headers["Content-Type"] == "application/json" +@pytest.fixture +def _local_model_cost_map(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", get_model_cost_map(url=litellm.model_cost_map_url)) + + +def test_foundry_gpt_6_astra_keeps_sampling_params_when_reasoning_effort_is_none(_local_model_cost_map): + optional_params = AzureAIStudioConfig().map_openai_params( + non_default_params={"reasoning_effort": "none", "temperature": 0.2, "top_p": 0.9}, + optional_params={}, + model="gpt-6-astra", + drop_params=False, + ) + + assert optional_params == {"reasoning_effort": "none", "temperature": 0.2, "top_p": 0.9} + + +def test_a_gpt_5_name_without_a_foundry_row_keeps_reading_its_own_entry( + monkeypatch: pytest.MonkeyPatch, _local_model_cost_map +): + """Most gpt-5-family names have no azure_ai/ row. Reading an azure_ai/ key for those finds + nothing, and an openai.azure.com base sends the name down the azure provider, which has no key + for it either, so every effort answer would silently fall back to false and take temperature, + top_p and logprobs down with it.""" + monkeypatch.setenv("AZURE_AI_API_BASE", "https://example-resource.openai.azure.com") + monkeypatch.setenv("AZURE_AI_API_KEY", "placeholder") + + optional_params = litellm.utils.get_optional_params( + model="gpt-5.1-chat-latest", + custom_llm_provider="azure_ai", + temperature=0.2, + top_p=0.9, + logprobs=True, + ) + + assert optional_params["temperature"] == 0.2 + assert optional_params["top_p"] == 0.9 + assert optional_params["logprobs"] is True + + def test_azure_ai_grok_stop_parameter_handling(): """ Test that Grok models properly handle stop parameter filtering in Azure AI Studio. diff --git a/tests/test_litellm/proxy/auth/test_model_checks.py b/tests/test_litellm/proxy/auth/test_model_checks.py index d58683fd1e5..36bfc4c5dd3 100644 --- a/tests/test_litellm/proxy/auth/test_model_checks.py +++ b/tests/test_litellm/proxy/auth/test_model_checks.py @@ -857,6 +857,24 @@ def test_add_known_models_refreshes_models_by_provider_for_wildcard_expansion(): litellm.add_known_models(model_cost_map={}) assert fake_model not in litellm.models_by_provider["vertex_ai"] + +def test_azure_ai_wildcard_lists_the_foundry_gpt_6_astra_entry(monkeypatch): + import litellm + from litellm.proxy.auth.model_checks import get_known_models_from_wildcard + + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + foundry_key = "azure_ai/gpt-6-astra" + local_entry = litellm.get_model_cost_map(url="")[foundry_key] + registered_before = foundry_key in litellm.azure_ai_models + try: + litellm.add_known_models(model_cost_map={foundry_key: local_entry}) + assert foundry_key in get_known_models_from_wildcard("azure_ai/*") + finally: + if not registered_before: + litellm.azure_ai_models.discard(foundry_key) + litellm.add_known_models(model_cost_map={}) + + def test_get_complete_model_list_drops_no_default_models_sentinel(): from litellm.proxy.auth.model_checks import get_complete_model_list diff --git a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py index 11ef911de3e..0bca7c9492c 100644 --- a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py +++ b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py @@ -7,6 +7,7 @@ import re from collections.abc import Callable from contextlib import asynccontextmanager from datetime import datetime, timezone +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, call, patch import pytest @@ -2936,7 +2937,9 @@ async def test_commit_spend_updates_retries_deadlock_on_every_entity_path(monkey "call_type, expects_flush", [("aresponses", True), ("responses", True), ("acompletion", False)], ) -async def test_insert_spend_log_asks_for_an_immediate_flush_on_responses_calls(call_type: str, expects_flush: bool): +async def test_insert_spend_log_asks_for_an_immediate_flush_on_rows_other_workers_read_back( + call_type: str, expects_flush: bool +): """ A `previous_response_id` chained straight off the previous turn reads the DB, so a Responses row cannot sit in this worker's queue until the monitor's next poll. @@ -2957,6 +2960,303 @@ async def test_insert_spend_log_asks_for_an_immediate_flush_on_responses_calls(c PrismaClient.spend_log_flush_requested.clear() +def _batch_cost_payload() -> dict: + return { + **_minimal_spend_payload(), + "request_id": "batch_abc_batch_cost", + "call_type": "aretrieve_batch", + "status": "success", + } + + +def _spend_logs_prisma(inserted: int, existing: object, taken_over: int = 1) -> MagicMock: + prisma = _tool_usage_prisma() + prisma.jsonify_object = lambda data: dict(data) + prisma.db.litellm_spendlogs.create_many = AsyncMock(return_value=inserted) + prisma.db.litellm_spendlogs.find_unique = AsyncMock(return_value=existing) + prisma.db.litellm_spendlogs.update_many = AsyncMock(return_value=taken_over) + return prisma + + +async def _update_database_with( + db_writer: DBSpendUpdateWriter, + prisma: MagicMock, + payload: dict, + disable_spend_logs: bool = False, + response_cost: float = 0.25, +) -> bool: + with ( + patch( # test-quality-ok: update_database reads this proxy_server global at call time, no seam + "litellm.proxy.proxy_server.disable_spend_logs", disable_spend_logs + ), + patch( # test-quality-ok: update_database reads this proxy_server global at call time, no seam + "litellm.proxy.proxy_server.prisma_client", prisma + ), + patch( # test-quality-ok: update_database reads this proxy_server global at call time, no seam + "litellm.proxy.proxy_server.litellm_proxy_budget_name", "test-budget" + ), + patch( # test-quality-ok: update_database imports the payload builder inside its body, no seam + "litellm.proxy.spend_tracking.spend_tracking_utils.get_logging_payload", + return_value=payload, + ), + ): + charged = await db_writer.update_database( + token="test-token", + user_id="test-user", + end_user_id=None, + team_id=None, + org_id=None, + kwargs={"model": "gpt-5.6-luna", "call_type": "aretrieve_batch"}, + completion_response=None, + start_time=datetime.now(timezone.utc), + end_time=datetime.now(timezone.utc), + response_cost=response_cost, + ) + await asyncio.sleep(0) + return charged + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("inserted", "existing", "charged"), + [ + (1, None, True), + (0, SimpleNamespace(call_type="aretrieve_batch", status="success", spend=0.25), False), + (0, SimpleNamespace(call_type="aretrieve_batch", status="success", spend=0.0), True), + (0, SimpleNamespace(call_type="aretrieve_batch", status="failure", spend=0.0), True), + (0, SimpleNamespace(call_type="aembedding", status="success", spend=0.25), True), + (0, None, True), + ], + ids=[ + "first_retrieve_owns_the_row", + "another_retrieve_already_charged", + "an_older_proxy_left_a_zero_row_while_the_batch_ran", + "failed_retrieve_holds_the_row", + "client_chosen_call_id_holds_the_row", + "row_gone_between_insert_and_lookup", + ], +) +async def test_update_database_charges_a_batch_only_from_the_retrieve_that_wrote_its_row( + inserted: int, existing: object, charged: bool +): + """ + Every retrieve of one batch shares one spend row, so the insert that lands first is + the charge and every later retrieve must leave the counters alone (LIT-7048). A row + that recorded no charge must not be able to take the charge away: neither one a + client planted under the batch id, nor the $0 row a pre-upgrade proxy wrote every + time it polled the batch while it was still running. + """ + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(inserted, existing) + + assert await _update_database_with(db_writer, prisma, _batch_cost_payload()) is charged + + claimed_rows = prisma.db.litellm_spendlogs.create_many.await_args.kwargs + assert claimed_rows["skip_duplicates"] is True + assert [(row["request_id"], row["spend"]) for row in claimed_rows["data"]] == [("batch_abc_batch_cost", 0.25)] + assert prisma.spend_log_transactions == [] + assert db_writer._batch_database_updates.await_count == (1 if charged else 0) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("taken_over", "charged"), + [(1, True), (0, False)], + ids=["this_retrieve_takes_it_over", "another_one_got_there_first"], +) +async def test_update_database_charges_a_batch_whose_row_a_pre_upgrade_poll_left_at_zero( + taken_over: int, charged: bool +): + """ + A proxy without this fix wrote the batch's row at $0 on every poll of a running batch, + and the row outlives the upgrade, so the charge has to land on the row itself. Charging + without writing it there would charge again on every later retrieve (LIT-7048). + """ + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + existing = SimpleNamespace(call_type="aretrieve_batch", status="success", spend=0.0) + prisma = _spend_logs_prisma(0, existing, taken_over) + + assert await _update_database_with(db_writer, prisma, _batch_cost_payload()) is charged + + taken = prisma.db.litellm_spendlogs.update_many.await_args.kwargs + assert taken["where"] == { + "request_id": "batch_abc_batch_cost", + "call_type": "aretrieve_batch", + "status": "success", + "spend": 0.0, + } + assert taken["data"]["spend"] == 0.25 + assert "request_id" not in taken["data"] + assert db_writer._batch_database_updates.await_count == (1 if charged else 0) + + +@pytest.mark.asyncio +async def test_update_database_leaves_a_batch_whose_zero_row_it_could_not_take_over_to_the_next_retrieve(): + """ + A DB that refuses the takeover leaves the row reading $0, so charging here would charge + the batch again on every later retrieve. The retrieve that does take the row over is the + one that charges. + """ + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + existing = SimpleNamespace(call_type="aretrieve_batch", status="success", spend=0.0) + prisma = _spend_logs_prisma(0, existing) + prisma.db.litellm_spendlogs.update_many = AsyncMock(side_effect=RuntimeError("db unreachable")) + + assert await _update_database_with(db_writer, prisma, _batch_cost_payload()) is False + + assert db_writer._batch_database_updates.await_count == 0 + + +@pytest.mark.asyncio +async def test_update_database_leaves_a_batch_that_cost_nothing_to_the_retrieve_that_wrote_its_row(): + """ + A batch every line of which failed costs $0, so its row reads $0 for the honest reason + and the retrieve that wrote it is still the one that accounted it. Taking that row over + on every later retrieve would count one batch as many requests. + """ + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + existing = SimpleNamespace(call_type="aretrieve_batch", status="success", spend=0.0) + prisma = _spend_logs_prisma(0, existing) + + assert await _update_database_with(db_writer, prisma, _batch_cost_payload(), response_cost=0.0) is False + + prisma.db.litellm_spendlogs.update_many.assert_not_called() + assert db_writer._batch_database_updates.await_count == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("inserted", "existing", "charged"), + [ + (1, None, True), + (0, SimpleNamespace(call_type="aretrieve_batch", status="success", spend=0.25), False), + ], + ids=["first_retrieve_owns_the_row", "another_retrieve_already_charged"], +) +async def test_update_database_charges_a_batch_once_even_with_spend_logs_disabled( + inserted: int, existing: object, charged: bool +): + """ + disable_spend_logs drops the per-request logs, not the batch's charge, so the one row + that makes a batch chargeable exactly once is still written and still read back. + """ + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(inserted, existing) + + assert await _update_database_with(db_writer, prisma, _batch_cost_payload(), True) is charged + + assert prisma.db.litellm_spendlogs.create_many.await_count == 1 + assert db_writer._batch_database_updates.await_count == (1 if charged else 0) + + +@pytest.mark.asyncio +async def test_update_database_writes_no_ordinary_spend_row_with_spend_logs_disabled(): + """The batch carve-out above stays a carve-out: every other row still goes unwritten.""" + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(1, None) + payload = {**_batch_cost_payload(), "call_type": "acompletion"} + + assert await _update_database_with(db_writer, prisma, payload, True) is True + + prisma.db.litellm_spendlogs.create_many.assert_not_called() + assert prisma.spend_log_transactions == [] + assert db_writer._batch_database_updates.await_count == 1 + + +_BATCH_CLAIM_FIELDS = {"request_id", "call_type", "status", "spend", "startTime", "endTime"} + + +def _logged_batch_cost_payload() -> dict: + return { + **_batch_cost_payload(), + "api_key": "0e5b0e9e5f", + "model": "gpt-5.6-luna", + "user": "test-user", + "metadata": '{"batch_models": ["gpt-5.6-luna"]}', + "requester_ip_address": "127.0.0.1", + "proxy_server_request": '{"headers": {"user-agent": "litellm-batch-cost-check"}}', + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("disable_spend_logs", "logs_the_request"), + [(False, True), (True, False)], + ids=["spend_logs_on", "spend_logs_off"], +) +async def test_update_database_claims_a_batch_without_logging_the_request_that_polled_it( + disable_spend_logs: bool, logs_the_request: bool +): + """ + disable_spend_logs has to keep meaning that no request gets logged, and the batch's cost + row is the one row it cannot drop, so with logging off that row carries only what tells + the retrieves apart: no metadata, no requester IP, no key, model, or token counts. + """ + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(1, None) + payload = _logged_batch_cost_payload() + + assert await _update_database_with(db_writer, prisma, payload, disable_spend_logs) is True + + claimed = prisma.db.litellm_spendlogs.create_many.await_args.kwargs["data"][0] + assert set(claimed) == (set(payload) if logs_the_request else _BATCH_CLAIM_FIELDS) + assert claimed["spend"] == 0.25 + assert db_writer._batch_database_updates.await_count == 1 + + +@pytest.mark.asyncio +async def test_update_database_queues_only_the_claim_for_a_batch_it_could_not_write_with_logs_disabled(): + """A refused claim is retried through the queue, so what it queues has to stay unlogged too.""" + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(0, None) + prisma.db.litellm_spendlogs.create_many = AsyncMock(side_effect=RuntimeError("db unreachable")) + + assert await _update_database_with(db_writer, prisma, _logged_batch_cost_payload(), True) is True + + assert [set(row) for row in prisma.spend_log_transactions] == [_BATCH_CLAIM_FIELDS] + assert db_writer._batch_database_updates.await_count == 1 + + +@pytest.mark.asyncio +async def test_update_database_queues_a_batch_cost_row_it_could_not_claim(): + """An unreachable DB must not drop the batch's only spend row, nor its charge.""" + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(0, None) + prisma.db.litellm_spendlogs.create_many = AsyncMock(side_effect=RuntimeError("db unreachable")) + + assert await _update_database_with(db_writer, prisma, _batch_cost_payload()) is True + + assert [row["request_id"] for row in prisma.spend_log_transactions] == ["batch_abc_batch_cost"] + assert db_writer._batch_database_updates.await_count == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "payload", + [{**_batch_cost_payload(), "call_type": "acompletion"}, {**_batch_cost_payload(), "status": "failure"}], + ids=["not_a_batch_retrieve", "failed_batch_retrieve"], +) +async def test_update_database_queues_every_other_spend_row_for_the_next_flush(payload: dict): + db_writer = DBSpendUpdateWriter() + db_writer._batch_database_updates = AsyncMock() + prisma = _spend_logs_prisma(1, None) + + assert await _update_database_with(db_writer, prisma, payload) is True + + prisma.db.litellm_spendlogs.create_many.assert_not_called() + assert prisma.spend_log_transactions == [payload] + assert db_writer._batch_database_updates.await_count == 1 + + @pytest.mark.asyncio @pytest.mark.parametrize( "injected_deployment, attributed", diff --git a/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py b/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py index f3fd9f41cc2..eff892f2d80 100644 --- a/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py +++ b/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py @@ -1,4 +1,4 @@ - +import asyncio from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch @@ -70,9 +70,7 @@ async def test_async_post_call_failure_hook(): # Check that metadata was properly updated assert "litellm_params" in call_args["kwargs"] - assert call_args["kwargs"]["litellm_params"]["proxy_server_request"] == { - "request_id": "test_request_id" - } + assert call_args["kwargs"]["litellm_params"]["proxy_server_request"] == {"request_id": "test_request_id"} metadata = call_args["kwargs"]["litellm_params"]["metadata"] assert metadata["user_api_key"] == "test_api_key" assert metadata["status"] == "failure" @@ -336,9 +334,7 @@ async def test_should_continue_failure_tracking_when_budget_release_fails(): ) assert mock_invalidate_budget_reservation_counters.await_count == 1 assert ( - mock_invalidate_budget_reservation_counters.await_args.kwargs[ - "budget_reservation" - ] + mock_invalidate_budget_reservation_counters.await_args.kwargs["budget_reservation"] is user_api_key_dict.budget_reservation ) assert user_api_key_dict.budget_reservation["finalized"] is True @@ -433,36 +429,21 @@ def test_get_budget_reservation_from_metadata_handles_dict_auth_object(): "entries": [{"counter_key": "spend:key:test_api_key"}], } + assert _get_budget_reservation_from_metadata(metadata={"user_api_key_auth": dict(UserAPIKeyAuth())}) is None assert ( _get_budget_reservation_from_metadata( - metadata={"user_api_key_auth": dict(UserAPIKeyAuth())} - ) - is None - ) - assert ( - _get_budget_reservation_from_metadata( - metadata={ - "user_api_key_auth": UserAPIKeyAuth( - budget_reservation=budget_reservation - ) - } + metadata={"user_api_key_auth": UserAPIKeyAuth(budget_reservation=budget_reservation)} ) == budget_reservation ) assert ( _get_budget_reservation_from_metadata( - metadata={ - "user_api_key_auth": dict( - UserAPIKeyAuth(budget_reservation=budget_reservation) - ) - } + metadata={"user_api_key_auth": dict(UserAPIKeyAuth(budget_reservation=budget_reservation))} ) == budget_reservation ) assert ( - _get_budget_reservation_from_metadata( - metadata={"user_api_key_budget_reservation": budget_reservation} - ) + _get_budget_reservation_from_metadata(metadata={"user_api_key_budget_reservation": budget_reservation}) is budget_reservation ) @@ -470,9 +451,7 @@ def test_get_budget_reservation_from_metadata_handles_dict_auth_object(): @pytest.mark.asyncio async def test_update_database_and_spend_counters_releases_reservation_when_db_update_fails(): proxy_logging_obj = MagicMock() - proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock( - side_effect=Exception("db unavailable") - ) + proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock(side_effect=Exception("db unavailable")) increment_spend_counters = AsyncMock() budget_reservation = {"reserved_cost": 0.5, "entries": []} @@ -508,9 +487,7 @@ async def test_update_database_and_spend_counters_releases_reservation_when_db_u async def test_update_database_and_spend_counters_preserves_db_exception_when_release_fails(): proxy_logging_obj = MagicMock() db_exception = RuntimeError("db unavailable") - proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock( - side_effect=db_exception - ) + proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock(side_effect=db_exception) increment_spend_counters = AsyncMock() budget_reservation = {"reserved_cost": 0.5, "entries": []} @@ -554,12 +531,8 @@ async def test_update_database_and_spend_counters_preserves_db_exception_when_re budget_reservation=budget_reservation, ) assert mock_log_exception.call_count == 2 - mock_log_exception.assert_any_call( - "Failed to release budget reservation after database update failed" - ) - mock_log_exception.assert_any_call( - "Failed to invalidate budget reservation counters after release failed" - ) + mock_log_exception.assert_any_call("Failed to release budget reservation after database update failed") + mock_log_exception.assert_any_call("Failed to invalidate budget reservation counters after release failed") increment_spend_counters.assert_not_awaited() @@ -778,6 +751,107 @@ async def test_track_cost_callback_defers_in_progress_background_interaction(): mock_proxy_logging.failed_tracking_alert.assert_not_called() +def _batch_retrieve_kwargs(call_type: str, reservation: dict | None = None) -> dict: + metadata = { + "user_api_key": "hashed_key", + "user_api_key_user_id": "user-1", + "user_api_key_team_id": "team-1", + **({"user_api_key_budget_reservation": reservation} if reservation is not None else {}), + } + return { + "call_type": call_type, + "model": "gpt-5.6-luna", + "litellm_call_id": "test-call-id", + "litellm_params": {"metadata": metadata}, + "standard_logging_object": {"response_cost": 0.0, "request_tags": None}, + "stream": False, + } + + +def _retrieved_batch(status: str, output_file_id: str | None): + from litellm.types.utils import LiteLLMBatch + + return LiteLLMBatch( + id="batch_abc", + completion_window="24h", + created_at=1, + endpoint="/v1/chat/completions", + input_file_id="file-in", + object="batch", + status=status, + output_file_id=output_file_id, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("call_type", "status", "output_file_id", "row_claimed", "spend_written", "charged"), + [ + ("aretrieve_batch", "in_progress", None, True, False, False), + ("aretrieve_batch", "completed", None, True, False, False), + ("aretrieve_batch", "completed", "file-out", False, True, False), + ("aretrieve_batch", "completed", "file-out", True, True, True), + ("aretrieve_batch", "failed", None, True, True, True), + ("acreate_batch", "validating", None, True, True, True), + ], + ids=[ + "retrieve_before_final", + "retrieve_completed_without_output_yet", + "retrieve_after_another_retrieve_charged", + "retrieve_first_final", + "retrieve_failed_batch", + "create_before_final", + ], +) +async def test_track_cost_callback_charges_a_batch_once_and_only_when_final( # test-quality-ok: whether the spend writer runs, whether the counters move, and whether the poll's reservation is handed back is the whole observable contract of the gate + call_type, status, output_file_id, row_claimed, spend_written, charged +): + """ + A poll before the batch is final used to pin its shared spend row at $0, and every + completed retrieve after the first charged the key again (LIT-7048). Only retrieves + are gated, since creating a batch is its own billable request, and a retrieve that + charges nothing hands its budget reservation back instead. + """ + logger = _ProxyDBLogger() + budget_reservation = None if charged else {"reserved_cost": 0.5, "entries": []} + kwargs = _batch_retrieve_kwargs(call_type, reservation=budget_reservation) + + with ( + patch( # test-quality-ok: increment_spend_counters is a proxy_server global the callback reads lazily, no seam + "litellm.proxy.proxy_server.increment_spend_counters", new_callable=AsyncMock + ) as mock_increment_spend_counters, + patch( # test-quality-ok: update_cache is a proxy_server global the callback reads lazily, no seam + "litellm.proxy.proxy_server.update_cache", new_callable=AsyncMock + ) as mock_update_cache, + patch( # test-quality-ok: callback imports proxy_logging_obj off proxy_server in its body, no seam + "litellm.proxy.proxy_server.proxy_logging_obj" + ) as mock_proxy_logging, + patch( # test-quality-ok: the release is imported inside the callback's helper, no seam + "litellm.proxy.spend_tracking.budget_reservation.release_budget_reservation", new_callable=AsyncMock + ) as mock_release_budget_reservation, + ): + mock_proxy_logging.failed_tracking_alert = AsyncMock() + mock_proxy_logging.db_spend_update_writer.update_database = AsyncMock(return_value=row_claimed) + mock_proxy_logging.slack_alerting_instance.customer_spend_alert = AsyncMock() + + await logger._PROXY_track_cost_callback( + kwargs=kwargs, + completion_response=_retrieved_batch(status, output_file_id), + start_time=datetime.now(), + end_time=datetime.now(), + ) + await asyncio.sleep(0) + + mock_proxy_logging.failed_tracking_alert.assert_not_called() + assert mock_proxy_logging.db_spend_update_writer.update_database.await_count == (1 if spend_written else 0) + assert mock_increment_spend_counters.await_count == (1 if charged else 0) + assert mock_update_cache.await_count == (1 if charged else 0) + if charged: + mock_release_budget_reservation.assert_not_awaited() + else: + mock_release_budget_reservation.assert_awaited_once_with(budget_reservation=budget_reservation) + + def _in_progress_interaction_kwargs(reservation: dict) -> dict: return { "call_type": "acreate_interaction", @@ -1101,10 +1175,7 @@ async def test_async_post_call_failure_hook_propagates_trace_id_from_logging_obj # standard_logging_object should have been propagated from logging obj assert call_kwargs.get("standard_logging_object") is not None - assert ( - call_kwargs["standard_logging_object"]["trace_id"] - == "trace-id-from-logging-obj" - ) + assert call_kwargs["standard_logging_object"]["trace_id"] == "trace-id-from-logging-obj" # litellm_trace_id should also be propagated as a fallback assert call_kwargs.get("litellm_trace_id") == "trace-id-from-logging-obj" @@ -1691,9 +1762,7 @@ async def test_async_post_call_failure_hook_records_recovered_partial_spend(): "metadata": {}, "proxy_server_request": {"request_id": "rid"}, "response_cost": 3.5e-05, - "combined_usage_object": Usage( - prompt_tokens=30, completion_tokens=1, total_tokens=31 - ), + "combined_usage_object": Usage(prompt_tokens=30, completion_tokens=1, total_tokens=31), } with patch( @@ -1772,15 +1841,10 @@ async def test_track_cost_callback_enriches_user_id_for_mcp_style_metadata(): assert mock_increment.call_args.kwargs["team_id"] == "team-123" assert mock_increment.call_args.kwargs["org_id"] == "org-456" - update_kwargs = ( - mock_proxy_logging.db_spend_update_writer.update_database.await_args.kwargs - ) + update_kwargs = mock_proxy_logging.db_spend_update_writer.update_database.await_args.kwargs assert update_kwargs["user_id"] == "mcp-user@example.com" assert update_kwargs["team_id"] == "team-123" - assert ( - kwargs["litellm_params"]["metadata"]["user_api_key_user_id"] - == "mcp-user@example.com" - ) + assert kwargs["litellm_params"]["metadata"]["user_api_key_user_id"] == "mcp-user@example.com" @pytest.mark.asyncio @@ -1875,9 +1939,7 @@ def test_should_track_cost_callback_pass_through_without_owner(call_type, expect ], ) @pytest.mark.asyncio -async def test_track_cost_callback_logs_unauthenticated_pass_through_request( - call_type, expect_spend_log -): +async def test_track_cost_callback_logs_unauthenticated_pass_through_request(call_type, expect_spend_log): """Regression for LIT-3782: a pass-through request with auth=false reaches the cost callback with no key/user/team/end-user. Before the fix the spend-log write was skipped and the request never appeared in request/usage logs. It @@ -1923,9 +1985,7 @@ async def test_track_cost_callback_logs_unauthenticated_pass_through_request( end_time=datetime.now(), ) - assert mock_proxy_logging.db_spend_update_writer.update_database.await_count == ( - 1 if expect_spend_log else 0 - ) + assert mock_proxy_logging.db_spend_update_writer.update_database.await_count == (1 if expect_spend_log else 0) class _FakeDeploymentLookup: diff --git a/tests/test_litellm/router_utils/test_reasoning_effort_capability.py b/tests/test_litellm/router_utils/test_reasoning_effort_capability.py index f181370455d..ccd6766b13a 100644 --- a/tests/test_litellm/router_utils/test_reasoning_effort_capability.py +++ b/tests/test_litellm/router_utils/test_reasoning_effort_capability.py @@ -389,14 +389,24 @@ class TestGpt6AstraAdvertisesItsDocumentedLevels: "max", ) - @pytest.mark.parametrize("model", ["azure/gpt-6-astra", "azure/us/gpt-6-astra"]) - def test_a_foundry_deployment_also_advertises_none(self, local_model_cost_map, model): - """Microsoft Foundry serves the same model but its API accepts reasoning_effort none - (verified live: 200 with zero reasoning tokens, and it unlocks temperature), which - OpenAI's rejects, so an Azure deployment offers none on top of low through max.""" + @pytest.mark.parametrize( + "model,custom_llm_provider", + [ + ("azure/gpt-6-astra", "azure"), + ("azure/us/gpt-6-astra", "azure"), + ("azure_ai/gpt-6-astra", "azure_ai"), + ], + ) + def test_an_azure_hosted_deployment_advertises_none_but_not_max( + self, local_model_cost_map, model, custom_llm_provider + ): + """Microsoft hosts the same model with a different level set than OpenAI does. Verified live + on both Azure routes: none returns 200 with zero reasoning tokens and unlocks temperature, + which OpenAI's API rejects, while max returns 400 unsupported_value naming none through + xhigh as the levels it does take.""" from litellm.utils import _get_model_info_helper - model_info = dict(_get_model_info_helper(model=model, custom_llm_provider="azure")) + model_info = dict(_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)) assert resolve_supported_reasoning_efforts(model_info, deployment_is_mapped=True) == ( "none", @@ -404,5 +414,4 @@ class TestGpt6AstraAdvertisesItsDocumentedLevels: "medium", "high", "xhigh", - "max", )