test(router): compare both backends' Redis writes against a live redis-server

Each scenario runs the same config and failures through PythonRouter and the Rust backend
against one redis-server and compares every key, raw stored value (write time masked) and
TTL: deployment cooldowns, allowed_fails counters and failure RPM keys. A Rust router also
honors cooldowns a Python router wrote. Skipped when redis-server is not installed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Aw1xktpUnix5tWhvSzey8t

litellm-rust/crates/router/POC.md

@@ -84,12 +84,12 @@ Later: other strategies, routing groups, filters (aliases, patterns, access grou
 ## Log
 
 - 2026-10-06: design agreed, branch created
-- 2026-10-07: steps 2, 3 and 4 landed, facade verification running
+- 2026-10-07: facade verified, steps 2, 3 and 4 landed, step 5 started (parametrized suites, Redis parity)
 
 ## Checkpoint 2026-10-07
 
 Done:
-- Step 1 verification: in progress (tests on branch vs its base `cb76270b`, then `make check`). Found so far: an isort failure in `python_router.py` and LIT010/LIT008 in `router.py`, both fixed
+- Step 1 verification: done. tests/unit, tests/router_unit_tests and tests/local_testing/test_router_fallbacks.py were run on the branch and on its base `cb76270b`; the branch-only failures were the facade identity issue (fixed in `71120e1f`) and two router_unit_tests that patched the facade instead of the backend (fixed). `make check` found an isort issue and an unformatted `python_router.py`, LIT010/LIT008 and four ANN204 in `router.py`, and a moved LIT013 suppression, all fixed. In this container `make check`'s dependency sync and the basedpyright gate cannot provision (a private git dependency returns 403) and the dashboard has no node_modules, so those steps were run per file instead
 - Step 2: `Route.ROUTER` (`RUST_OPT_IN`) in `litellm/rust_bridge/catalog.py`. `litellm/router_backends/selection.py::select_backend` runs in `Router.__init__`; `rust_support.py` decides what the Rust backend can serve (declines to `PythonRouter` under opt-in, raises under `RUST_REQUIRED`). `RustRouter` is still a stub, so everything declines
 - Step 3: the Rust crate. `engine::Engine::route` runs fallbacks -> retries -> attempt against a `host::RouterHost` (`invoke`, `sleep`). Modules: `snapshot` (registry), `settings`, `failure` (`Classified` record, `Rejection` for router-raised errors), `selection` (resolve, cooldown filter, retry skip, `simple_shuffle`), `retry` (`should_retry_this_error`, backoff), `fallback` (chain lookup), `cooldown` (failure callbacks), `store` (memory + Redis with Python's encodings), `pyrepr` (Python `str()` of dicts), `random` (CPython MT19937), `operation` (`Operation`, `Dispatch`, all `Invoke` today)
 
@@ -105,6 +105,8 @@ Design notes made while implementing:
 - `litellm/router_backends/rust_call.py::RoutedCall` is the per-call driver: it mirrors `_acompletion`/`_completion`'s body after selection (same `PythonRouter` helpers through a typed `AttemptRouter` protocol), classifies exceptions, and applies the router's ops to numbered metadata buckets
 - Tests: `tests/unit/router_backends/test_rust_call.py` (classification, buckets, rejections, debug text), `tests/test_litellm_rust/router/test_rust_router.py` (native; same config and seed through both backends: identical pick sequences, headers, error text, cooldown-then-reject sequence, sync path, facade selection)
 
-Next: step 5. Parametrize the existing `Router(...)` behavior tests over both backends (the Rust side's `NotImplementedError`s and mismatches become the to-do list), and grow the differential harness: scripted failures per deployment, a live `redis-server` shared by both backends and compared key by key (values, TTLs). Then streaming (buffer to the first content chunk in the driver) and the generic operations (`aresponses`, `aanthropic_messages`)
+- Step 5 (in progress): `litellm/router_backends/selection.py::pinned_backend` fixes the backend decision; `tests/unit/router_backends/backends.py::router_backend` runs an opted-in module on both backends (Rust pinned to `RUST_REQUIRED`), with known Rust-side failures as strict xfails in `RUST_GAPS`. Opted in so far: `tests/unit/test_router_per_deployment_num_retries.py`, `tests/unit/test_router_exception_redaction.py`. `tests/test_litellm_rust/router/test_redis_parity.py` runs both backends against a live `redis-server` and compares every key, raw value (timestamp masked) and TTL: cooldowns, `allowed_fails` counters and failure RPM keys match byte for byte, and a Rust router honors cooldowns a Python router wrote
+
+Next: opt more suites into `router_backend` (`tests/unit/test_router_order_fallback.py` and the fallback/cooldown tests in `tests/unit/router_utils` that go through `Router(...)`), then streaming (the driver buffers to the first content chunk and reports `StreamFailedBeforeContent`; decide whether chat keeps Python's fallback-only rule), then the generic operations (`aresponses`, `aanthropic_messages`, `aimage_generation`), then `upsert_deployment`/`delete_deployment`/`update_settings` through `Registry::replace`
 
 Known gaps to close or decline: chat streaming pre-first-chunk failures (Python goes straight to fallbacks, no in-group retry), `include_fallback_errors`, request-level `model_group_retry_policy`/`specific_deployment`/`priority`, CustomLogger fallback-event hooks, `router_cooldown_event_callback`, the 200-key `InMemoryCache` eviction, fallback keys containing `/` (provider-prefixed matching needs the cost map)

tests/test_litellm_rust/router/test_redis_parity.py

@@ -0,0 +1,149 @@
+"""Both router backends write the same Redis keys, values and TTLs for the same failures."""
+
+from __future__ import annotations
+
+import random
+import re
+import shutil
+import socket
+import subprocess
+import time
+from collections.abc import Generator, Mapping, Sequence
+from typing import Final
+
+import pytest
+import redis
+
+import litellm
+from litellm.router_backends.python_router import PythonRouter
+from litellm.router_backends.rust_router import NATIVE_ROUTER, RustRouter
+from litellm.types.router import RouterRateLimitError
+
+pytestmark = pytest.mark.requires_rust_extension
+
+MESSAGES: Final = ({"role": "user", "content": "hi"},)
+_MINUTE: Final = re.compile(r":\d{2}-\d{2}$")
+_TIMESTAMP: Final = re.compile(r"'timestamp': [0-9.e+-]+")
+
+
+@pytest.fixture(scope="module")
+def redis_port() -> Generator[int]:
+    server: Final = shutil.which("redis-server")
+    if server is None:
+        pytest.skip("redis-server is not installed")
+    with socket.socket() as probe:
+        probe.bind(("127.0.0.1", 0))
+        port: Final = int(probe.getsockname()[1])
+    process: Final = subprocess.Popen(
+        [server, "--port", str(port), "--save", "", "--appendonly", "no"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    client: Final = redis.Redis(port=port)
+    deadline: Final = time.monotonic() + 10
+    while time.monotonic() < deadline:
+        try:
+            client.ping()
+            break
+        except redis.ConnectionError:
+            time.sleep(0.05)
+    yield port
+    process.terminate()
+    process.wait(timeout=10)
+
+
+@pytest.fixture
+def store(redis_port: int) -> redis.Redis:
+    client: Final = redis.Redis(port=redis_port, decode_responses=True)
+    client.flushdb()
+    return client
+
+
+def _deployment(name: str, response: str) -> dict[str, object]:  # mutable-ok: PythonRouter consumes model_list dicts
+    return {
+        "model_name": "g",
+        "litellm_params": {"model": f"openai/{name}", "api_key": "k", "mock_response": response},
+        "model_info": {"id": name},
+    }
+
+
+def _snapshot(client: redis.Redis) -> Mapping[str, tuple[str, int]]:
+    """Every key with its decoded value and TTL; volatile parts (timestamps, the UTC minute) normalized."""
+    keys: Final = sorted(str(key) for key in client.keys("*"))
+    return {_MINUTE.sub(":<minute>", key): (_stable(str(client.get(key))), int(client.ttl(key))) for key in keys}
+
+
+def _stable(raw: str) -> str:
+    """The stored text as is, with only the write time masked, so encoding differences still show."""
+    return _TIMESTAMP.sub("'timestamp': <time>", raw)
+
+
+async def _run(backend: object, calls: int) -> Sequence[str]:
+    outcomes: Final[list[str]] = []  # mutable-ok: one outcome per call
+    for _ in range(calls):
+        try:
+            await backend.acompletion("g", MESSAGES)  # pyright: ignore[reportAttributeAccessIssue]  # both backends serve acompletion
+            outcomes.append("ok")
+        except RouterRateLimitError:
+            outcomes.append("rejected")
+        except litellm.RateLimitError:
+            outcomes.append("rate_limited")
+        except litellm.InternalServerError:
+            outcomes.append("server_error")
+    return outcomes
+
+
+def _arguments(
+    port: int, responses: Sequence[str], extra: Mapping[str, object]
+) -> dict[str, object]:  # mutable-ok: Router(...) keyword arguments
+    return {
+        "model_list": [_deployment(f"d{index}", response) for index, response in enumerate(responses)],
+        "redis_host": "127.0.0.1",
+        "redis_port": port,
+        "num_retries": 0,
+        **extra,
+    }
+
+
+@pytest.mark.parametrize(
+    ("responses", "extra", "calls"),
+    (
+        pytest.param(("litellm.RateLimitError", "litellm.RateLimitError"), {}, 3, id="rate-limit-cooldowns"),
+        pytest.param(
+            ("litellm.InternalServerError",), {"allowed_fails": 1, "cooldown_time": 30}, 3, id="allowed-fails"
+        ),
+        pytest.param(("litellm.InternalServerError", "ok"), {}, 4, id="failure-usage-only"),
+    ),
+)
+async def test_both_backends_leave_the_same_redis_state(
+    store: redis.Redis, redis_port: int, responses: Sequence[str], extra: Mapping[str, object], calls: int
+) -> None:
+    arguments: Final = _arguments(redis_port, responses, extra)
+    random.seed(0)
+    python_router: Final = PythonRouter(**arguments)
+    python_outcomes: Final = await _run(python_router, calls)
+    python_router.discard()
+    python_state: Final = _snapshot(store)
+    store.flushdb()
+
+    native: Final = NATIVE_ROUTER.load()
+    assert native is not None
+    rust_outcomes: Final = await _run(RustRouter(_arguments(redis_port, responses, extra), native, 0), calls)
+    rust_state: Final = _snapshot(store)
+
+    assert rust_outcomes == python_outcomes
+    assert python_state
+    assert rust_state == python_state
+
+
+async def test_a_rust_router_honors_cooldowns_a_python_router_wrote(store: redis.Redis, redis_port: int) -> None:
+    responses: Final = ("litellm.RateLimitError", "litellm.RateLimitError")
+    python_router: Final = PythonRouter(**_arguments(redis_port, responses, {}))
+    assert await _run(python_router, 2) == ["rate_limited", "rate_limited"]
+    python_router.discard()
+    native: Final = NATIVE_ROUTER.load()
+    assert native is not None
+
+    rust_outcomes: Final = await _run(RustRouter(_arguments(redis_port, responses, {}), native, 0), 1)
+
+    assert rust_outcomes == ["rejected"]