refactor(lint): allow dict and mapping annotations in LIT001

AGENTS.md

@@ -62,7 +62,7 @@ The four lint gates (`scripts/ruff_strict_gate.py`, `scripts/type_discipline_gat
 
 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 cap in `ANY_CAPS`, 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 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 sequences or sets. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `frozenset()` instead of seeding an empty `list`/`set` and appending to it over time. Ideally, `# mutable-ok` is never used; reach for it only as a last resort when an immutable rewrite is impossible, and always pair it with a real reason. Plain `dict` is allowed: most of the Python ecosystem takes and returns dicts, and converting to `MappingProxyType` at every boundary costs more than it protects. Never deep copy a value just to hand out an immutable view; a defensive copy of a self-referential or large object is worse than the mutation it guards against
 
 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
 
@@ -93,7 +93,7 @@ Follow these coding conventions for new/updated code (a three-line fix in a lega
 - Composition over inheritance
 - Never-nester: early returns over deep nesting
 - Don't throw; model failures as values (One function (e.g., raise_public) maps error union to existing public exception contracts via exhaustive match + assert_never)
-- No mutation; don't reassign variables, global or local. Instead of mutable lists and dicts, prefer tuples, frozen dataclasses (with slots=True), `MappingProxyType`, etc.
+- No mutation; don't reassign variables, global or local. Prefer tuples over lists and frozensets over sets. For structured records prefer frozen dataclasses (with slots=True) or `ReadOnly` TypedDicts over dicts. A `dict` is fine for a mapping with arbitrary keys, especially at a library boundary; `MappingProxyType` is optional there, and never worth a copy
   - Annotate every variable with `: Final` (LIT010). Unpacking and walrus targets cannot carry the annotation, so they are implicitly final. Don't rebind them. Never rebind or mutate function parameters (LIT011); `self`/`cls` attribute stores are the exception. If rebinding or in-place mutation is truly unavoidable, suppress with `# rebind-ok: <reason>`
   - Qualify every TypedDict field with `ReadOnly[...]` (LIT012), which nests freely with `Required` / `NotRequired` / `Annotated` in any order. If making the key writable is truly unavoidable, suppress with `# writable-ok: <reason>`
   - Comprehensions take at most one `for` clause and one `if` clause (LIT014); split stacked clauses into a helper generator, a named intermediate, or a plain loop. Suppress with `# comprehension-ok: <reason>` only when unavoidable

enterprise/litellm_enterprise/proxy/hooks/managed_files.py

@@ -1098,13 +1098,13 @@ async def get_unified_file_ids_for_provider_file_ids(
         if owner_filter is None:
             return MappingProxyType({})
 
-        provider_file_ids_list: Final = [  # mutable-ok: Prisma hasSome requires a list
+        provider_file_ids_list: Final = [
             provider_file_id for provider_file_id in unique_provider_file_ids
         ]
         rows: Final = await _managed_file_table(self.prisma_client).find_many(
-            where={  # mutable-ok: Prisma requires a plain dictionary for where
+            where={
                 **owner_filter,
-                "flat_model_file_ids": {  # mutable-ok: Prisma requires a plain filter dictionary
+                "flat_model_file_ids": {
                     "hasSome": provider_file_ids_list,
                 },
             }

litellm-proxy-extras/litellm_proxy_extras/utils.py

@@ -1158,7 +1158,7 @@ def _run_database_v2(
 
                 if next_budget.attempts_left < budget.attempts_left:
                     time.sleep(random.randrange(5, 15))
-                budget = next_budget  # rebind-ok: the loop carries the budget from one migrate deploy pass to the next
+                budget = next_budget
 
             raise RuntimeError(
                 f"Database migration failed after {MAX_MIGRATE_DEPLOY_ATTEMPTS} "

litellm/_redis.py

@@ -205,7 +205,7 @@ def _str_to_bool(value: str) -> bool:
 def _coerce_redis_kwargs_types(
     redis_kwargs: Mapping[str, object],
     client: type | tuple[type, ...] = redis.Redis,
-) -> dict[str, object]:  # mutable-ok: a caller mutates the returned kwargs before constructing its client
+) -> dict[str, object]:
     """Coerces string values to the numeric/boolean type ``client``'s constructor
     declares for that parameter. ``client`` may be a tuple of client classes; a
     parameter's type is taken from the first signature that declares it, which

litellm/caching/redis_batch.py

@@ -466,13 +466,13 @@ class RequestRedisBatches:
     )
 
     def __init__(self, post_call_deadline: float = POST_CALL_FLUSH_DEADLINE_SECONDS) -> None:
-        self._batches: Final[dict[object, RedisBatch]] = {}  # mutable-ok: lazily filled per backend
-        self._post_call: Final[dict[object, RedisBatch]] = {}  # mutable-ok: lazily filled per backend
+        self._batches: Final[dict[object, RedisBatch]] = {}
+        self._post_call: Final[dict[object, RedisBatch]] = {}
         self.post_call_deadline: Final = post_call_deadline
         self._deadline: asyncio.TimerHandle | None = None
         self._deadline_flush: asyncio.Task[None] | None = None
         # Reads declared early for a consumer that runs later in the request, keyed by consumer name.
-        self.prefetched: Final[dict[str, object]] = {}  # mutable-ok: armed pre-admission, taken at use
+        self.prefetched: Final[dict[str, object]] = {}
 
     def batch(self, redis_cache: RedisCache) -> RedisBatch:
         key: Final = _backend_key(redis_cache)

litellm/caching/redis_cluster_node_isolation.py

@@ -142,9 +142,7 @@ def __init__(
                 self._litellm_reinit_requests = 0
                 self._litellm_tolerated_timeouts = 0
                 super().__init__(*args, **kwargs)
-                self._litellm_consecutive_timeouts: dict[  # mutable-ok: per-node counter updated on the command hot path
-                    str, int
-                ] = {}
+                self._litellm_consecutive_timeouts: dict[str, int] = {}
 
             @property
             def _initialize(self) -> bool:

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -1554,7 +1554,7 @@ def __init__(
         super().__init__(streaming_response, sync_stream, json_mode)
         self._chat_completion_id: str | None = None
         self._served_service_tier: str | None = None
-        self._tool_call_index_map: dict[int, int] = {}  # mutable-ok: per-stream accumulator state
+        self._tool_call_index_map: dict[int, int] = {}
 
     def _handle_string_chunk(
         self, str_line: Union[str, "BaseModel"]
@@ -1575,7 +1575,7 @@ def _handle_string_chunk(
 
     @staticmethod
     def _sequential_tool_call_index(
-        tool_call_index_map: dict[int, int] | None,  # mutable-ok: per-stream state, remapped in place
+        tool_call_index_map: dict[int, int] | None,
         output_index: int,
     ) -> int:
         """Chat-completions tool_call indices must be 0-based and sequential, but
@@ -1594,7 +1594,7 @@ def _sequential_tool_call_index(
     @staticmethod
     def translate_responses_chunk_to_openai_stream(
         parsed_chunk: dict | BaseModel,
-        tool_call_index_map: dict[int, int] | None = None,  # mutable-ok: per-stream state, remapped in place
+        tool_call_index_map: dict[int, int] | None = None,
     ) -> "ModelResponseStream":
         """
         Translate a Responses API streaming chunk to OpenAI chat completion streaming format.

litellm/harness/context.py

@@ -15,9 +15,7 @@
 if TYPE_CHECKING:
     from litellm.harness.endpoint import ModelEndpoint
 
-ApprovalHandler: TypeAlias = Callable[
-    [Approval], bool | Awaitable[bool]  # mutable-ok: Callable parameter list in a type alias, not a runtime collection
-]
+ApprovalHandler: TypeAlias = Callable[[Approval], bool | Awaitable[bool]]
 
 
 @dataclass(frozen=True)

litellm/harness/endpoint.py

@@ -303,7 +303,7 @@ def gateway_headers(
         for name, value in incoming.items()
         if name.lower() not in DROPPED_REQUEST_HEADERS and not name.lower().startswith("x-litellm-")
     )
-    metadata_json = json.dumps(dict(metadata), default=str) if metadata else None  # mutable-ok: for json.dumps
+    metadata_json = json.dumps(dict(metadata), default=str) if metadata else None
     metadata_header = (("x-litellm-spend-logs-metadata", metadata_json),) if metadata_json is not None else ()
     added = (
         ("authorization", f"Bearer {gateway.api_key}"),
@@ -333,15 +333,15 @@ def error_status(exc: BaseException) -> int:
     return 500
 
 
-def error_body(exc: BaseException, message: str) -> dict[str, dict[str, str]]:  # mutable-ok: JSONResponse body
-    return {"error": {"type": type(exc).__name__, "message": message}}  # mutable-ok: JSONResponse body
+def error_body(exc: BaseException, message: str) -> dict[str, dict[str, str]]:
+    return {"error": {"type": type(exc).__name__, "message": message}}
 
 
 def to_jsonable(obj: object) -> object:
     if hasattr(obj, "model_dump"):
         return obj.model_dump(mode="json", exclude_none=True)
     if isinstance(obj, Mapping):
-        return dict(obj)  # mutable-ok: plain-dict copy so json.dumps can serialize any Mapping
+        return dict(obj)
     return obj
 
 
@@ -457,9 +457,7 @@ def _gateway_client(self) -> httpx.AsyncClient:
             return self._injected_client
         handler = get_async_httpx_client(
             llm_provider=httpxSpecialProvider.AgentHarness,
-            params={  # mutable-ok: get_async_httpx_client takes a dict params argument
-                "timeout": HARNESS_ENDPOINT_REQUEST_TIMEOUT_SECONDS
-            },
+            params={"timeout": HARNESS_ENDPOINT_REQUEST_TIMEOUT_SECONDS},
         )
         return handler.client
 
@@ -494,17 +492,12 @@ def _build_app(self, deps: _ServerDeps) -> Starlette:
             Route(
                 f"{prefix}/{route}",
                 self._handle,
-                methods=["POST"],  # mutable-ok: Starlette Route takes a methods list
+                methods=["POST"],
             )
             for prefix, route in itertools.product(ROUTE_PREFIXES, POST_ROUTES)
         )
-        get_routes = tuple(
-            Route(f"{prefix}/models", self._models, methods=["GET"])  # mutable-ok: Starlette Route takes a methods list
-            for prefix in ROUTE_PREFIXES
-        )
-        return deps.applications.Starlette(
-            routes=[*post_routes, *get_routes]  # mutable-ok: Starlette takes a routes list
-        )
+        get_routes = tuple(Route(f"{prefix}/models", self._models, methods=["GET"]) for prefix in ROUTE_PREFIXES)
+        return deps.applications.Starlette(routes=[*post_routes, *get_routes])
 
     @property
     def _responses(self) -> _ResponsesModule:
@@ -521,7 +514,7 @@ def _json(self, body: object, status_code: int = 200) -> Response:
 
     def _unauthorized(self) -> Response:
         return self._json(
-            {"error": {"type": "authentication_error", "message": "invalid token"}},  # mutable-ok: JSONResponse body
+            {"error": {"type": "authentication_error", "message": "invalid token"}},
             401,
         )
 
@@ -536,9 +529,9 @@ def _secrets(self) -> tuple[str | None, ...]:
     async def _models(self, request: Request) -> Response:
         if not self._authorized(request):
             return self._unauthorized()
-        entry = {"id": self.model, "object": "model", "created": 0, "owned_by": "litellm"}  # mutable-ok: JSON body
+        entry = {"id": self.model, "object": "model", "created": 0, "owned_by": "litellm"}
         data = (entry,) if self.model else ()
-        return self._json({"object": "list", "data": data})  # mutable-ok: JSON response body for Starlette JSONResponse
+        return self._json({"object": "list", "data": data})
 
     async def _handle(self, request: Request) -> Response:
         if not self._authorized(request):
@@ -574,7 +567,7 @@ async def _forward(self, request: Request, route: str, body: Mapping[str, object
         if self._client is None or self.gateway is None:
             raise HarnessError("gateway client is not started")
         if self.model:
-            body = {**body, "model": self.model}  # mutable-ok: JSON request body re-sent upstream via httpx json=
+            body = {**body, "model": self.model}
         upstream_request = self._client.build_request(
             "POST",
             f"{self.gateway.api_base}/v1/{route}",
@@ -626,10 +619,8 @@ def _record_relayed(
                 tokens = (0, 0)
         self._record(model, tokens[0], tokens[1], header_cost(upstream.headers))
 
-    def _sdk_kwargs(
-        self, body: Mapping[str, object]
-    ) -> dict[str, object]:  # mutable-ok: SDK call kwargs, mutated by _invoke_sdk then splatted
-        kwargs: dict[str, object] = {**body}  # mutable-ok: SDK call kwargs built from the JSON body, then overridden
+    def _sdk_kwargs(self, body: Mapping[str, object]) -> dict[str, object]:
+        kwargs: dict[str, object] = {**body}
         if self.model:
             kwargs["model"] = self.model
         if self.api_key:
@@ -641,14 +632,14 @@ def _sdk_kwargs(
     async def _invoke_sdk(
         self,
         route: str,
-        kwargs: dict[str, Any],  # mutable-ok: injects stream_options into the SDK kwargs
+        kwargs: dict[str, Any],
     ) -> object:
         if route == ROUTE_MESSAGES:
             return await litellm.anthropic.messages.acreate(**kwargs)
         if route == ROUTE_CHAT:
             if kwargs.get("stream"):
-                stream_options = kwargs.get("stream_options") or {}  # mutable-ok: empty default for a JSON field
-                kwargs["stream_options"] = {  # mutable-ok: JSON field sent to litellm.acompletion
+                stream_options = kwargs.get("stream_options") or {}
+                kwargs["stream_options"] = {
                     "include_usage": True,
                     **stream_options,
                 }

litellm/harness/handlers/cli_handler.py

@@ -73,7 +73,7 @@ def sandbox_path(private_dir: str, path: str) -> str:
 
 async def persist_dir(sandbox: Sandbox, link_path: str, cache_subpath: str) -> None:
     script_args: Final = ("-c", PERSIST_DIR_SCRIPT, "sh", link_path, cache_subpath)
-    cmd: Final = ["sh", *script_args]  # mutable-ok: Sandbox.run takes list[str]
+    cmd: Final = ["sh", *script_args]
     run = await sandbox.run(cmd)
     if run.exit_code != 0:
         verbose_logger.debug(
@@ -120,7 +120,7 @@ async def turn(self, ctx: SessionContext, prompt: str) -> AsyncIterator[Event]:
         if self._setup is None or self._private_dir is None:
             raise RuntimeError("CLIHarnessHandler.turn() called before start()")
         request = self.config.transform_turn_request(ctx, self._setup, self._private_dir, prompt, self._native_id)
-        argv: Final = list(request.argv)  # mutable-ok: Sandbox.exec takes list[str]
+        argv: Final = list(request.argv)
         proc = await ctx.sandbox.exec(argv, env=request.env, cwd=request.cwd)
         self._proc = proc
         tail: Final[deque[str]] = deque(maxlen=HARNESS_STDERR_TAIL_LINES)  # mutable-ok: bounded stderr ring buffer