Merge 8abb1c97aa into b4a8bc715f

.agents/skills/redis-access/SKILL.md

@@ -0,0 +1,23 @@
+---
+name: redis-access
+description: Add Redis operations, move raw redis-py clients, scripts or keyspaces behind RedisTransport and stores in litellm/redis, without porting them to Rust or expanding migration scope
+metadata:
+  internal: true
+---
+
+# Redis access
+
+Python reaches Redis through `RedisTransport` (`litellm/redis/transport/contract.py`), and each keyspace lives behind a store in `litellm/redis/stores/`. Callers get typed domain results from a store and keep policy (limits, fail open or closed, retries, logging)
+
+Read the reference for the task
+
+- `references/transport.md` for adding a command, the failure kinds, the namespace and TTL rules, and moving raw `init_async_client` / `async_register_script` / `.redis_client` access onto the transport
+- `references/store.md` for adding coordination state, moving a keyspace behind a store, store tests, and the Rust port
+
+Rules that always apply
+
+- New code never calls `init_async_client`, `init_pubsub_client`, `async_register_script` or `.redis_client`, and never imports redis-py outside `litellm/redis/transport/redis_py.py`
+- New coordination state gets a store, not a new `RedisCache` method, and a Python store doesn't need a Rust port
+- Keys and values in Redis are a wire format shared with older pods and Rust. A move keeps them byte for byte, and a change is a versioned migration of its own
+- A `transport` failure on a write is an unknown outcome, never "not applied"
+- Do only what was asked. An extraction isn't a store move, and a store move isn't a Rust port. Confirm before any Rust work

.agents/skills/redis-access/references/store.md

@@ -0,0 +1,84 @@
+# Redis stores
+
+A store owns one keyspace. Read this before adding coordination state, moving a keyspace out of its caller, or porting a store to Rust
+
+## Why stores
+
+Keys, encoding, TTLs, Lua and reply decoding for each keyspace are spread across Python modules today, and Rust already has a second response-cache key builder in `cache_response::cache_key`. Python and Rust pods share one Redis, so if their keys or values differ they silently split counters, cooldowns or locks. Each shared keyspace needs one definition of its data and one set of operations, and the store is that definition
+
+## Shape
+
+- One module per keyspace in `litellm/redis/stores/<keyspace>.py`
+- The module defines a `Protocol` of domain operations, named for what the caller wants (`acquire`, `release`, `increment`, `get`), never for Redis commands
+- The Python implementation is a frozen dataclass that takes a `RedisTransport` and runs its own commands on it. It owns, and nothing outside it knows
+  - the key builders, with logical keys and no namespace
+  - the value codec
+  - expiry, taken from the caller
+  - Lua, once the transport runs it
+  - reply decoding into typed results
+- Every operation returns its result or a `RedisFailure`. A value that doesn't decode is a `decode` failure, never a miss or a default
+- No logging, no retries, no fallback. Those are policy
+
+```python
+class SessionBudgetStore(Protocol):
+    async def increment(self, session_id: str, cost: float, *, ttl: timedelta) -> float | RedisFailure: ...
+    async def read(self, session_id: str) -> float | None | RedisFailure: ...
+
+
+@dataclass(frozen=True, slots=True)
+class TransportSessionBudgetStore:
+    transport: RedisTransport
+    ...
+```
+
+## Callers
+
+- Depend on the keyspace's `Protocol`, built once from `redis_cache.transport()` and injected where the caller can take it
+- `match` the result, ending in `assert_never` when it has more than two cases
+- Keep policy in the caller: which TTL applies, fail open or closed, and the `with_service_target` that labels its commands. The transport already logs failures and emits service events
+- A caller never builds a key, touches a client or runs a command for a keyspace that has a store. Exposing a key builder for an existing public helper is fine as a delegate
+
+## Adding new coordination state
+
+Add a store, not a new `RedisCache` method. A Python store is the finished state for now, and adding one does not require a Rust port
+
+## Moving an existing keyspace into a store
+
+- Move the key builders, codec, TTLs, Lua and decoding verbatim. Keep policy in the caller
+- Keys and values already in Redis are a wire format. Old and new pods read each other's writes during a rolling deploy, so a move keeps them byte for byte
+- Changing a key, codec or TTL unit is never part of a move. It is its own change that writes a new versioned key and reads both for one release
+- Move one keyspace per change
+
+Keyspaces still in their callers, as candidates (paths under `litellm/`, not exhaustive)
+
+- response cache, `caching/caching.py`
+- rate limit windows, `proxy/hooks/parallel_request_limiter_v3.py`
+- deployment cooldowns, `router_utils/cooldown_cache.py`
+- routing affinity, `caching/affinity_cache.py`
+- pod and MCP refresh locks, `proxy/db/db_transaction_queue/pod_lock_manager.py` and `proxy/mcp_server/outbound_credentials/redis_distributed_lock.py`, which share one owner-checked release
+- login throttle, `proxy/auth/login_throttle.py`
+- session caps, `proxy/hooks/max_budget_per_session_limiter.py` and `max_iterations_limiter.py`, which share one counter pattern
+- enqueued batch tokens, `proxy/hooks/batch_enqueued_tokens.py`
+- spend counters, `proxy/spend_tracking/spend_counter_batch.py`
+
+Response cache identity (normalization, hashing into a logical key) is backend-independent and is not store logic. The store only maps that logical key to storage. Reuse or move `cache_response::cache_key`, never write a third key builder
+
+## Tests
+
+- Test the store through its caller against `fake_redis_cache(server)` from `tests/_support/fake_redis.py`. That is a real `RedisCache` on a `fakeredis.FakeServer`, so namespace, breaker and failure mapping are the production ones
+- Cover each result the caller acts on, that the TTL lands (`pttl` on a raw `FakeAsyncRedis`), and an outage (`server.connected = False`)
+- When a keyspace moves, seed with the old writer and read back with the old reader, so old and new pods read each other's writes
+- Assert returned values and Redis state, never the commands sent
+
+## Porting a store to Rust
+
+Confirm with the user before starting the first port, since the bridge for it doesn't exist yet
+
+- The keyspace's module in `litellm-rust/crates/redis-store` implements the same operations as async fns on `crates/redis-client`. Its `AGENTS.md` has the crate rules
+- A `#[pyclass]` per store in `litellm-rust/crates/python-bridge/src/redis/` wraps that module, built with the shared `NativeRedisClient`, so Python opens no extra connections
+- A `Native<Keyspace>Store` in the Python module implements the same `Protocol` over the pyclass, so callers don't change. The implementation is picked at construction, and the transport-backed one stays for config the native client doesn't support (Sentinel, cloud auth, TLS cert options)
+- Results come back as Pydantic models generated from the Rust types, frozen, so callers keep matching on the same classes. A decode error maps to `RedisFailure(kind="decode")`
+- Rust callers use the same module, so there is one implementation per operation
+- Native stores skip the Python guard, so the breaker and the REDIS service events have to come from Rust before the first port
+- The codec has to write the exact bytes Python writes. Pod lock owners are `json.dumps` output, which escapes non-ASCII, and `serde_json` doesn't, so a Rust pod couldn't release a lock a Python pod took
+- Each port ships cross-implementation tests against one Redis, covering keys, values, expiry and atomicity through a rolling deploy

