chore(lint): remove the LIT002 mutable-construction rule

AGENTS.md

@@ -62,7 +62,7 @@ Never edit or commit `ruff-strict-budget.json`, `type-discipline-budget.json`, `
 
 If you're trying to create a new function that relies on untyped stuff, instead of adding more Any's and pushing `reportAny` / `reportExplicitAny` closer to their basedpyright ceilings, just validate it in the caller with Pydantic (a model or `TypeAdapter` that returns the typed thing or raises will do) and then pass the now typed variable in
 
-If you get an LIT001 or LIT002 fail, refactor the code to follow functional programming best practices rather than introducing mutable data structures. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `MappingProxyType()` / `frozenset()` instead of seeding an empty `list`/`dict`/`set` and mutating it over time. Ideally, `# mutable-ok` is never used; reach for it only as a genuine last resort when an immutable rewrite is truly impossible, and always pair it with a real reason
+If you get an LIT001 fail, refactor the code to follow functional programming best practices rather than introducing mutable data structures. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `MappingProxyType()` / `frozenset()` instead of seeding an empty `list`/`dict`/`set` and mutating it over time. Ideally, `# mutable-ok` is never used; reach for it only as a genuine last resort when an immutable rewrite is truly impossible, and always pair it with a real reason
 
 Every lint or type suppression must name the exact rule inside brackets and carry a reason comment, e.g. `# pyright: ignore[reportArgumentType]  # stubs lack async overload` or `# noqa: TID251  # <reason>`. `# type: ignore` is banned (LIT009): pyrightconfig.json sets `enableTypeIgnoreComments` to false, so it silently does nothing
 

enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py

@@ -616,7 +616,7 @@ def _redact_prompt(self, data: dict) -> int:
             data["prompt"] = self.redact_text(prompt, source="prompt")
             return 1
         if isinstance(prompt, list):
