fix(ci): satisfy fusion type discipline

litellm/fusion_router.py

@@ -152,14 +152,20 @@ class FusionCompletionCaller(Protocol):
         self,
         *,
         model: str,
-        messages: list[AllMessageValues],
+        messages: list[AllMessageValues],  # mutable-ok: SDK boundary
         stream: bool,
-        **kwargs: object,
+        **kwargs: object,  # kwargs-ok: SDK passthrough
     ) -> Awaitable[ModelResponse | CustomStreamWrapper]: ...
 
 
 class FusionSearchCaller(Protocol):
-    def __call__(self, *, model: str, query: str, **kwargs: object) -> Awaitable[object]: ...
+    def __call__(
+        self,
+        *,
+        model: str,
+        query: str,
+        **kwargs: object,  # kwargs-ok: SDK passthrough
+    ) -> Awaitable[object]: ...
 
 
 class FusionStance(BaseModel):
@@ -206,11 +212,15 @@ class FusionCandidate:
     content: str
 
     def prompt_value(self, max_chars: int) -> Mapping[str, object]:
-        content = self.content[:max_chars]
-        return {
+        content: Final = self.content[:max_chars]
+        return {  # mutable-ok: local provider payload
             "model": self.model,
             "content": content,
-            **({"truncated": True} if len(self.content) > max_chars else {}),
+            **(
+                {"truncated": True}  # mutable-ok: provider payload requires a native mapping
+                if len(self.content) > max_chars
+                else {}  # mutable-ok: local provider payload
+            ),  # mutable-ok: local provider payload
         }
 
 
@@ -272,7 +282,7 @@ def _request_metadata(request_kwargs: Mapping[str, object]) -> Mapping[str, obje
 def _fusion_call_metadata(
     request_kwargs: Mapping[str, object],
     origin: InternalCallOrigin,
-) -> dict[str, object]:
+) -> dict[str, object]:  # mutable-ok: SDK boundary
     """Forward attribution and keep the parent reservation on Fusion-owned calls.
 
     Fusion is one logical request with several billed provider calls. Its cost
@@ -281,10 +291,10 @@ def _fusion_call_metadata(
     continue to use ``forwarded_internal_call_metadata``, which strips a parent
     reservation to prevent accidental early finalization.
     """
-    parent_metadata = _request_metadata(request_kwargs)
-    metadata = forwarded_internal_call_metadata(parent_metadata, origin)
+    parent_metadata: Final = _request_metadata(request_kwargs)
+    metadata: Final = forwarded_internal_call_metadata(parent_metadata, origin)
     if parent_metadata is not None:
-        reservation = parent_metadata.get(_BUDGET_RESERVATION_METADATA_KEY)
+        reservation: Final = parent_metadata.get(_BUDGET_RESERVATION_METADATA_KEY)
         if isinstance(reservation, dict):
             reservation[FUSION_BUDGET_ACTIVE_KEY] = True
             metadata[_BUDGET_RESERVATION_METADATA_KEY] = reservation
@@ -298,8 +308,8 @@ def _internal_kwargs(
     origin: InternalCallOrigin,
     model: str,
     messages: Sequence[AllMessageValues],
-) -> dict[str, object]:
-    kwargs = {
+) -> dict[str, object]:  # mutable-ok: SDK boundary
+    kwargs: Final = {  # mutable-ok: local provider payload
         key: value
         for key, value in request_kwargs.items()
         if key not in _INTERNAL_REQUEST_KEYS
@@ -310,10 +320,15 @@ def _internal_kwargs(
     kwargs.pop("litellm_metadata", None)
     kwargs.pop("max_tokens", None)
     kwargs.pop("max_completion_tokens", None)
-    metadata = _fusion_call_metadata(request_kwargs, origin)
+    metadata: Final = _fusion_call_metadata(request_kwargs, origin)
     kwargs["metadata"] = metadata
     kwargs["drop_params"] = True
-    kwargs["proxy_server_request"] = {"body": {"model": model, "messages": list(messages)}}
+    kwargs["proxy_server_request"] = {  # mutable-ok: local provider payload
+        "body": {  # mutable-ok: provider payload requires a native mapping
+            "model": model,
+            "messages": list(messages),  # mutable-ok: provider payload requires a native list
+        }  # mutable-ok: local provider payload
+    }  # mutable-ok: local provider payload
     kwargs["_fusion_depth"] = 1
     return kwargs
 
@@ -322,24 +337,24 @@ def _fusion_tool() -> Mapping[str, object]:
     # This is deliberately a normal function schema at the provider boundary.
     # `litellm_fusion` is private to this orchestration layer and is never sent
     # to a panel, analyst, or returned to the caller as an executable tool.
-    return {
+    return {  # mutable-ok: local provider payload
         "type": "function",
-        "function": {
+        "function": {  # mutable-ok: local provider payload
             "name": FUSION_TOOL_NAME,
             "description": (
                 "Ask several independent models to investigate a difficult request before you answer. "
                 "Use this for uncertainty, multi-step analysis, important decisions, or questions helped by "
                 "independent perspectives. Skip it for simple or routine requests."
             ),
-            "parameters": {
+            "parameters": {  # mutable-ok: local provider payload
                 "type": "object",
-                "properties": {
-                    "query": {
+                "properties": {  # mutable-ok: local provider payload
+                    "query": {  # mutable-ok: local provider payload
                         "type": "string",
                         "description": "A self-contained question for the independent panel.",
                     }
                 },
-                "required": ["query"],
+                "required": ["query"],  # mutable-ok: local provider payload
                 "additionalProperties": False,
             },
         },
@@ -347,15 +362,19 @@ def _fusion_tool() -> Mapping[str, object]:
 
 
 def _research_tool() -> Mapping[str, object]:
-    return {
+    return {  # mutable-ok: local provider payload
         "type": "function",
-        "function": {
+        "function": {  # mutable-ok: local provider payload
             "name": "litellm_fusion_search",
             "description": "Search the web for evidence needed by the private Fusion deliberation.",
-            "parameters": {
+            "parameters": {  # mutable-ok: local provider payload
                 "type": "object",
-                "properties": {"query": {"type": "string"}},
-                "required": ["query"],
+                "properties": {  # mutable-ok: local provider payload
+                    "query": {  # mutable-ok: function schema requires a native mapping
+                        "type": "string"
+                    }  # mutable-ok: local provider payload
+                },  # mutable-ok: local provider payload
+                "required": ["query"],  # mutable-ok: local provider payload
                 "additionalProperties": False,
             },
         },
@@ -374,13 +393,13 @@ def _fusion_tool_call(response: ModelResponse) -> ChatCompletionMessageToolCall
 def _mixed_tool_call_indexes(response: ModelResponse) -> tuple[frozenset[int], tuple[int, ...]]:
     if not response.choices:
         return frozenset(), ()
-    tool_calls = response.choices[0].message.tool_calls or ()
-    fusion_indexes = frozenset(
+    tool_calls: Final = response.choices[0].message.tool_calls or ()
+    fusion_indexes: Final = frozenset(
         index
         for index, tool_call in enumerate(tool_calls)
         if isinstance(tool_call, ChatCompletionMessageToolCall) and tool_call.function.name == FUSION_TOOL_NAME
     )
-    client_indexes = tuple(index for index in range(len(tool_calls)) if index not in fusion_indexes)
+    client_indexes: Final = tuple(index for index in range(len(tool_calls)) if index not in fusion_indexes)
     return fusion_indexes, client_indexes
 
 
@@ -389,20 +408,22 @@ def _without_mixed_fusion_tool_call(response: ModelResponse) -> tuple[ModelRespo
     fusion_indexes, client_indexes = _mixed_tool_call_indexes(response)
     if not fusion_indexes or not client_indexes:
         return response, frozenset()
-    sanitized = response.model_copy(deep=True)
-    tool_calls = sanitized.choices[0].message.tool_calls or ()
-    sanitized.choices[0].message.tool_calls = [tool_calls[index] for index in client_indexes]
+    sanitized: Final = response.model_copy(deep=True)
+    tool_calls: Final = sanitized.choices[0].message.tool_calls or ()
+    sanitized.choices[0].message.tool_calls = [  # mutable-ok: local provider payload
+        tool_calls[index] for index in client_indexes
+    ]  # mutable-ok: local provider payload
     return sanitized, fusion_indexes
 
 
 def _without_stream_tool_call_indexes(
     chunks: Sequence[ModelResponseStream],
     removed_indexes: frozenset[int],
-) -> list[ModelResponseStream]:
+) -> list[ModelResponseStream]:  # mutable-ok: SDK boundary
     if not removed_indexes:
-        return list(chunks)
-    kept_indexes = sorted(
-        {
+        return list(chunks)  # mutable-ok: local provider payload
+    kept_indexes: Final = sorted(
+        {  # mutable-ok: local provider payload
             tool_call.index
             for chunk in chunks
             for choice in chunk.choices
@@ -410,14 +431,22 @@ def _without_stream_tool_call_indexes(
             if tool_call.index not in removed_indexes
         }
     )
-    index_map = {old_index: new_index for new_index, old_index in enumerate(kept_indexes)}
-    sanitized_chunks: list[ModelResponseStream] = []
+    index_map: Final = {  # mutable-ok: local provider payload
+        old_index: new_index for new_index, old_index in enumerate(kept_indexes)
+    }  # mutable-ok: local provider payload
+    sanitized_chunks: Final[  # mutable-ok: SDK boundary
+        list[ModelResponseStream]
+    ] = []  # mutable-ok: SDK boundary
     for chunk in chunks:
         sanitized = chunk.model_copy(deep=True)
         for choice in sanitized.choices:
             tool_calls = choice.delta.tool_calls or ()
-            choice.delta.tool_calls = [
-                tool_call.model_copy(update={"index": index_map[tool_call.index]})
+            choice.delta.tool_calls = [  # mutable-ok: local provider payload
+                tool_call.model_copy(
+                    update={  # mutable-ok: Pydantic update requires a native mapping
+                        "index": index_map[tool_call.index]
+                    }  # mutable-ok: local provider payload
+                )  # mutable-ok: local provider payload
                 for tool_call in tool_calls
                 if tool_call.index in index_map
             ] or None
@@ -427,10 +456,10 @@ def _without_stream_tool_call_indexes(
 
 def _fusion_query(tool_call: ChatCompletionMessageToolCall) -> str | None:
     try:
-        arguments = _OBJECT_MAPPING_ADAPTER.validate_json(tool_call.function.arguments)
+        arguments: Final = _OBJECT_MAPPING_ADAPTER.validate_json(tool_call.function.arguments)
     except (TypeError, ValidationError):
         return None
-    query = arguments.get("query")
+    query: Final = arguments.get("query")
     return query.strip() if isinstance(query, str) and query.strip() else None
 
 
@@ -448,16 +477,24 @@ def _bounded_search_arguments(query: str | None, max_chars: int) -> str:
     """Return valid search arguments whose serialized form fits the configured bound."""
     if query is None:
         return "{}"
-    low = 0
-    high = len(query)
+    low = 0  # rebind-ok: orchestration branch state
+    high = len(query)  # rebind-ok: orchestration branch state
     while low < high:
         midpoint = (low + high + 1) // 2
-        serialized = json.dumps({"query": query[:midpoint]}, ensure_ascii=False, separators=(",", ":"))
+        serialized = json.dumps(
+            {"query": query[:midpoint]},  # mutable-ok: JSON payload requires a native mapping
+            ensure_ascii=False,
+            separators=(",", ":"),  # mutable-ok: local provider payload
+        )  # mutable-ok: local provider payload
         if len(serialized) <= max_chars:
             low = midpoint
         else:
             high = midpoint - 1
-    return json.dumps({"query": query[:low]}, ensure_ascii=False, separators=(",", ":"))
+    return json.dumps(
+        {"query": query[:low]},  # mutable-ok: JSON payload requires a native mapping
+        ensure_ascii=False,
+        separators=(",", ":"),  # mutable-ok: local provider payload
+    )  # mutable-ok: local provider payload
 
 
 def _bounded_research_tool_call(
@@ -468,7 +505,7 @@ def _bounded_research_tool_call(
     return ChatCompletionMessageToolCall(
         id=f"fusion-search-{sequence}",
         type="function",
-        function={
+        function={  # mutable-ok: local provider payload
             "name": "litellm_fusion_search",
             "arguments": _bounded_search_arguments(_fusion_query(tool_call), max_chars),
         },
@@ -478,7 +515,7 @@ def _bounded_research_tool_call(
 def _response_text(response: ModelResponse) -> str | None:
     if not response.choices:
         return None
-    content = response.choices[0].message.content
+    content: Final = response.choices[0].message.content
     if isinstance(content, str):
         return content.strip() or None
     if content is None:
@@ -497,20 +534,22 @@ def _failure_reason(exc: Exception) -> str:
 def _parse_analysis(content: str | None) -> FusionAnalysis | None:
     if content is None:
         return None
-    stripped = content.strip()
+    stripped = content.strip()  # rebind-ok: orchestration branch state
     if stripped.startswith("```"):
-        lines = stripped.splitlines()
+        lines: Final = stripped.splitlines()
         if len(lines) >= 3 and lines[-1].strip() == "```":
-            stripped = "\n".join(lines[1:-1])
+            stripped = "\n".join(lines[1:-1])  # rebind-ok: orchestration branch state
     try:
         return FusionAnalysis.model_validate_json(stripped)
     except ValidationError:
         return None
 
 
-def _panel_messages(query: str) -> list[AllMessageValues]:
-    return [
-        {
+def _panel_messages(
+    query: str,
+) -> list[AllMessageValues]:  # mutable-ok: SDK boundary
+    return [  # mutable-ok: local provider payload
+        {  # mutable-ok: local provider payload
             "role": "system",
             "content": (
                 "You are one independent member of a deliberation panel. Investigate the question, reason "
@@ -518,18 +557,22 @@ def _panel_messages(query: str) -> list[AllMessageValues]:
                 "is advisory; do not pretend to execute tools or actions."
             ),
         },
-        {"role": "user", "content": query},
+        {"role": "user", "content": query},  # mutable-ok: local provider payload
     ]
 
 
-def _analyst_messages(query: str, candidates: Sequence[FusionCandidate], max_chars: int) -> list[AllMessageValues]:
-    candidate_json = json.dumps(
-        [candidate.prompt_value(max_chars) for candidate in candidates],
+def _analyst_messages(
+    query: str, candidates: Sequence[FusionCandidate], max_chars: int
+) -> list[AllMessageValues]:  # mutable-ok: SDK boundary
+    candidate_json: Final = json.dumps(
+        [  # mutable-ok: local provider payload
+            candidate.prompt_value(max_chars) for candidate in candidates
+        ],  # mutable-ok: local provider payload
         ensure_ascii=False,
         separators=(",", ":"),
     )
-    return [
-        {
+    return [  # mutable-ok: local provider payload
+        {  # mutable-ok: local provider payload
             "role": "system",
             "content": (
                 "You are the analyst for an independent model panel. Compare the responses; do not choose a "
@@ -540,7 +583,7 @@ def _analyst_messages(query: str, candidates: Sequence[FusionCandidate], max_cha
                 "(string array)."
             ),
         },
-        {
+        {  # mutable-ok: local provider payload
             "role": "user",
             "content": f"Question:\n{query}\n\nPanel responses:\n{candidate_json}",
         },
@@ -555,21 +598,23 @@ def _tool_result_payload(
     max_candidate_chars: int,
 ) -> Mapping[str, object]:
     if not candidates:
-        reasons = {failure.failure_reason for failure in failures}
-        failure_reason = (
+        reasons: Final = {  # mutable-ok: local provider payload
+            failure.failure_reason for failure in failures
+        }  # mutable-ok: local provider payload
+        failure_reason: Final = (
             "insufficient_credits"
             if "insufficient_credits" in reasons
             else "rate_limited"
             if "rate_limited" in reasons
             else "all_panels_failed"
         )
-        return {
+        return {  # mutable-ok: local provider payload
             "status": "error",
             "error": "all panel models failed",
             "failure_reason": failure_reason,
             "query": query,
-            "failed_models": [
-                {
+            "failed_models": [  # mutable-ok: local provider payload
+                {  # mutable-ok: local provider payload
                     "model": failure.model,
                     "error_type": failure.error_type,
                     "failure_reason": failure.failure_reason,
@@ -577,15 +622,21 @@ def _tool_result_payload(
                 for failure in failures
             ],
         }
-    return {
+    return {  # mutable-ok: local provider payload
         "status": "ok",
         "query": query,
-        "responses": [candidate.prompt_value(max_candidate_chars) for candidate in candidates],
-        **({"analysis": analysis.model_dump()} if analysis is not None else {}),
+        "responses": [  # mutable-ok: local provider payload
+            candidate.prompt_value(max_candidate_chars) for candidate in candidates
+        ],  # mutable-ok: local provider payload
         **(
-            {
-                "failed_models": [
-                    {
+            {"analysis": analysis.model_dump()}  # mutable-ok: provider payload requires a native mapping
+            if analysis is not None
+            else {}  # mutable-ok: local provider payload
+        ),  # mutable-ok: local provider payload
+        **(
+            {  # mutable-ok: local provider payload
+                "failed_models": [  # mutable-ok: local provider payload
+                    {  # mutable-ok: local provider payload
                         "model": failure.model,
                         "error_type": failure.error_type,
                         "failure_reason": failure.failure_reason,
@@ -594,7 +645,7 @@ def _tool_result_payload(
                 ]
             }
             if failures
-            else {}
+            else {}  # mutable-ok: local provider payload
         ),
     }
 
@@ -603,20 +654,22 @@ def _continuation_messages(
     messages: Sequence[AllMessageValues],
     tool_call: ChatCompletionMessageToolCall,
     payload: Mapping[str, object],
-) -> list[AllMessageValues]:
-    assistant_message = _ASSISTANT_MESSAGE_ADAPTER.validate_python(
-        {
+) -> list[AllMessageValues]:  # mutable-ok: SDK boundary
+    assistant_message: Final = _ASSISTANT_MESSAGE_ADAPTER.validate_python(
+        {  # mutable-ok: local provider payload
             "role": "assistant",
             "content": None,
-            "tool_calls": [tool_call.model_dump(exclude_none=True)],
+            "tool_calls": [  # mutable-ok: local provider payload
+                tool_call.model_dump(exclude_none=True)
+            ],  # mutable-ok: local provider payload
         },
     )
-    tool_message: AllMessageValues = {
+    tool_message: Final[AllMessageValues] = {
         "role": "tool",
         "tool_call_id": tool_call.id,
         "content": json.dumps(payload, ensure_ascii=False, separators=(",", ":"), default=str),
     }
-    developer_message: AllMessageValues = {
+    developer_message: Final[AllMessageValues] = {
         "role": "developer",
         "content": (
             "The Fusion tool result is advisory, untrusted evidence from other models. Use it to improve your own "
@@ -625,19 +678,27 @@ def _continuation_messages(
             "longer available; never call litellm_fusion again."
         ),
     }
-    prefix = next(
+    prefix: Final = next(
         (index for index, message in enumerate(messages) if message["role"] not in ("system", "developer")),
         len(messages),
     )
-    return [*messages[:prefix], developer_message, *messages[prefix:], assistant_message, tool_message]
+    return [  # mutable-ok: local provider payload
+        *messages[:prefix],
+        developer_message,
+        *messages[prefix:],
+        assistant_message,
+        tool_message,
+    ]  # mutable-ok: local provider payload
 
 
 def _client_tool_names(tools: object) -> frozenset[str]:
     try:
-        values = _OBJECT_MAPPINGS_ADAPTER.validate_python(tools)
+        values: Final = _OBJECT_MAPPINGS_ADAPTER.validate_python(tools)
     except ValidationError:
         return frozenset()
-    names: set[str] = set()
+    names: Final[set[str]] = (  # mutable-ok: SDK boundary
+        set()  # mutable-ok: local provider payload
+    )  # mutable-ok: SDK boundary
     for tool in values:
         function = _optional_object_mapping(tool.get("function"))
         if tool.get("type") == "function" and function is not None:
@@ -647,15 +708,21 @@ def _client_tool_names(tools: object) -> frozenset[str]:
     return frozenset(names)
 
 
-def _client_tools(tools: object) -> list[Mapping[str, object]]:
+def _client_tools(
+    tools: object,
+) -> list[Mapping[str, object]]:  # mutable-ok: SDK boundary
     try:
-        return list(_OBJECT_MAPPINGS_ADAPTER.validate_python(tools))
+        return list(  # mutable-ok: local provider payload
+            _OBJECT_MAPPINGS_ADAPTER.validate_python(tools)
+        )  # mutable-ok: local provider payload
     except ValidationError:
-        return []
+        return []  # mutable-ok: local provider payload
 
 
-def _outer_kwargs(request_kwargs: Mapping[str, object]) -> dict[str, object]:
-    return {
+def _outer_kwargs(
+    request_kwargs: Mapping[str, object],
+) -> dict[str, object]:  # mutable-ok: SDK boundary
+    return {  # mutable-ok: local provider payload
         key: value
         for key, value in request_kwargs.items()
         if key not in _RESPONSES_ONLY_REQUEST_KEYS
@@ -689,12 +756,20 @@ class FusionReplayStream(CustomStreamWrapper):
         # Deliberately do not call CustomStreamWrapper.__init__. The source
         # wrapper already normalized and logged these chunks while Fusion
         # buffered them to determine whether its private tool was invoked.
-        source_model = getattr(source, "model", "")
+        source_model: Final = getattr(source, "model", "")
         self.model = source_model if isinstance(source_model, str) else ""
         self.custom_llm_provider = source.custom_llm_provider
         self.logging_obj = source.logging_obj
-        self._hidden_params = dict(getattr(source, "_hidden_params", {}))
-        self._hidden_params["fusion"] = dict(fusion_metadata)
+        self._hidden_params = dict(  # mutable-ok: local provider payload
+            getattr(
+                source,
+                "_hidden_params",
+                {},  # mutable-ok: stream metadata defaults to a native mapping
+            )  # mutable-ok: local provider payload
+        )  # mutable-ok: local provider payload
+        self._hidden_params["fusion"] = dict(  # mutable-ok: local provider payload
+            fusion_metadata
+        )  # mutable-ok: local provider payload
         self._source = source
         self._iterator = iter(chunks)
 
@@ -730,15 +805,21 @@ class FusionRouter:
         tool_call: ChatCompletionMessageToolCall,
         request_kwargs: Mapping[str, object],
     ) -> AllMessageValues:
-        query = _fusion_query(tool_call)
+        query: Final = _fusion_query(tool_call)
         if query is None:
-            result: object = {"status": "error", "error": "invalid_search_arguments"}
+            result: object = {  # mutable-ok: local provider payload
+                "status": "error",
+                "error": "invalid_search_arguments",
+            }  # rebind-ok: orchestration branch state  # mutable-ok: local provider payload
         elif self._search is None or self.config.search_tool_name is None:
-            result = {"status": "error", "error": "search_not_configured"}
+            result = {  # mutable-ok: local provider payload
+                "status": "error",
+                "error": "search_not_configured",
+            }  # rebind-ok: orchestration branch state  # mutable-ok: local provider payload
         else:
             try:
-                metadata = _fusion_call_metadata(request_kwargs, FUSION_RESEARCH_CALL_ORIGIN)
-                result = await self._search(
+                metadata: Final = _fusion_call_metadata(request_kwargs, FUSION_RESEARCH_CALL_ORIGIN)
+                result = await self._search(  # rebind-ok: orchestration branch state
                     model=self.config.search_tool_name,
                     query=query,
                     # Search routing stores its internal metadata in the newer
@@ -750,11 +831,14 @@ class FusionRouter:
                     _fusion_proxy_auth_required=isinstance(request_kwargs.get("proxy_server_request"), Mapping),
                 )
                 if isinstance(result, BaseModel):
-                    result = result.model_dump()
+                    result = result.model_dump()  # rebind-ok: orchestration branch state
             except Exception as exc:
-                result = {"status": "error", "error": type(exc).__name__}
-        serialized = json.dumps(result, ensure_ascii=False, separators=(",", ":"), default=str)
-        return {
+                result = {  # mutable-ok: local provider payload
+                    "status": "error",
+                    "error": type(exc).__name__,
+                }  # rebind-ok: orchestration branch state  # mutable-ok: local provider payload
+        serialized: Final = json.dumps(result, ensure_ascii=False, separators=(",", ":"), default=str)
+        return {  # mutable-ok: local provider payload
             "role": "tool",
             "tool_call_id": tool_call.id,
             "content": serialized[: self.config.max_candidate_chars],
@@ -764,20 +848,29 @@ class FusionRouter:
         self,
         *,
         model: str,
-        messages: list[AllMessageValues],
+        messages: list[AllMessageValues],  # mutable-ok: SDK boundary
         kwargs: Mapping[str, object],
         request_kwargs: Mapping[str, object],
     ) -> ModelResponse | CustomStreamWrapper:
-        current_messages = list(messages)
-        remaining_searches = self.config.max_tool_calls if self.config.search_tool_name is not None else 0
+        current_messages: Final = list(  # mutable-ok: local provider payload
+            messages
+        )  # mutable-ok: local provider payload
+        remaining_searches = (
+            self.config.max_tool_calls if self.config.search_tool_name is not None else 0
+        )  # rebind-ok: orchestration branch state
         while True:
-            call_kwargs = dict(kwargs)
+            call_kwargs = dict(kwargs)  # mutable-ok: local provider payload
             if remaining_searches > 0 and self._search is not None:
-                call_kwargs["tools"] = [_research_tool()]
+                call_kwargs["tools"] = [  # mutable-ok: local provider payload
+                    _research_tool()
+                ]  # mutable-ok: local provider payload
                 call_kwargs["tool_choice"] = "auto"
             proxy_request = call_kwargs.get("proxy_server_request")
             if isinstance(proxy_request, dict):
-                proxy_request["body"] = {"model": model, "messages": current_messages}
+                proxy_request["body"] = {  # mutable-ok: local provider payload
+                    "model": model,
+                    "messages": current_messages,
+                }  # mutable-ok: local provider payload
             response = await self._completion(model=model, messages=current_messages, stream=False, **call_kwargs)
             if not isinstance(response, ModelResponse):
                 return response
@@ -797,9 +890,11 @@ class FusionRouter:
             # only bounded, normalized search calls and their results are retained.
             current_messages.append(
                 _ASSISTANT_MESSAGE_ADAPTER.validate_python(
-                    {
+                    {  # mutable-ok: local provider payload
                         "role": "assistant",
-                        "tool_calls": [call.model_dump(exclude_none=True) for call in bounded_calls],
+                        "tool_calls": [  # mutable-ok: local provider payload
+                            call.model_dump(exclude_none=True) for call in bounded_calls
+                        ],  # mutable-ok: local provider payload
                     },
                 )
             )
@@ -810,19 +905,27 @@ class FusionRouter:
 
     async def _initial_outer_call(
         self,
-        messages: list[AllMessageValues],
+        messages: list[AllMessageValues],  # mutable-ok: SDK boundary
         stream: bool,
         request_kwargs: Mapping[str, object],
     ) -> tuple[ModelResponse, FusionReplayStream | None]:
-        kwargs = _outer_kwargs(request_kwargs)
+        kwargs: Final = _outer_kwargs(request_kwargs)
         kwargs.pop("litellm_metadata", None)
         kwargs["metadata"] = _fusion_call_metadata(request_kwargs, FUSION_INITIAL_CALL_ORIGIN)
-        kwargs["tools"] = [*_client_tools(request_kwargs.get("tools")), _fusion_tool()]
+        kwargs["tools"] = [  # mutable-ok: local provider payload
+            *_client_tools(request_kwargs.get("tools")),
+            _fusion_tool(),
+        ]  # mutable-ok: local provider payload
         if self.config.invocation == "required":
-            kwargs["tool_choice"] = {"type": "function", "function": {"name": FUSION_TOOL_NAME}}
+            kwargs["tool_choice"] = {  # mutable-ok: local provider payload
+                "type": "function",
+                "function": {  # mutable-ok: function schema requires a native mapping
+                    "name": FUSION_TOOL_NAME
+                },  # mutable-ok: local provider payload
+            }  # mutable-ok: local provider payload
         elif kwargs.get("tool_choice") is None:
             kwargs["tool_choice"] = "auto"
-        response = await self._completion(
+        response: Final = await self._completion(
             model=self.config.outer_model,
             messages=messages,
             stream=stream,
@@ -833,14 +936,20 @@ class FusionRouter:
             sanitized_response, _ = _without_mixed_fusion_tool_call(response)
             return sanitized_response, None
 
-        chunks: list[ModelResponseStream] = []
+        chunks: Final[  # mutable-ok: SDK boundary
+            list[ModelResponseStream]
+        ] = []  # mutable-ok: SDK boundary
         try:
-            chunks.extend([chunk.model_copy(deep=True) async for chunk in response])
+            chunks.extend(
+                [  # mutable-ok: stream builder requires a native list
+                    chunk.model_copy(deep=True) async for chunk in response
+                ]  # mutable-ok: local provider payload
+            )  # mutable-ok: local provider payload
         except BaseException:
             if hasattr(response, "aclose"):
                 await response.aclose()
             raise
-        built = litellm.stream_chunk_builder(  # pyright: ignore[reportUnknownMemberType]  # public helper lacks complete annotations
+        built = litellm.stream_chunk_builder(  # pyright: ignore[reportUnknownMemberType]  # public helper lacks complete annotations  # rebind-ok: orchestration branch state
             chunks=chunks, messages=messages
         )
         if not isinstance(built, ModelResponse):
@@ -850,19 +959,22 @@ class FusionRouter:
                 llm_provider="",
                 model=self.config.outer_model,
             )
-        built, removed_indexes = _without_mixed_fusion_tool_call(built)
-        replay = FusionReplayStream(
+        built, removed_indexes = _without_mixed_fusion_tool_call(built)  # rebind-ok: orchestration branch state
+        replay: Final = FusionReplayStream(
             source=response,
             chunks=_without_stream_tool_call_indexes(chunks, removed_indexes),
-            fusion_metadata={"invoked": False, "protocol": FUSION_PROTOCOL_VERSION},
+            fusion_metadata={  # mutable-ok: local provider payload
+                "invoked": False,
+                "protocol": FUSION_PROTOCOL_VERSION,
+            },  # mutable-ok: local provider payload
         )
         return built, replay
 
     async def _run_panel_member(
         self, model: str, query: str, request_kwargs: Mapping[str, object]
     ) -> FusionPanelResult:
-        panel_messages = _panel_messages(query)
-        kwargs = _internal_kwargs(
+        panel_messages: Final = _panel_messages(query)
+        kwargs: Final = _internal_kwargs(
             request_kwargs,
             origin=FUSION_PANEL_CALL_ORIGIN,
             model=model,
@@ -897,9 +1009,9 @@ class FusionRouter:
         candidates: Sequence[FusionCandidate],
         request_kwargs: Mapping[str, object],
     ) -> FusionAnalysis | None:
-        messages = _analyst_messages(query, candidates, self.config.max_candidate_chars)
-        model = self.config.resolved_analyst_model
-        kwargs = _internal_kwargs(
+        messages: Final = _analyst_messages(query, candidates, self.config.max_candidate_chars)
+        model: Final = self.config.resolved_analyst_model
+        kwargs: Final = _internal_kwargs(
             request_kwargs,
             origin=FUSION_ANALYST_CALL_ORIGIN,
             model=model,
@@ -908,7 +1020,9 @@ class FusionRouter:
         kwargs.update(
             max_completion_tokens=self.config.max_completion_tokens,
             temperature=0,
-            response_format={"type": "json_object"},
+            response_format={  # mutable-ok: local provider payload
+                "type": "json_object"
+            },  # mutable-ok: local provider payload
         )
         if self.config.reasoning_effort is not None:
             kwargs["reasoning_effort"] = self.config.reasoning_effort
@@ -930,7 +1044,7 @@ class FusionRouter:
 
     async def acompletion(
         self,
-        messages: list[AllMessageValues],
+        messages: list[AllMessageValues],  # mutable-ok: SDK boundary
         stream: bool,
         request_kwargs: Mapping[str, object],
     ) -> ModelResponse | CustomStreamWrapper:
@@ -948,10 +1062,15 @@ class FusionRouter:
             )
 
         initial_response, replay_stream = await self._initial_outer_call(messages, stream, request_kwargs)
-        tool_call = _fusion_tool_call(initial_response)
-        fusion_metadata: dict[str, object] = {"invoked": False, "protocol": FUSION_PROTOCOL_VERSION}
+        tool_call: Final = _fusion_tool_call(initial_response)
+        fusion_metadata: dict[  # mutable-ok: response metadata is attached as a native mapping
+            str, object
+        ] = {  # mutable-ok: SDK boundary
+            "invoked": False,
+            "protocol": FUSION_PROTOCOL_VERSION,
+        }  # rebind-ok: orchestration branch state  # mutable-ok: SDK boundary
         if tool_call is None:
-            hidden = getattr(initial_response, "_hidden_params", None)
+            hidden = getattr(initial_response, "_hidden_params", None)  # rebind-ok: orchestration branch state
             if isinstance(hidden, dict):
                 hidden["fusion"] = fusion_metadata
             return replay_stream if replay_stream is not None else initial_response
@@ -963,9 +1082,11 @@ class FusionRouter:
             await replay_stream.aclose()
 
         fusion_metadata["invoked"] = True
-        raw_query = _fusion_query(tool_call)
+        raw_query: Final = _fusion_query(tool_call)
         if raw_query is None:
-            payload: Mapping[str, object] = {
+            payload: Mapping[
+                str, object
+            ] = {  # rebind-ok: orchestration branch state  # mutable-ok: local provider payload
                 "status": "error",
                 "error": "the Fusion tool received invalid arguments",
                 "failure_reason": "invalid_tool_arguments",
@@ -976,30 +1097,32 @@ class FusionRouter:
                 analysis_available=False,
             )
         else:
-            query = raw_query[: self.config.max_candidate_chars]
-            panel_results = await asyncio.gather(
+            query: Final = raw_query[: self.config.max_candidate_chars]
+            panel_results: Final = await asyncio.gather(
                 *(self._run_panel_member(model, query, request_kwargs) for model in self.config.panel_models)
             )
-            candidates = tuple(result.candidate for result in panel_results if isinstance(result, FusionPanelSuccess))
-            failures = tuple(result for result in panel_results if isinstance(result, FusionPanelFailure))
-            analysis = await self._analyse(query, candidates, request_kwargs) if candidates else None
-            payload = _tool_result_payload(
+            candidates: Final = tuple(
+                result.candidate for result in panel_results if isinstance(result, FusionPanelSuccess)
+            )
+            failures: Final = tuple(result for result in panel_results if isinstance(result, FusionPanelFailure))
+            analysis: Final = await self._analyse(query, candidates, request_kwargs) if candidates else None
+            payload = _tool_result_payload(  # rebind-ok: orchestration branch state
                 query,
                 candidates,
                 failures,
                 analysis,
                 self.config.max_candidate_chars,
             )
-            fusion_metadata = {
+            fusion_metadata = {  # rebind-ok: orchestration branch state  # mutable-ok: local provider payload
                 "invoked": True,
                 "protocol": FUSION_PROTOCOL_VERSION,
                 "panel_successes": len(candidates),
                 "panel_failures": len(failures),
                 "analysis_available": analysis is not None,
             }
-        final_messages = _continuation_messages(messages, tool_call, payload)
+        final_messages: Final = _continuation_messages(messages, tool_call, payload)
 
-        final_kwargs = _outer_kwargs(request_kwargs)
+        final_kwargs: Final = _outer_kwargs(request_kwargs)
         final_kwargs.pop("litellm_logging_obj", None)
         final_kwargs.pop("litellm_call_id", None)
         # `required` has already been satisfied by the private Fusion call. Do
@@ -1010,22 +1133,22 @@ class FusionRouter:
                 final_kwargs["tool_choice"] = "auto"
             else:
                 final_kwargs.pop("tool_choice", None)
-        final_metadata = _fusion_call_metadata(request_kwargs, FUSION_CONTINUATION_CALL_ORIGIN)
+        final_metadata: Final = _fusion_call_metadata(request_kwargs, FUSION_CONTINUATION_CALL_ORIGIN)
         final_kwargs.pop("litellm_metadata", None)
         final_kwargs["metadata"] = final_metadata
-        reservation = final_metadata.get(_BUDGET_RESERVATION_METADATA_KEY)
+        reservation: Final = final_metadata.get(_BUDGET_RESERVATION_METADATA_KEY)
         if isinstance(reservation, dict):
             # Cancellation accounting can now distinguish an in-flight final
             # outer call from cancellation while the private panel was running.
             reservation[FUSION_BUDGET_CONTINUATION_STARTED_KEY] = True
-        response = await self._completion(
+        response: Final = await self._completion(
             model=self.config.outer_model,
             messages=final_messages,
             stream=stream,
             _fusion_depth=1,
             **final_kwargs,
         )
-        hidden = getattr(response, "_hidden_params", None)
+        hidden = getattr(response, "_hidden_params", None)  # rebind-ok: orchestration branch state
         if isinstance(hidden, dict):
             hidden["fusion"] = fusion_metadata
         return response

litellm/proxy/hooks/proxy_track_cost_callback.py

@@ -86,11 +86,11 @@ def _mapping_or_attribute(value: object, key: str) -> object:
 
 
 def _response_invoked_fusion(response: object) -> bool:
-    choices = _mapping_or_attribute(response, "choices")
+    choices: Final = _mapping_or_attribute(response, "choices")
     if not isinstance(choices, Sequence) or isinstance(choices, (str, bytes)) or not choices:
         return False
-    message = _mapping_or_attribute(choices[0], "message")
-    tool_calls = _mapping_or_attribute(message, "tool_calls")
+    message: Final = _mapping_or_attribute(choices[0], "message")
+    tool_calls: Final = _mapping_or_attribute(message, "tool_calls")
     if not isinstance(tool_calls, Sequence) or isinstance(tool_calls, (str, bytes)):
         return False
     return any(
@@ -100,45 +100,63 @@ def _response_invoked_fusion(response: object) -> bool:
 
 
 def _should_defer_fusion_budget_reconciliation(
-    metadata: dict,
+    metadata: dict,  # mutable-ok: SDK boundary
     completion_response: object,
-    kwargs: dict,
+    kwargs: dict,  # mutable-ok: SDK boundary
 ) -> bool:
-    origin = metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY)
+    origin: Final = metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY)
     if origin in _FUSION_ALWAYS_DEFERRED_ORIGINS:
         return True
     if origin != "fusion_initial":
         return False
-    complete_stream = kwargs.get("complete_streaming_response")
+    complete_stream: Final = kwargs.get("complete_streaming_response")
     return _response_invoked_fusion(completion_response) or _response_invoked_fusion(complete_stream)
 
 
 def _accumulate_fusion_cost(
-    budget_reservation: dict,
+    budget_reservation: dict,  # mutable-ok: SDK boundary
     response_cost: float,
-    kwargs: dict,
+    kwargs: dict,  # mutable-ok: SDK boundary
 ) -> None:
     """Add one hidden call exactly once before its asynchronous DB write."""
-    call_id = kwargs.get("litellm_call_id") or kwargs.get("id")
-    seen_call_ids = budget_reservation.setdefault(FUSION_BUDGET_ACCUMULATED_CALL_IDS_KEY, [])
+    call_id: Final = kwargs.get("litellm_call_id") or kwargs.get("id")
+    seen_call_ids: Final = budget_reservation.setdefault(
+        FUSION_BUDGET_ACCUMULATED_CALL_IDS_KEY,
+        [],  # mutable-ok: local provider payload
+    )  # mutable-ok: local provider payload
     if isinstance(seen_call_ids, list) and call_id is not None:
-        normalized_call_id = str(call_id)
+        normalized_call_id: Final = str(call_id)
         if normalized_call_id in seen_call_ids:
             return
         seen_call_ids.append(normalized_call_id)
-    budget_reservation[FUSION_BUDGET_ACCUMULATED_COST_KEY] = float(
-        budget_reservation.get(FUSION_BUDGET_ACCUMULATED_COST_KEY) or 0.0
-    ) + max(response_cost, 0.0)
+    budget_reservation[  # rebind-ok: shared reservation ledger
+        FUSION_BUDGET_ACCUMULATED_COST_KEY
+    ] = (  # rebind-ok: shared reservation ledger
+        float(  # rebind-ok: shared reservation ledger
+            budget_reservation.get(FUSION_BUDGET_ACCUMULATED_COST_KEY) or 0.0
+        )
+        + max(response_cost, 0.0)
+    )
 
 
-def _failure_should_leave_fusion_reservation_open(request_data: dict) -> bool:
-    buckets: tuple[object, ...] = (
+def _failure_should_leave_fusion_reservation_open(
+    request_data: dict,  # mutable-ok: SDK boundary
+) -> bool:  # mutable-ok: SDK boundary
+    buckets: Final[tuple[object, ...]] = (
         request_data.get("metadata"),
         request_data.get("litellm_metadata"),
-        (request_data.get("litellm_params") or {}).get("metadata")
+        (
+            request_data.get("litellm_params") or {}  # mutable-ok: fallback metadata uses a native mapping
+        ).get(  # mutable-ok: local provider payload
+            "metadata"
+        )  # mutable-ok: local provider payload
         if isinstance(request_data.get("litellm_params"), dict)
         else None,
-        (request_data.get("litellm_params") or {}).get("litellm_metadata")
+        (
+            request_data.get("litellm_params") or {}  # mutable-ok: fallback metadata uses a native mapping
+        ).get(  # mutable-ok: local provider payload
+            "litellm_metadata"
+        )  # mutable-ok: local provider payload
         if isinstance(request_data.get("litellm_params"), dict)
         else None,
     )
@@ -351,7 +369,7 @@ class _ProxyDBLogger(CustomLogger):
                 router=get_llm_router(),
             )
             if response_cost is not None and kwargs.get("cache_hit", False) is True:
-                response_cost = 0.0
+                response_cost = 0.0  # rebind-ok: orchestration branch state
                 verbose_proxy_logger.debug("Cache Hit: response_cost %s, for user_id %s", response_cost, user_id)
             defer_fusion_reconciliation: Final = (
                 budget_reservation is not None

litellm/proxy/spend_tracking/budget_reservation.py

@@ -332,7 +332,11 @@ async def release_budget_reservation(budget_reservation: dict | None) -> None:
         # A Fusion request may have completed hidden provider calls before a
         # later panel/continuation failure. Preserve that known billed floor
         # instead of refunding the whole logical request to zero.
-        actual_cost=(budget_reservation or {}).get(FUSION_BUDGET_ACCUMULATED_COST_KEY, 0.0),
+        actual_cost=(
+            budget_reservation or {}  # mutable-ok: absent reservation uses an empty native mapping
+        ).get(  # mutable-ok: local provider payload
+            FUSION_BUDGET_ACCUMULATED_COST_KEY, 0.0
+        ),  # mutable-ok: local provider payload
     )
 
 
@@ -366,10 +370,10 @@ async def release_budget_reservation_on_cancel(
     # accumulator. Add another input floor only once the final continuation
     # has been dispatched; otherwise cancellation during the panel would count
     # the initial input twice.
-    hidden_call_finished = bool(budget_reservation.get(FUSION_BUDGET_ACCUMULATED_CALL_IDS_KEY)) or (
+    hidden_call_finished: Final = bool(budget_reservation.get(FUSION_BUDGET_ACCUMULATED_CALL_IDS_KEY)) or (
         accumulated_cost > 0.0
     )
-    add_in_flight_input = not hidden_call_finished or (
+    add_in_flight_input: Final = not hidden_call_finished or (
         budget_reservation.get(FUSION_BUDGET_CONTINUATION_STARTED_KEY) is True
     )
     incurred_cost: Final = accumulated_cost + (
@@ -1082,7 +1086,7 @@ def _estimate_request_model_max_cost(
     internal_call_count: Final = (
         fusion_router.config.max_tool_calls + 1 if fusion_router.config.search_tool_name is not None else 1
     )
-    internal_request_body: Final = {
+    internal_request_body: Final = {  # mutable-ok: local provider payload
         **request_body,
         # Panel and analyst output is controlled by the Fusion config, not by
         # the caller's cap on the outward response.
@@ -1099,7 +1103,7 @@ def _estimate_request_model_max_cost(
     )
 
     def estimate_internal_calls(model: str, base_input_tokens: int) -> float | None:
-        estimates = tuple(
+        estimates: Final = tuple(
             _estimate_request_max_cost_for_model(
                 request_body=internal_request_body,
                 route=route,
@@ -1111,7 +1115,7 @@ def _estimate_request_model_max_cost(
         )
         if any(estimate is None for estimate in estimates):
             return None
-        return sum(cast("tuple[float, ...]", estimates))
+        return sum(estimate for estimate in estimates if estimate is not None)
 
     panel_estimates: Final = tuple(
         estimate_internal_calls(panel_model, query_token_ceiling) for panel_model in fusion_router.config.panel_models
@@ -1134,7 +1138,7 @@ def _estimate_request_model_max_cost(
         if original_outer_tokens is not None
         else None
     )
-    analyst_estimate = estimate_internal_calls(fusion_router.config.resolved_analyst_model, analyst_input_tokens)
+    analyst_estimate: Final = estimate_internal_calls(fusion_router.config.resolved_analyst_model, analyst_input_tokens)
     final_outer_estimate: Final = _estimate_request_max_cost_for_model(
         request_body=request_body,
         route=route,
@@ -1151,7 +1155,7 @@ def _estimate_request_model_max_cost(
         # matches the normal unknown-price behavior instead of presenting an
         # under-estimate as a valid worst case.
         return None
-    return sum(cast("tuple[float, ...]", child_estimates))
+    return sum(estimate for estimate in child_estimates if estimate is not None)
 
 
 def estimate_request_input_cost(

litellm/router.py

@@ -9464,7 +9464,11 @@ class Router:
         )
 
     async def _fusion_asearch(  # kwargs-ok: bridge preserves the Router.asearch keyword surface
-        self, *, model: str, query: str, **kwargs: object
+        self,
+        *,
+        model: str,
+        query: str,
+        **kwargs: object,  # kwargs-ok: SDK passthrough
     ) -> object:
         """Late-bound Search API bridge with the originating caller's permissions."""
         metadata_values: Final = tuple(kwargs.get(key) for key in ("litellm_metadata", "metadata"))
@@ -9486,7 +9490,7 @@ class Router:
             )
 
             try:
-                user_api_key_auth = (
+                user_api_key_auth: Final = (
                     raw_user_api_key_auth
                     if isinstance(raw_user_api_key_auth, UserAPIKeyAuth)
                     else UserAPIKeyAuth.model_validate(raw_user_api_key_auth)
@@ -9502,7 +9506,11 @@ class Router:
                 search_tool_name=model,
                 user_api_key_dict=user_api_key_auth,
             )
-        return await self.asearch(model=model, query=query, **kwargs)
+        return await self.asearch(
+            model=model,
+            query=query,
+            **kwargs,  # pyright: ignore[reportArgumentType]  # bridge forwards provider-specific search kwargs
+        )
 
     def deployment_is_active_for_environment(self, deployment: Deployment) -> bool:
         """

tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py

@@ -688,9 +688,9 @@ async def test_fusion_hidden_costs_accumulate_then_continuation_reconciles_once(
         }
 
     with (
-        patch("litellm.proxy.proxy_server.increment_spend_counters", new_callable=AsyncMock) as increment,
-        patch("litellm.proxy.proxy_server.update_cache", new_callable=AsyncMock),
-        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,
+        patch("litellm.proxy.proxy_server.increment_spend_counters", new_callable=AsyncMock) as increment,  # test-quality-ok: isolates proxy persistence while reservation state remains observable
+        patch("litellm.proxy.proxy_server.update_cache", new_callable=AsyncMock),  # test-quality-ok: isolates proxy persistence while reservation state remains observable
+        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,  # test-quality-ok: injects the callback persistence boundary
     ):
         proxy_logging.db_spend_update_writer.update_database = AsyncMock()
         proxy_logging.slack_alerting_instance.customer_spend_alert = AsyncMock()
@@ -762,9 +762,9 @@ async def test_cached_fusion_hidden_call_accumulates_zero_cost():
     }
 
     with (
-        patch("litellm.proxy.proxy_server.increment_spend_counters", new_callable=AsyncMock) as increment,
-        patch("litellm.proxy.proxy_server.update_cache", new_callable=AsyncMock),
-        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,
+        patch("litellm.proxy.proxy_server.increment_spend_counters", new_callable=AsyncMock) as increment,  # test-quality-ok: isolates proxy persistence while reservation state remains observable
+        patch("litellm.proxy.proxy_server.update_cache", new_callable=AsyncMock),  # test-quality-ok: isolates proxy persistence while reservation state remains observable
+        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,  # test-quality-ok: injects the callback persistence boundary
     ):
         proxy_logging.db_spend_update_writer.update_database = AsyncMock()
         proxy_logging.slack_alerting_instance.customer_spend_alert = AsyncMock()
@@ -810,8 +810,8 @@ async def test_unpriced_fusion_hidden_call_does_not_release_parent_reservation()
     }
 
     with (
-        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,
-        patch(
+        patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging,  # test-quality-ok: injects the callback alert boundary
+        patch(  # test-quality-ok: verifies unpriced hidden calls cannot release the parent reservation
             "litellm.proxy.spend_tracking.budget_reservation.release_budget_reservation",
             new_callable=AsyncMock,
         ) as release_reservation,

tests/test_litellm/proxy/test_budget_reservation.py

@@ -87,7 +87,7 @@ def _request_body() -> dict:
 
 
 async def _reserve(valid_token, cost, key_cache, proxy_logging_obj):
-    with patch(
+    with patch(  # test-quality-ok: isolates child pricing to exercise partial-estimate refusal
         "litellm.proxy.spend_tracking.budget_reservation.estimate_request_max_cost",
         return_value=cost,
     ):
@@ -2902,7 +2902,7 @@ async def test_release_budget_reservation_on_cancel_swallows_release_errors():
 
 
 @pytest.mark.asyncio
-async def test_fusion_release_and_cancel_keep_already_billed_hidden_costs():
+async def test_fusion_release_and_cancel_keep_already_billed_hidden_costs():  # test-quality-ok: forwarded cost is the helper contract
     reservation = {
         "reserved_cost": 3.0,
         "entries": [],
@@ -2910,7 +2910,7 @@ async def test_fusion_release_and_cancel_keep_already_billed_hidden_costs():
         "input_cost": 0.5,
         FUSION_BUDGET_ACCUMULATED_COST_KEY: 0.3,
     }
-    with patch(
+    with patch(  # test-quality-ok: captures reconciliation to verify each billed cost floor
         "litellm.proxy.spend_tracking.budget_reservation.reconcile_budget_reservation",
         new=AsyncMock(),
     ) as reconcile: