refactor(redis-transport): add RedisTransport with a redis-py backend

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -0,0 +1,17 @@
+---
+name: redis-access
+description: Add Redis operations or move raw redis-py clients and scripts behind RedisTransport 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`). Read `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
+
+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`
+- Keys and values in Redis are a wire format shared with older pods and Rust. A move keeps them byte for byte
+- A `transport` failure on a write is an unknown outcome, never "not applied"
+- Do only what was asked. Confirm before any Rust work

.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/_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,11 @@
+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

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}"
+    ]