-            data["prompt"] = [  # mutable-ok: data["prompt"] is a list on the wire
+            data["prompt"] = [
                 self.redact_text(item, source="prompt")
                 if isinstance(item, str) and item
                 else item

litellm/__init__.py

@@ -663,7 +663,7 @@ def identify(event_details):
 azure_text_models: Set = set()
 anyscale_models: Set = set()
 cerebras_models: Set = set()
-nadir_models: Set = set()  # mutable-ok: provider registry, filled from model_cost at import like every sibling provider
+nadir_models: Set = set()
 galadriel_models: Set = set()
 nvidia_nim_models: Set = set()
 nvidia_riva_models: Set = set()
@@ -697,7 +697,7 @@ def identify(event_details):
 cometapi_models: Set = set()
 oci_models: Set = set()
 vercel_ai_gateway_models: Set = set()
-edenai_models: Set = set()  # mutable-ok: filled from the price map at import, like the sibling provider sets
+edenai_models: Set = set()
 volcengine_models: Set = set()
 wandb_models: Set = set(WANDB_MODELS)
 ovhcloud_models: Set = set()

litellm/_logging.py

@@ -352,13 +352,9 @@ def _replace_string_leaves(value: object, values: Iterator[str]) -> object:
     if isinstance(value, str):
         return next(values)
     if isinstance(value, dict):
-        return {  # mutable-ok: LogRecord extras must keep JSON dict shape for handlers
-            key: _replace_string_leaves(child, values) for key, child in value.items()
-        }
+        return {key: _replace_string_leaves(child, values) for key, child in value.items()}
     if isinstance(value, list):
-        return [  # mutable-ok: LogRecord extras must keep JSON list shape for handlers
-            _replace_string_leaves(child, values) for child in value
-        ]
+        return [_replace_string_leaves(child, values) for child in value]
     if isinstance(value, tuple):
         return tuple(_replace_string_leaves(child, values) for child in value)
     return value
@@ -368,13 +364,9 @@ def _sort_processed_sets(original: object, processed: object) -> object:
     if isinstance(original, set) and isinstance(processed, list):
         return sorted(processed)
     if isinstance(original, dict) and isinstance(processed, dict):
-        return {  # mutable-ok: sorting nested sets must preserve the surrounding JSON dict
-            key: _sort_processed_sets(original.get(key), value) for key, value in processed.items()
-        }
+        return {key: _sort_processed_sets(original.get(key), value) for key, value in processed.items()}
     if isinstance(original, list) and isinstance(processed, list):
-        return [  # mutable-ok: sorting nested sets must preserve the surrounding JSON list
-            _sort_processed_sets(before, after) for before, after in zip(original, processed)
-        ]
+        return [_sort_processed_sets(before, after) for before, after in zip(original, processed)]
     if isinstance(original, tuple) and isinstance(processed, tuple):
         return tuple(_sort_processed_sets(before, after) for before, after in zip(original, processed))
     return processed

litellm/_redis.py

@@ -233,7 +233,7 @@ def _coerce_redis_kwargs_types(
             "socket_keepalive": bool,
         }
     )
-    result: Final = dict(redis_kwargs)  # mutable-ok: per-key try/except coercion below needs to drop individual keys
+    result: Final = dict(redis_kwargs)
     for key, value in redis_kwargs.items():
         if not isinstance(value, str):
             continue
@@ -803,7 +803,7 @@ def _credential_provider_auth_kwargs(redis_kwargs: dict) -> dict:
 
     superseded: Final = frozenset({"redis_connect_func", "username", "password"})
     kept: Final = ((k, v) for k, v in redis_kwargs.items() if k not in superseded)
-    return dict(kept, credential_provider=credential_provider)  # mutable-ok: the branches below mutate these kwargs
+    return dict(kept, credential_provider=credential_provider)
 
 
 def get_redis_client(**env_overrides):

litellm/a2a_protocol/main.py

@@ -145,7 +145,7 @@ def _a2a_cost_params(litellm_params: Mapping[str, object] | None) -> Mapping[str
 
 
 def _card_http_kwargs(extra_headers: dict[str, str] | None) -> dict[str, object] | None:
-    return {"headers": extra_headers} if extra_headers else None  # mutable-ok: a2a-sdk's get_agent_card takes a dict
+    return {"headers": extra_headers} if extra_headers else None
 
 
 def _agent_card_path(litellm_params: Mapping[str, object]) -> str | None:
@@ -612,7 +612,7 @@ def _build_streaming_logging_obj(
         logging_obj.model_call_details["agent_id"] = agent_id
 
     _request_context: Final = (("metadata", metadata), ("proxy_server_request", proxy_server_request))
-    _litellm_params: Final = dict(  # mutable-ok: Logging.litellm_params is declared as a dict
+    _litellm_params: Final = dict(
         (*_a2a_cost_params(litellm_params).items(), *((key, value) for key, value in _request_context if value))
     )
 

litellm/batches/batch_utils.py

@@ -127,7 +127,7 @@ async def _handle_completed_batch(
         return BatchCostUsageResult(
             cost=0.0,
             usage=Usage(prompt_tokens=0, completion_tokens=0, total_tokens=0),
-            models=[],  # mutable-ok: no output file means no model was ever priced; BatchCostUsageResult.models requires list[str]
+            models=[],
             successful_requests=0,
             failed_requests=await count_error_file_failed_requests(
                 batch, custom_llm_provider=custom_llm_provider, litellm_params=litellm_params

litellm/caching/affinity_cache.py

@@ -103,10 +103,10 @@ async def claim_affinity_pin(
         try:
             claim_script: Final = redis_cache.async_register_script(_CLAIM_PIN_SCRIPT)
             args: Final = (
-                json.dumps(dict(pin_value)),  # mutable-ok: JSON serialization requires dict, not a generic Mapping
+                json.dumps(dict(pin_value)),
                 int(ttl_seconds),
                 *(
-                    (json.dumps(tuple(dict(value) for value in eligible_values)),)  # mutable-ok: JSON requires dict
+                    (json.dumps(tuple(dict(value) for value in eligible_values)),)
                     if eligible_values is not None
                     else ()
                 ),

litellm/caching/caching_handler.py

@@ -702,14 +702,14 @@ def _combine_cached_embedding_response_with_api_result(
         )
         merged: Final = EmbeddingResponse(
             model=cached.model,
-            data=[  # mutable-ok: EmbeddingResponse.data is a pydantic list field
+            data=[
                 item
                 if item is not None
                 else Embedding(embedding=next(fresh_items)["embedding"], index=position, object="embedding")
                 for position, item in enumerate(cached.data)
             ],
             usage=merged_usage,
-            hidden_params={  # mutable-ok: EmbeddingResponse._hidden_params is a mutable dict field
+            hidden_params={
                 **cached._hidden_params,
                 "cache_hit": True,
             },

litellm/caching/dual_cache.py

@@ -252,9 +252,7 @@ def batch_get_cache(
                     if value is not None:
                         self.in_memory_cache.set_cache(key, value, **self._backfill_kwargs(kwargs))
 
-            return list(  # mutable-ok: public list contract
-                redis_result.get(key) if value is None else value for key, value in zip(keys, result)
-            )
+            return list(redis_result.get(key) if value is None else value for key, value in zip(keys, result))
         except Exception as e:
             log_redis_failure(
                 verbose_logger, logging.ERROR, "LiteLLM Cache: exception in batch_get_cache", e, with_traceback=True
@@ -329,8 +327,8 @@ def _reserve_redis_batch_keys(
     def reserve_redis_batch_reads(self, keys: Sequence[str]) -> tuple[list[str], dict[str, float | None]]:
         """Reserve the memory-missed keys whose throttled Redis reads are due, as a batch read would."""
         if self.redis_cache is None:
-            return [], {}  # mutable-ok: API contract returns an empty list and dictionary
-        key_list: Final = list(keys)  # mutable-ok: batch_get_cache takes a list
+            return [], {}
+        key_list: Final = list(keys)
         memory: Final = self.in_memory_cache
         in_memory_result: Final = (
             None
@@ -386,7 +384,7 @@ async def _apply_batch_get(
 
     async def declare_batch_get(self, keys: Sequence[str], batch: RedisBatch) -> DeclaredBatchRead:
         pending: Final = await self._prepare_batch_get(
-            list(keys),  # mutable-ok: the shared batch read takes a list
+            list(keys),
             local_only=False,
             throttle_redis=False,
         )
@@ -627,7 +625,7 @@ async def async_increment_cache_pipeline_post_call(
         parent_otel_span: Span | None = None,
     ) -> None:
         batch: Final = None if self.redis_cache is None else active_post_call_redis_batch(self.redis_cache)
-        operations: Final = list(increment_list)  # mutable-ok: both increment pipelines take a list
+        operations: Final = list(increment_list)
         if batch is None:
             await self.async_increment_cache_pipeline(operations, parent_otel_span=parent_otel_span)
             return

litellm/caching/evicted_client_closer.py

@@ -238,7 +238,7 @@ def _enqueue(self, pending: _PendingClose) -> None:
         the front rather than having to be searched for.
         """
         with self._queue_lock:
-            bucket: Final = self._buckets.setdefault(_bucket_key(pending), deque())  # mutable-ok: FIFO by design
+            bucket: Final = self._buckets.setdefault(_bucket_key(pending), deque())
             while bucket and bucket[0].client_ref() is None:
                 bucket.popleft()
                 self._pending_count -= 1

litellm/caching/redis_batch.py

@@ -33,7 +33,7 @@
 
 _T = TypeVar("_T")
 _ScriptArg = str | bytes | int | float
-SettledHook = Callable[[asyncio.Future[_T]], Awaitable[None] | None]  # mutable-ok: Callable params
+SettledHook = Callable[[asyncio.Future[_T]], Awaitable[None] | None]
 POST_CALL_FLUSH_DEADLINE_SECONDS: Final = 1.0
 
 
@@ -139,7 +139,7 @@ def resolve(self, replies: Sequence[object]) -> Mapping[str, object]:
         )
 
     async def run_alone(self) -> Mapping[str, object]:
-        found: Mapping[str, object] = await self._redis_cache.async_batch_get_cache(key_list=list(self._keys))  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # untyped cache API  # mutable-ok: the cache API takes a list
+        found: Mapping[str, object] = await self._redis_cache.async_batch_get_cache(key_list=list(self._keys))  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # untyped cache API
         if any(key not in found for key in self._keys):
             raise ConnectionError("batch get did not return every key")
         return found

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -123,13 +123,13 @@ def _reasoning_input_items(msg: "AllMessageValues") -> list[dict[str, object]]:
     blocks are the fallback for turns that arrived over another API surface.
     """
     items: Final = _get_reasoning_items(msg)
-    stored: Final = [_reasoning_item_to_response_input(item) for item in items]  # mutable-ok: API message payload
+    stored: Final = [_reasoning_item_to_response_input(item) for item in items]
     if stored:
         return stored
     raw_blocks: Final = msg.get("thinking_blocks") or ()
     blocks: Final = cast("Iterable[ChatCompletionThinkingBlock]", raw_blocks)  # cast-ok: untyped client json
     replayed: Final = responses_reasoning_items_from_thinking_blocks(blocks)
-    return [dict(item) for item in replayed]  # mutable-ok: API message payload
+    return [dict(item) for item in replayed]
 
 
 def _build_reasoning_item(
@@ -441,7 +441,7 @@ def convert_chat_completion_messages_to_responses_api(
                 input_items.extend(_reasoning_input_items(msg))
                 if content:
                     input_items.append(
-                        {  # mutable-ok: API message payload
+                        {
                             "type": "message",
                             "role": "assistant",
                             "content": self._convert_content_to_responses_format(content, "assistant"),
@@ -475,7 +475,7 @@ def convert_chat_completion_messages_to_responses_api(
                 if role == "assistant":
                     input_items.extend(_reasoning_input_items(msg))
                 input_items.append(
-                    {  # mutable-ok: API message payload
+                    {
                         "type": "message",
                         "role": role,
                         "content": self._convert_content_to_responses_format(content, cast(str, role)),
@@ -531,11 +531,11 @@ def _merge_text(
     ) -> "ResponseText":
         existing: Final = cast(  # cast-ok: text field is a ResponseText | dict[str, Any] | None union
             "dict[str, object]",
-            dict(responses_api_request).get("text") or {},  # mutable-ok: one-shot merge seed
+            dict(responses_api_request).get("text") or {},
         )
         return cast(  # cast-ok: merged mapping is a valid ResponseText shape
             "ResponseText",
-            {**existing, **update},  # mutable-ok: one-shot merged payload
+            {**existing, **update},
         )
 
     def _build_sanitized_litellm_params(self, litellm_params: dict) -> dict[str, object]:
@@ -1506,7 +1506,7 @@ def translate_responses_chunk_to_openai_stream(
                     # tool call; per-stream callers already received it via
                     # output_item.added and the argument delta events
                     return ModelResponseStream(
-                        choices=[  # mutable-ok: ModelResponseStream coerces only list choices
+                        choices=[
                             StreamingChoices(
                                 index=0,
                                 delta=Delta(
@@ -1612,7 +1612,7 @@ def translate_responses_chunk_to_openai_stream(
                     )
                 ],
                 usage=usage,
-                provider_specific_fields=dict(provider_metadata) or None,  # mutable-ok: field is typed dict
+                provider_specific_fields=dict(provider_metadata) or None,
                 **(
                     MappingProxyType({"service_tier": served_service_tier})
                     if isinstance(served_service_tier, str)

litellm/cost_calculator.py

@@ -2874,9 +2874,7 @@ def collect_and_combine_usage_from_responses_ws_results(
         collected_usage_objects: Final = ResponsesWebSocketTokenUsageProcessor.collect_usage_from_responses_ws_results(
             results
         )
-        return ResponsesWebSocketTokenUsageProcessor.combine_usage_objects(
-            list(collected_usage_objects)  # mutable-ok: combine_usage_objects requires a list parameter
-        )
+        return ResponsesWebSocketTokenUsageProcessor.combine_usage_objects(list(collected_usage_objects))
 
 
 _TRANSCRIPTION_COMPLETED_EVENT_TYPE: Final = "conversation.item.input_audio_transcription.completed"

litellm/experimental_mcp_client/client.py

@@ -1136,7 +1136,7 @@ async def list_resource_templates(self, *, raise_on_error: bool = False) -> list
         async def _list_resource_templates_operation(session: ClientSession) -> ListResourceTemplatesResult:
             capabilities: Final = session.server_capabilities
             if capabilities is not None and capabilities.resources is None:
-                return ListResourceTemplatesResult(resource_templates=[])  # mutable-ok: MCP result payload
+                return ListResourceTemplatesResult(resource_templates=[])
             try:
                 return ListResourceTemplatesResult(
                     resource_templates=await self._list_optional_pages(
@@ -1150,7 +1150,7 @@ async def _list_resource_templates_operation(session: ClientSession) -> ListReso
                 verbose_logger.debug(
                     "MCP client list_resource_templates is unsupported by %s: %s", self.server_url or "stdio", error
                 )
-                return ListResourceTemplatesResult(resource_templates=[])  # mutable-ok: MCP result payload
+                return ListResourceTemplatesResult(resource_templates=[])
 
         try:
             result: Final = await self.run_with_session(_list_resource_templates_operation)

litellm/experimental_mcp_client/tools.py

@@ -171,9 +171,7 @@ async def load_mcp_tools(
     """
     tools: Final = await list_tools_with_pagination(session)
     if format == "openai":
-        return [  # mutable-ok: public API returns a list
-            transform_mcp_tool_to_openai_tool(mcp_tool=tool) for tool in tools
-        ]
+        return [transform_mcp_tool_to_openai_tool(mcp_tool=tool) for tool in tools]
     return tools
 
 

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -1131,7 +1131,7 @@ async def send_model_deprecation_alert(
             message=message,
             level=level,
             alert_type=AlertType.model_deprecation_warnings,
-            alerting_metadata={  # mutable-ok: send_alert takes a dict payload
+            alerting_metadata={
                 "deprecated_count": len(snapshot.deprecated),
                 "imminent_count": len(snapshot.imminent),
                 "upcoming_count": len(snapshot.upcoming),
@@ -1245,8 +1245,8 @@ async def _construct_user_invitation_link(self, recipient_user_id: str | None, b
         try:
             existing_invitations: Final = TypeAdapter(list[InvitationModel]).validate_python(
                 await InvitationLinkRepository(prisma_client).table.find_many(  # pyright: ignore[reportAny]  # untyped prisma boundary (any-ok), result validated by TypeAdapter
-                    where={"user_id": recipient_user_id},  # mutable-ok: prisma find_many requires a dict where filter
-                    order={"created_at": "desc"},  # mutable-ok: prisma find_many requires a dict order arg
+                    where={"user_id": recipient_user_id},
+                    order={"created_at": "desc"},
                 ),
                 from_attributes=True,
             )
@@ -2011,7 +2011,7 @@ async def send_user_spend_alerts(self, prisma_client: "PrismaClient | None" = No
                     message="\n\n".join(event.message for event in typed_events),
                     level="High",
                     alert_type=alert_type,
-                    alerting_metadata={},  # mutable-ok: send_alert takes a dict payload
+                    alerting_metadata={},
                 )
                 for event in typed_events:
                     await self.internal_usage_cache.async_set_cache(