.agents/skills/redis-access/references/transport.md

@@ -0,0 +1,76 @@
+# RedisTransport
+
+How Python code reaches Redis. Read this before adding a command, or moving raw redis-py access off `RedisCache`
+
+## Why it exists
+
+Today commands reach Redis three ways
+
+- Cache methods on `RedisCache` and `RedisBatch` (get, set, incr, lists, pipelines), which apply the namespace and the circuit breaker
+- Raw access, where callers take `init_async_client()`, `init_pubsub_client()`, `async_register_script()` or `.redis_client` and run commands, Lua and pub/sub themselves
+  - Scripts from `async_register_script` keep the namespace and breaker
+  - Raw clients skip both, so every caller namespaces keys and channels by hand, and some forget
+- Vector search (`RedisSemanticCache` through RedisVL, `ValkeySemanticCache` and the Valkey vector store through `Redis.from_url`), which builds its own clients on separate connections
+
+`RedisTransport` replaces the raw path with one command-shaped seam, so the backend behind it can later swap from redis-py to `litellm-rust/crates/redis-client` without touching callers
+
+## The contract
+
+Defined in `litellm/redis/transport/contract.py`
+
+- Commands are frozen dataclasses, `Get(key)`, `Set(key, value, ttl, nx=False)` and `Delete(key)`. `Arg` is `str | bytes | int | float`
+- The transport has one method, `execute(command)`, overloaded per command for its reply type (`bytes | None`, `bool`, `int`)
+- Every command returns its value or a `RedisFailure`. Nothing raises for a Redis problem
+- `RedisFailure.kind`
+  - `transport` is a timeout or lost connection. The command may still have applied, so a write that fails this way has an unknown outcome. `timeout=True` marks timeouts
+  - `server` means Redis answered with an error (WRONGTYPE). The connection is fine and the breaker does not count it
+  - `decode` means the reply was not the shape the method or store expected
+  - `breaker_open` means the breaker refused the call without touching the socket
+- The namespace is applied once, inside the transport. A key that already starts with `{namespace}:` is left alone, matching `RedisCache.check_and_fix_namespace` and `crates/redis-client`. Callers and stores pass logical keys
+- Every write takes its TTL from the caller. `ttl=None` means no expiry, there is no default
+- Replies are undecoded bytes. Decoding belongs to whoever owns the keyspace
+
+These rules are shared with `crates/redis-client` (its `AGENTS.md` has the Rust side). If you change one, change both, since Python and Rust pods share keys in one Redis
+
+## The redis-py backend
+
+- `litellm/redis/transport/redis_py.py` holds `RedisPyTransport`, the only module that touches redis-py's client
+- `RedisCache.transport()` builds one over that cache's own async client, namespace and circuit breaker. No new connection, no second breaker
+- The guard is `BreakerGuard` in `redis_cache.py`. It runs the call under the cache's breaker and maps exceptions with `_redis_failure`
+  - `ResponseError` / `DataError` that are not health failures become `server`
+  - health failures and other `RedisError`s become `transport`
+  - anything else re-raises, since it is a bug, not a Redis failure
+- `BreakerGuard` wraps `ReportingGuard`, which owns observability for every command
+  - one REDIS service success or failure event per command, with the command name as `call_type`, sent to the cache's `service_logger_obj`. A `breaker_open` refusal sends none
+  - the event and the log line carry the caller's `with_service_target`, so callers label their commands and emit nothing themselves
+  - one WARNING per failure through the timeout log throttle, `breaker_open` at DEBUG
+- Callers don't log Redis health. Log only what a failure means for the feature, and only when it adds something the transport line doesn't say
+
+## Adding a command
+
+- Add a frozen dataclass for the command to `contract.py`, add it to `Command`, and add an `execute` overload for its reply type. Keep it command-shaped, never redis-py shaped (no `**kwargs`, no client options), so the native client fits behind it unchanged
+- Add a `match` arm in every backend's `execute`, and add the redis-py method to the `AsyncRedisCommands` Protocol there and `_AsyncRedisCommands` in `redis_cache.py`
+- Namespace every key argument with `_key`, take TTLs as `timedelta`, send them as milliseconds
+- Validate the reply shape and return `decode` for anything unexpected
+- Test it in `tests/unit/redis/transport/test_redis_py.py` against fakeredis: the namespace is applied once, TTL lands, a rejected command is `server`, an outage is `transport`. Service events and logging come from the guard, so a new command needs no test of its own for them
+- Multi-key commands must keep their keys in one slot, or they break on cluster depending on slot ownership
+
+## Moving raw access onto the transport
+
+- Replace the raw client or script handle with transport calls, and delete hand-written namespacing and script registration
+- Keep the caller's outcome per failure. If it failed closed on any exception, it fails closed on any `RedisFailure`
+- Don't treat a `transport` failure on a write as "not applied"
+
+## Not on the transport yet
+
+- Pub/sub (config and auth-cache invalidation), admin commands, pipelines and `RedisBatch` stay on `RedisCache` until the transport grows them
+- Vector and semantic search keep their own clients. The plan gives them native index handles on `crates/redis-client`, not `RedisTransport`
+
+## Where it's going
+
+- A `NativeRedisTransport` (a PyO3 adapter over one `RedisClient` per resolved target) goes behind the same Protocol and takes the same `Guard`. The backend is picked at construction, with redis-py kept for config the native client doesn't support yet (Sentinel, cloud auth, TLS cert options)
+- Settings for both stacks resolve through `crates/config` (`RedisSettings.resolve`), so Python and Rust connect to the same target with the same namespace
+- During the swap the Python `RedisCircuitBreaker` still wraps the transport. The breaker and the service events have to be in Rust before any store runs natively, because native stores call `crates/redis-client` directly
+- Once stores and admin/pub-sub have replacements, redis-py and redisvl are dropped or become optional extras
+
+Don't build any of this unless asked

litellm-rust/Cargo.lock

@@ -4371,6 +4371,10 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "litellm-redis-store"
+version = "0.1.0"
+
 [[package]]
 name = "litellm-redis-testing"
 version = "0.1.0"

litellm-rust/Cargo.toml

@@ -72,6 +72,7 @@ litellm-host-python = { path = "crates/host-python" }
 litellm-python-compat = { path = "crates/python-compat" }
 litellm-circuit-breaker = { path = "crates/circuit-breaker" }
 litellm-redis-client = { path = "crates/redis-client" }
+litellm-redis-store = { path = "crates/redis-store" }
 litellm-redis-testing = { path = "crates/redis-testing" }
 litellm-cost = { path = "crates/cost" }
 litellm-model-catalog = { path = "crates/model-catalog" }

litellm-rust/crates/redis-store/AGENTS.md

@@ -0,0 +1,22 @@
+- Redis-backed domain operations shared by the Rust gateway and Python, like a `crates/db` repository. Callers ask for domain operations (`acquire`, `increment`, `get`) and never see a Redis command
+- One module per keyspace (rate limit windows, session caps, enqueued tokens, cooldowns, affinity pins, locks, login throttle, spend counters, update buffers, response polling). The module owns its key layout, value codec, expiry, Lua and reply decoding, and nothing outside it builds those keys
+- Operations are async fns that take a `&litellm_redis_client::RedisClient` and run their commands on it. The client applies the namespace and the breaker, so a store never does either and never goes back through Python's `RedisTransport`
+- Results are domain types (`Acquire::Taken`, `Option<f64>`), never a raw `redis::Value`. A reply that doesn't decode is an error, never a miss or a default count
+- Callers keep policy, like which limit applies, fail open or closed, retries and logging
+
+## Wire contract
+- Keys, value bytes, TTL units and Lua behavior are a wire format shared with Python pods and older releases on the same Redis. A difference doesn't error, it silently splits counters, budgets or locks between pods during a rolling deploy or rollback
+- A keyspace ported from Python keeps the bytes Python writes exactly, including hash tags, `json.dumps` spacing and ASCII escaping, float formatting and seconds vs milliseconds
+- Never change a key, codec or TTL unit in place. Write under a new versioned key, read both for one release, then drop the old one, as its own change
+
+## Shape
+- Logical keys only, the client adds the namespace. Lua that touches more than one key keeps them in one slot with a `{hash tag}`, and declares every key in KEYS
+- Lua is a `&'static str` next to the operation that runs it
+- Response-cache identity (normalization, hashing into a logical key) is not store logic. This crate only maps that logical key to storage. Reuse or move `cache-response`'s key builder, never write a third
+- Pub/sub, admin commands and vector search are not keyspaces and don't belong here
+- Python reaches a store through one `#[pyclass]` per module in `crates/python-bridge/src/redis/`, built with the shared `NativeRedisClient`, behind the keyspace's Python store Protocol in `litellm/redis/stores/`
+
+## Tests
+- Run operations against real Redis from `litellm-redis-testing`, on each topology through `#[values(Topology::Standalone, Topology::Cluster)]`. Never assert command bytes or a mocked wire
+- Assert decoded results, TTLs and atomicity through the store's own operations and Redis state
+- A keyspace ported from Python ships cross-implementation tests against one Redis, where the Python store writes and this crate reads and the reverse

litellm-rust/crates/redis-store/Cargo.toml

@@ -0,0 +1,6 @@
+[package]
+name = "litellm-redis-store"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true

litellm/_service_logger.py

@@ -1,5 +1,5 @@
 import asyncio
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable, Coroutine, Mapping
 from datetime import datetime, timedelta
 from typing import TYPE_CHECKING, Any, Final, Protocol
 
@@ -224,15 +224,16 @@ class ServiceLogging(CustomLogger):
         parent_otel_span: Span | None = None,
         start_time: datetime | float | None = None,
         end_time: datetime | float | None = None,
-        event_metadata: dict | None = None,
+        event_metadata: Mapping[str, object] | None = None,
         caller: str | None = None,
-    ):
+    ) -> None:
         """
         - For counting if the redis, postgres call is successful
         """
         if self.mock_testing:
             self.mock_testing_async_success_hook += 1
 
+        metadata: Final = None if event_metadata is None else dict(event_metadata)
         payload: Final = ServiceLoggerPayload(
             is_error=False,
             error=None,
@@ -241,7 +242,7 @@ class ServiceLogging(CustomLogger):
             call_type=call_type,
             caller=caller,
             target=current_service_target() if service == ServiceTypes.REDIS else None,
-            event_metadata=event_metadata,
+            event_metadata=metadata,
         )
 
         # OTel loggers already fired this event. ``service_callback`` can hold more
@@ -262,7 +263,7 @@ class ServiceLogging(CustomLogger):
                     parent_otel_span=parent_otel_span,
                     start_time=start_time,
                     end_time=end_time,
-                    event_metadata=event_metadata,
+                    event_metadata=metadata,
                 )
             else:
                 _otel_logger_to_use = self._resolve_otel_service_logger(callback)
@@ -278,7 +279,7 @@ class ServiceLogging(CustomLogger):
                         parent_otel_span=parent_otel_span,
                         start_time=start_time,
                         end_time=end_time,
-                        event_metadata=event_metadata,
+                        event_metadata=metadata,
                     )
 
     async def init_prometheus_services_logger_if_none(self):
@@ -325,9 +326,9 @@ class ServiceLogging(CustomLogger):
         parent_otel_span: Span | None = None,
         start_time: datetime | float | None = None,
         end_time: float | datetime | None = None,
-        event_metadata: dict | None = None,
+        event_metadata: Mapping[str, object] | None = None,
         caller: str | None = None,
-    ):
+    ) -> None:
         """
         - For counting if the redis, postgres call is unsuccessful
         """
@@ -340,6 +341,7 @@ class ServiceLogging(CustomLogger):
         elif isinstance(error, str):
             error_message = error
 
+        metadata: Final = None if event_metadata is None else dict(event_metadata)
         payload: Final = ServiceLoggerPayload(
             is_error=True,
             error=error_message,
@@ -348,7 +350,7 @@ class ServiceLogging(CustomLogger):
             call_type=call_type,
             caller=caller,
             target=current_service_target() if service == ServiceTypes.REDIS else None,
-            event_metadata=event_metadata,
+            event_metadata=metadata,
         )
 
         # Dedupe OTel loggers per event — see ``async_service_success_hook`` for why
@@ -369,7 +371,7 @@ class ServiceLogging(CustomLogger):
                     parent_otel_span=parent_otel_span,
                     start_time=start_time,
                     end_time=end_time,
-                    event_metadata=event_metadata,
+                    event_metadata=metadata,
                 )
             else:
                 _otel_logger_to_use = self._resolve_otel_service_logger(callback)
@@ -387,7 +389,7 @@ class ServiceLogging(CustomLogger):
                         parent_otel_span=parent_otel_span,
                         start_time=start_time,
                         end_time=end_time,
-                        event_metadata=event_metadata,
+                        event_metadata=metadata,
                     )
 
     async def async_post_call_failure_hook(

litellm/caching/redis_cache.py

@@ -30,7 +30,7 @@ from typing import TYPE_CHECKING, Any, Final, Protocol, TypeVar, cast
 from pydantic import TypeAdapter
 
 import litellm
-from litellm._internal_context import current_service_caller
+from litellm._internal_context import current_service_caller, current_service_target
 from litellm._logging import print_verbose, verbose_logger
 from litellm.constants import (
     DEFAULT_REDIS_MAJOR_VERSION,
@@ -42,6 +42,7 @@ from litellm.constants import (
 )
 from litellm.litellm_core_utils.core_helpers import get_parent_otel_span_from_kwargs
 from litellm.litellm_core_utils.coroutine_checker import coroutine_checker
+from litellm.redis.transport.contract import RedisFailure, RedisTransport
 from litellm.types.caching import (
     RedisPipelineIncrementOperation,
     RedisPipelineLpopOperation,
@@ -81,6 +82,10 @@ class _AsyncRedisCommands(Protocol):
 
     def ping(self) -> Awaitable[bool]: ...
 
+    def get(self, name: str) -> Awaitable[object]: ...
+
+    def set(self, name: str, value: str | bytes | float, *, nx: bool, px: int | None) -> Awaitable[object]: ...
+
     def delete(self, *names: str) -> Awaitable[int]: ...
 
     def ttl(self, name: str) -> Awaitable[int]: ...
@@ -103,6 +108,7 @@ _GENERIC_CALLER_MODULES: Final = frozenset(
         "litellm.caching.dual_cache",
         "litellm.caching.caching",
         "litellm.rust_bridge.execution",
+        "litellm.redis.transport.redis_py",
         "contextlib",
     }
 )
@@ -512,25 +518,45 @@ def log_redis_failure(
     if isinstance(exc, RedisCircuitBreakerOpenError):
         logger.debug("%s: %s", message, exc, stacklevel=2)
         return
-    exc_info: Final = exc if with_traceback else None
-    if not is_redis_timeout_failure(exc):
-        logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2)
+    _log_throttling_timeouts(
+        logger, level, message, exc, timeout=is_redis_timeout_failure(exc), exc_info=exc if with_traceback else None
+    )
+
+
+def _log_redis_failure_value(logger: logging.Logger, level: int, message: str, failure: RedisFailure) -> None:
+    if failure.kind == "breaker_open":
+        logger.debug("%s: %s", message, failure.message, stacklevel=2)
+        return
+    _log_throttling_timeouts(logger, level, message, failure.message, timeout=failure.timeout, exc_info=None)
+
+
+def _log_throttling_timeouts(
+    logger: logging.Logger,
+    level: int,
+    message: str,
+    detail: object,
+    *,
+    timeout: bool,
+    exc_info: BaseException | None,
+) -> None:
+    if not timeout:
+        logger.log(level, "%s: %s", message, detail, exc_info=exc_info, stacklevel=3)
         return
     suppressed: Final = _redis_timeout_log_throttle.admit()
     if suppressed is None:
-        logger.debug("%s: %s", message, exc, stacklevel=2)
+        logger.debug("%s: %s", message, detail, stacklevel=3)
         return
     if suppressed == 0:
-        logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2)
+        logger.log(level, "%s: %s", message, detail, exc_info=exc_info, stacklevel=3)
         return
     logger.log(
         level,
         "%s: %s (%d more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
         message,
-        exc,
+        detail,
         suppressed,
         exc_info=exc_info,
-        stacklevel=2,
+        stacklevel=3,
     )
 
 
@@ -584,6 +610,87 @@ async def _run_under_circuit_breaker(
     return result
 
 
+def _redis_failure(exc: Exception) -> RedisFailure | None:
+    from redis.exceptions import DataError, RedisError, ResponseError
+
+    if isinstance(exc, (ResponseError, DataError)) and not _is_redis_health_failure(exc):
+        return RedisFailure(kind="server", message=str(exc))
+    if _is_redis_health_failure(exc) or isinstance(exc, RedisError):
+        return RedisFailure(kind="transport", message=str(exc), timeout=is_redis_timeout_failure(exc))
+    return None
+
+
+class _Admit(Protocol):
+    def __call__(self, call: Callable[[], Awaitable[_RedisCallResult]], /) -> Awaitable[_RedisCallResult]: ...
+
+
+def _admit_directly(call: Callable[[], Awaitable[_RedisCallResult]]) -> Awaitable[_RedisCallResult]:
+    return call()
+
+
+@dataclass(frozen=True, slots=True)
+class ReportingGuard:
+    events: "ServiceLogging"
+
+    async def __call__(self, name: str, call: Callable[[], Awaitable[_RedisCallResult]], /) -> _RedisCallResult:
+        caller: Final = _get_call_stack_info()
+        start_time: Final = time.time()
+        result: Final = await call()
+        end_time: Final = time.time()
+        if not isinstance(result, RedisFailure):
+            asyncio.create_task(
+                self.events.async_service_success_hook(
+                    service=ServiceTypes.REDIS,
+                    call_type=name,
+                    duration=end_time - start_time,
+                    start_time=start_time,
+                    end_time=end_time,
+                    caller=caller,
+                )
+            )
+            return result
+        target: Final = current_service_target()
+        _log_redis_failure_value(
+            verbose_logger, logging.WARNING, f"Redis {name} failed" + (f" for {target}" if target else ""), result
+        )
+        if result.kind != "breaker_open":
+            asyncio.create_task(
+                self.events.async_service_failure_hook(
+                    service=ServiceTypes.REDIS,
+                    duration=end_time - start_time,
+                    error=result.message,
+                    call_type=name,
+                    start_time=start_time,
+                    end_time=end_time,
+                    caller=caller,
+                )
+            )
+        return result
+
+
+@dataclass(frozen=True, slots=True)
+class BreakerGuard:
+    breaker: RedisCircuitBreaker
+    report: ReportingGuard
+    admit: _Admit = _admit_directly
+
+    async def __call__(
+        self, name: str, call: Callable[[], Awaitable[_RedisCallResult]], /
+    ) -> _RedisCallResult | RedisFailure:
+        return await self.report(name, lambda: self._run(name, call))
+
+    async def _run(self, name: str, call: Callable[[], Awaitable[_RedisCallResult]]) -> _RedisCallResult | RedisFailure:
+        try:
+            return await _run_under_circuit_breaker(self.breaker, name, lambda: self.admit(call))
+        except RedisCircuitBreakerOpenError as e:
+            return RedisFailure(kind="breaker_open", message=str(e))
+        except Exception as e:
+            failure: Final = _redis_failure(e)
+            if failure is None:
+                raise
+            return failure
+
+
 def _run_under_circuit_breaker_sync(
     breaker: RedisCircuitBreaker,
     name: str,
@@ -829,6 +936,15 @@ class RedisCache(BaseCache):
     def _async_commands(self) -> _AsyncRedisCommands:
         return self.init_async_client()
 
+    def transport(self) -> RedisTransport:
+        from litellm.redis.transport.redis_py import RedisPyTransport
+
+        return RedisPyTransport(
+            client=self._async_commands,
+            namespace=self.namespace,
+            guard=BreakerGuard(self._circuit_breaker, ReportingGuard(self.service_logger_obj), self._admit_async_call),
+        )
+
     def check_and_fix_namespace(self, key: str) -> str:
         """
         Make sure each key starts with the given namespace

litellm/redis/AGENTS.md

@@ -0,0 +1,14 @@
+For adding Redis access or moving raw Redis access here, follow [.agents/skills/redis-access/SKILL.md](../../.agents/skills/redis-access/SKILL.md)
+
+- `transport/contract.py` is the command-shaped seam. Keep it one command dataclass per Redis command the Python code uses, run through `execute`, never redis-py shaped, so `litellm-rust/crates/redis-client` can sit behind it unchanged
+- Rules shared with `crates/redis-client`, so the two implementations agree
+  - The namespace is applied here, once. A key that already starts with `{namespace}:` is left alone
+  - Every write takes its TTL from the caller, `None` means no expiry, there is no default
+  - Replies are undecoded bytes
+  - Failures are values. `transport` covers timeouts and connection loss, where the command may still have applied. `server` means Redis answered with an error and the connection is fine
+- `transport/redis_py.py` is the only module that touches redis-py's client. The breaker and the exception mapping come from `RedisCache` through the injected guard
+- Every backend runs each command through a `Guard` that emits one REDIS service event and logs Redis health failures, so swapping backends never drops metrics. Callers don't log Redis health
+- `litellm/caching` imports `litellm/redis`, never the reverse
+- `stores/` holds one module per keyspace, and a keyspace that has a store is only reached through it. A store is a `Protocol` of domain operations plus implementations of it. It owns its keys, value codec, expiry, Lua and decoding, does its own IO, and callers never see a Redis command
+- The Python implementation runs on any `RedisTransport`. A native implementation wraps the keyspace's `crates/redis-store` module and runs on `crates/redis-client` directly, never through `RedisTransport`
+- Keys, value bytes and TTL units in Redis are a wire format shared with older pods and Rust. Never change them in place, see `crates/redis-store/AGENTS.md`

litellm/redis/transport/contract.py

@@ -0,0 +1,56 @@
+import math
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Final, Literal, Protocol, TypeAlias, TypeVar, overload
+
+Arg: TypeAlias = str | bytes | int | float
+
+_T: Final = TypeVar("_T")
+
+
+@dataclass(frozen=True, slots=True)
+class RedisFailure:
+    kind: Literal["transport", "server", "decode", "breaker_open"]
+    message: str
+    timeout: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class Get:
+    key: str
+
+
+@dataclass(frozen=True, slots=True)
+class Set:
+    key: str
+    value: Arg
+    ttl: timedelta | None
+    nx: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class Delete:
+    key: str
+
+
+Command: TypeAlias = Get | Set | Delete
+
+
+class Guard(Protocol):
+    def __call__(self, name: str, call: Callable[[], Awaitable[_T]], /) -> Awaitable[_T | RedisFailure]: ...
+
+
+def ttl_ms(ttl: timedelta) -> int:
+    return max(1, math.ceil(ttl / timedelta(milliseconds=1)))
+
+
+class RedisTransport(Protocol):
+    @overload
+    async def execute(self, command: Get) -> bytes | None | RedisFailure: ...
+
+    @overload
+    async def execute(self, command: Set) -> bool | RedisFailure: ...
+
+    @overload
+    async def execute(self, command: Delete) -> int | RedisFailure: ...

litellm/redis/transport/redis_py.py

@@ -0,0 +1,59 @@
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Final, Protocol, assert_never, overload
+
+from litellm.redis.transport.contract import Arg, Command, Delete, Get, Guard, RedisFailure, Set, ttl_ms
+
+
+class AsyncRedisCommands(Protocol):
+    def get(self, name: str) -> Awaitable[object]: ...
+
+    def set(self, name: str, value: Arg, *, nx: bool, px: int | None) -> Awaitable[object]: ...
+
+    def delete(self, *names: str) -> Awaitable[int]: ...
+
+
+def _decode_failure(value: object) -> RedisFailure:
+    return RedisFailure(kind="decode", message=f"unexpected Redis reply of type {type(value).__name__}")
+
+
+def _bytes_reply(reply: object) -> bytes | None | RedisFailure:
+    if reply is None or isinstance(reply, (bytes, RedisFailure)):
+        return reply
+    if isinstance(reply, str):
+        return reply.encode()
+    return _decode_failure(reply)
+
+
+@dataclass(frozen=True, slots=True)
+class RedisPyTransport:
+    client: Callable[[], AsyncRedisCommands]
+    namespace: str | None
+    guard: Guard
+
+    def _key(self, key: str) -> str:
+        if not self.namespace or key.startswith(f"{self.namespace}:"):
+            return key
+        return f"{self.namespace}:{key}"
+
+    @overload
+    async def execute(self, command: Get) -> bytes | None | RedisFailure: ...
+
+    @overload
+    async def execute(self, command: Set) -> bool | RedisFailure: ...
+
+    @overload
+    async def execute(self, command: Delete) -> int | RedisFailure: ...
+
+    async def execute(self, command: Command) -> bytes | None | bool | int | RedisFailure:
+        match command:
+            case Get(key=key):
+                return _bytes_reply(await self.guard("get", lambda: self.client().get(self._key(key))))
+            case Set(key=key, value=value, ttl=ttl, nx=nx):
+                px: Final = None if ttl is None else ttl_ms(ttl)
+                written: Final = await self.guard("set", lambda: self.client().set(self._key(key), value, nx=nx, px=px))
+                return written if isinstance(written, RedisFailure) else written is not None and written is not False
+            case Delete(key=key):
+                return await self.guard("delete", lambda: self.client().delete(self._key(key)))
+            case _:
+                assert_never(command)

tests/_support/fake_redis.py

@@ -0,0 +1,43 @@
+import fakeredis
+
+import litellm
+from litellm._service_logger import ServiceLogging
+from litellm.caching.redis_cache import (
+    BreakerGuard,
+    RedisCache,
+    RedisCircuitBreaker,
+    ReportingGuard,
+)
+from litellm.redis.transport.redis_py import RedisPyTransport
+
+
+def fake_redis_cache(
+    server: fakeredis.FakeServer, namespace: str | None = None, service_logger: ServiceLogging | None = None
+) -> RedisCache:
+    cache = RedisCache(
+        host="127.0.0.1",
+        port=1,
+        namespace=namespace,
+        socket_timeout=0.1,
+        service_logger_obj=service_logger or ServiceLogging(),
+    )
+    litellm.in_memory_llm_clients_cache.set_cache(
+        key=cache._get_async_client_cache_key(),  # pyright: ignore[reportPrivateUsage]  # seeds the per-loop client slot RedisCache reads
+        value=fakeredis.FakeAsyncRedis(server=server),
+    )
+    return cache
+
+
+def fake_redis_transport(
+    client: fakeredis.FakeAsyncRedis,
+    *,
+    failure_threshold: int,
+    events: ServiceLogging | None = None,
+    namespace: str | None = None,
+) -> RedisPyTransport:
+    breaker = RedisCircuitBreaker(failure_threshold=failure_threshold, recovery_timeout=3600)
+    return RedisPyTransport(
+        client=lambda: client,
+        namespace=namespace,
+        guard=BreakerGuard(breaker, ReportingGuard(events or ServiceLogging())),
+    )

tests/unit/redis/conftest.py

@@ -0,0 +1,12 @@
+import fakeredis
+import pytest
+
+
+@pytest.fixture
+def server() -> fakeredis.FakeServer:
+    return fakeredis.FakeServer()
+
+
+@pytest.fixture
+def raw(server: fakeredis.FakeServer) -> fakeredis.FakeAsyncRedis:
+    return fakeredis.FakeAsyncRedis(server=server)

tests/unit/redis/transport/test_contract.py

@@ -0,0 +1,17 @@
+from datetime import timedelta
+
+import pytest
+
+from litellm.redis.transport.contract import ttl_ms
+
+
+@pytest.mark.parametrize(
+    ("ttl", "expected"),
+    [
+        pytest.param(timedelta(seconds=1.5), 1500, id="whole-ms"),
+        pytest.param(timedelta(microseconds=1500), 2, id="rounds-up"),
+        pytest.param(timedelta(microseconds=1), 1, id="sub-ms-is-not-zero"),
+    ],
+)
+def test_ttl_ms_never_shortens_the_ttl_or_sends_zero(ttl: timedelta, expected: int) -> None:
+    assert ttl_ms(ttl) == expected

tests/unit/redis/transport/test_redis_py.py

@@ -0,0 +1,237 @@
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Final, Literal
+
+import fakeredis
+import pytest
+
+from litellm._internal_context import current_service_target, service_target
+from litellm._service_logger import ServiceLogging
+from litellm.redis.transport.contract import Delete, Get, RedisFailure, Set
+from litellm.types.services import ServiceTypes
+from tests._support.fake_redis import fake_redis_cache, fake_redis_transport
+
+
+@pytest.mark.asyncio
+async def test_every_command_prefixes_the_namespace_once(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    transport = fake_redis_cache(server, namespace="ns").transport()
+
+    assert await transport.execute(Set("k", "v", ttl=None)) is True
+    assert await transport.execute(Set("ns:gone", "v", ttl=None)) is True
+
+    assert sorted(await raw.keys("*")) == [b"ns:gone", b"ns:k"]
+    assert await transport.execute(Get("k")) == await transport.execute(Get("ns:k")) == b"v"
+    assert await transport.execute(Delete("gone")) == 1
+    assert await raw.keys("*") == [b"ns:k"]
+
+
+@pytest.mark.asyncio
+async def test_set_nx_refuses_a_live_key_and_applies_the_ttl(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    transport = fake_redis_cache(server).transport()
+
+    assert await transport.execute(Set("k", "first", ttl=timedelta(seconds=30), nx=True)) is True
+    assert await transport.execute(Set("k", "second", ttl=timedelta(seconds=30), nx=True)) is False
+
+    assert await raw.get("k") == b"first"
+    assert 0 < await raw.pttl("k") <= 30_000
+
+
+@pytest.mark.asyncio
+async def test_set_without_ttl_persists(server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis) -> None:
+    transport = fake_redis_cache(server).transport()
+
+    assert await transport.execute(Set("k", "v", ttl=None)) is True
+
+    assert await raw.pttl("k") == -1
+
+
+@pytest.mark.asyncio
+async def test_delete_reports_how_many_keys_it_removed(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    transport = fake_redis_cache(server).transport()
+    await raw.set("k", "v")
+
+    assert await transport.execute(Delete("k")) == 1
+    assert await transport.execute(Delete("k")) == 0
+    assert await raw.exists("k") == 0
+
+
+@pytest.mark.asyncio
+async def test_a_rejected_command_is_a_server_failure_and_keeps_the_connection(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    transport = fake_redis_cache(server).transport()
+    await raw.rpush("list", "x")
+
+    rejected = await transport.execute(Get("list"))
+
+    assert isinstance(rejected, RedisFailure)
+    assert rejected.kind == "server"
+    assert await transport.execute(Set("k", "v", ttl=None)) is True
+
+
+@pytest.mark.asyncio
+async def test_an_unreachable_server_is_a_transport_failure(server: fakeredis.FakeServer) -> None:
+    transport = fake_redis_cache(server).transport()
+    server.connected = False
+
+    failure = await transport.execute(Get("k"))
+
+    assert isinstance(failure, RedisFailure)
+    assert failure.kind == "transport"
+
+
+@pytest.mark.asyncio
+async def test_a_transport_failure_opens_the_breaker_and_it_answers_without_redis(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    transport = fake_redis_transport(raw, failure_threshold=1)
+    server.connected = False
+
+    tripped = await transport.execute(Get("k"))
+    server.connected = True
+    refused = await transport.execute(Get("k"))
+
+    assert isinstance(tripped, RedisFailure)
+    assert tripped.kind == "transport"
+    assert isinstance(refused, RedisFailure)
+    assert refused.kind == "breaker_open"
+
+
+@pytest.mark.asyncio
+async def test_replies_stay_bytes_on_a_client_that_decodes_responses(server: fakeredis.FakeServer) -> None:
+    transport = fake_redis_transport(
+        fakeredis.FakeAsyncRedis(server=server, decode_responses=True), failure_threshold=1
+    )
+
+    assert await transport.execute(Set("k", "v", ttl=None)) is True
+    assert await transport.execute(Get("k")) == b"v"
+
+
+async def _settle() -> None:
+    pending: Final = {task for task in asyncio.all_tasks() if task is not asyncio.current_task()}
+    if pending:
+        await asyncio.gather(*pending)
+        await _settle()
+
+
+@dataclass(frozen=True, slots=True)
+class _Event:
+    outcome: Literal["success", "failure"]
+    call_type: str
+    target: str | None
+    failed_with: str | None = None
+
+
+class _RecordingEvents(ServiceLogging):
+    def __init__(self) -> None:
+        super().__init__()
+        self.seen: tuple[_Event, ...] = ()
+
+    async def async_service_success_hook(
+        self,
+        service: ServiceTypes,
+        call_type: str,
+        duration: float,
+        parent_otel_span: object | None = None,
+        start_time: datetime | float | None = None,
+        end_time: datetime | float | None = None,
+        event_metadata: object | None = None,
+        caller: str | None = None,
+    ) -> None:
+        assert service == ServiceTypes.REDIS and 0 <= duration
+        self.seen = (*self.seen, _Event("success", call_type, current_service_target()))
+
+    async def async_service_failure_hook(
+        self,
+        service: ServiceTypes,
+        duration: float,
+        error: str | Exception,
+        call_type: str,
+        parent_otel_span: object | None = None,
+        start_time: datetime | float | None = None,
+        end_time: float | datetime | None = None,
+        event_metadata: object | None = None,
+        caller: str | None = None,
+    ) -> None:
+        assert service == ServiceTypes.REDIS and 0 <= duration
+        self.seen = (*self.seen, _Event("failure", call_type, current_service_target(), str(error)))
+
+
+@pytest.mark.asyncio
+async def test_every_command_reports_one_service_event_under_the_callers_target(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    events = _RecordingEvents()
+    transport = fake_redis_transport(raw, failure_threshold=2, events=events)
+
+    with service_target("response_polling"):
+        await transport.execute(Set("k", "v", ttl=None))
+        await transport.execute(Get("k"))
+        server.connected = False
+        failed = await transport.execute(Delete("k"))
+    await asyncio.sleep(0)
+
+    assert isinstance(failed, RedisFailure)
+    assert events.seen == (
+        _Event("success", "set", "response_polling"),
+        _Event("success", "get", "response_polling"),
+        _Event("failure", "delete", "response_polling", failed.message),
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_call_the_breaker_refuses_reports_no_service_event(
+    server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
+) -> None:
+    events = _RecordingEvents()
+    transport = fake_redis_transport(raw, failure_threshold=1, events=events)
+    server.connected = False
+
+    await transport.execute(Get("k"))
+    refused = await transport.execute(Get("k"))
+    await asyncio.sleep(0)
+
+    assert isinstance(refused, RedisFailure) and refused.kind == "breaker_open"
+    assert tuple(event.outcome for event in events.seen) == ("failure",)
+
+
+@pytest.mark.asyncio
+async def test_the_cache_transport_reports_to_the_caches_service_logger(server: fakeredis.FakeServer) -> None:
+    service_logger = ServiceLogging(mock_testing=True)
+    transport = fake_redis_cache(server, service_logger=service_logger).transport()
+    await _settle()
+    before = (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook)
+
+    await transport.execute(Get("k"))
+    server.connected = False
+    await transport.execute(Get("k"))
+    await _settle()
+
+    after = (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook)
+    assert (after[0] - before[0], after[1] - before[1]) == (1, 1)
+
+
+@pytest.mark.asyncio
+async def test_a_failure_is_logged_once_with_the_command_and_target(
+    server: fakeredis.FakeServer, caplog: pytest.LogCaptureFixture
+) -> None:
+    transport = fake_redis_cache(server).transport()
+    await _settle()
+    server.connected = False
+    caplog.clear()
+
+    with caplog.at_level(logging.WARNING), service_target("response_polling"):
+        failure = await transport.execute(Get("k"))
+
+    assert isinstance(failure, RedisFailure)
+    assert [record.getMessage() for record in caplog.records] == [
+        f"Redis get failed for response_polling: {failure.message}"
+    ]