test(ci): make the image-gen record/replay proxy report cache mode and per-request HIT/MISS (#29802)

The recorder could come up pointed at a missing or unreachable cassette redis
and silently forward every request live; the health check still passed and the
process logged nothing, so a CI run looked identical whether it replayed from
the cassette or paid OpenAI for a fresh call every commit. There was no way to
tell from the logs whether the 24h caching was actually happening.

It now announces its mode at startup (REPLAY when the cassette redis is
reachable, PASSTHROUGH when CASSETTE_REDIS_URL is unset, DEGRADED when it is set
but the redis is unreachable) and logs a HIT/MISS line per request. _cache_set
returns whether the write landed so a mid-run redis failure surfaces as a
warning instead of masquerading as a successful record.

Adds unit tests covering the three startup modes and the HIT/MISS/not-recorded
request paths; both new behaviors were mutation-checked.

tests/_openai_record_replay_proxy.py

@@ -19,6 +19,11 @@ write, never refreshed on read. A recording therefore goes stale a day after
 capture and the next run past that point re-records live and catches provider
 contract drift, exactly matching the lapse-after-write contract in
 ``tests/_vcr_redis_persister.py``.
+
+The process logs its mode at startup (REPLAY when the cassette redis is
+reachable, PASSTHROUGH or DEGRADED otherwise) and a HIT/MISS line per request,
+so a CI run shows whether it served from the cassette or went live instead of
+silently degrading.
 """
 
 from __future__ import annotations
@@ -26,9 +31,13 @@ from __future__ import annotations
 import base64
 import hashlib
 import json
+import logging
 import os
 from typing import Awaitable, Callable, List, Optional, Tuple
 
+_LOGGER = logging.getLogger("openai_record_replay")
+_LOGGER.setLevel(logging.INFO)
+
 CASSETTE_TTL_SECONDS = 24 * 60 * 60
 RECORD_KEY_PREFIX = "litellm:openai:record:"
 RECORDER_REDIS_URL_ENV = "CASSETTE_REDIS_URL"
@@ -112,12 +121,21 @@ class OpenAIRecordReplay:
         key = self.record_key(method, path, body)
         cached = self._cache_get(key)
         if cached is not None:
+            _LOGGER.info("HIT replayed from cassette: %s %s", method, path)
             return cached
 
         status, headers, resp_body = await fetch_upstream()
         sanitized = _sanitize_headers(headers)
-        if 200 <= status < 300:
-            self._cache_set(key, status, sanitized, resp_body)
+        if not (200 <= status < 300):
+            _LOGGER.info("MISS forwarded live, not cached (status=%s): %s %s", status, method, path)
+        elif self._cache_set(key, status, sanitized, resp_body):
+            _LOGGER.info("MISS forwarded live and recorded: %s %s", method, path)
+        else:
+            _LOGGER.warning(
+                "MISS forwarded live but NOT recorded (redis unset or unreachable): %s %s",
+                method,
+                path,
+            )
         return status, sanitized, resp_body
 
     def _cache_get(self, key: str) -> Optional[UpstreamResult]:
@@ -138,9 +156,9 @@ class OpenAIRecordReplay:
             return None
         return status, headers, resp_body
 
-    def _cache_set(self, key: str, status: int, headers: Headers, body: bytes) -> None:
+    def _cache_set(self, key: str, status: int, headers: Headers, body: bytes) -> bool:
         if self._redis is None:
-            return
+            return False
         payload = json.dumps(
             {
                 "status": status,
@@ -150,8 +168,31 @@ class OpenAIRecordReplay:
         )
         try:
             self._redis.set(key, payload, ex=self._ttl_seconds)
+            return True
         except Exception:
-            pass
+            return False
+
+    def log_startup_mode(self) -> None:
+        if self._redis is None:
+            _LOGGER.warning(
+                "PASSTHROUGH: %s unset, every request goes live to %s and nothing is cached",
+                RECORDER_REDIS_URL_ENV,
+                self.upstream_base_url,
+            )
+            return
+        try:
+            self._redis.ping()
+        except Exception as exc:
+            _LOGGER.warning(
+                "DEGRADED to live: %s set but cassette redis unreachable (%s); nothing is cached",
+                RECORDER_REDIS_URL_ENV,
+                type(exc).__name__,
+            )
+            return
+        _LOGGER.info(
+            "REPLAY mode: cassette redis reachable, recordings expire %ss after write (no refresh on read)",
+            self._ttl_seconds,
+        )
 
 
 def _build_default_redis_client():
@@ -186,6 +227,7 @@ def create_app(recorder: Optional[OpenAIRecordReplay] = None, http_client=None):
 
     @contextlib.asynccontextmanager
     async def lifespan(_app):
+        recorder.log_startup_mode()
         try:
             yield
         finally:
@@ -231,6 +273,7 @@ if __name__ == "__main__":
 
     import uvicorn
 
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--host", default="0.0.0.0")
     parser.add_argument("--port", type=int, default=8090)

tests/llm_translation/test_openai_record_replay_proxy.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import sys
 
@@ -214,3 +215,53 @@ def test_app_lifespan_leaves_injected_http_client_open():
         pass
 
     assert client.closed is False
+
+
+def test_handle_logs_miss_then_hit(caplog):
+    """Each request self-reports so a CI run shows cassette vs live."""
+    recorder = _recorder()
+    upstream = _Upstream()
+    body_in = b'{"model":"gpt-image-1"}'
+
+    with caplog.at_level(logging.INFO, logger="openai_record_replay"):
+        _run(recorder.handle("POST", "/v1/images/generations", body_in, upstream))
+        _run(recorder.handle("POST", "/v1/images/generations", body_in, upstream))
+
+    messages = [r.getMessage() for r in caplog.records]
+    assert any("MISS forwarded live and recorded" in m for m in messages)
+    assert any("HIT replayed from cassette" in m for m in messages)
+
+
+def test_handle_warns_when_recording_not_persisted(caplog):
+    """A redis failure must surface loudly, not look like a successful record."""
+    recorder = _recorder(_BoomRedis())
+    upstream = _Upstream()
+
+    with caplog.at_level(logging.WARNING, logger="openai_record_replay"):
+        _run(recorder.handle("POST", "/v1/images/generations", b'{"model":"gpt-image-1"}', upstream))
+
+    assert any(r.levelno == logging.WARNING and "NOT recorded" in r.getMessage() for r in caplog.records)
+
+
+def test_log_startup_mode_distinguishes_replay_from_passthrough(caplog):
+    """Startup must announce whether the recorder will actually cache."""
+    with caplog.at_level(logging.INFO, logger="openai_record_replay"):
+        OpenAIRecordReplay(None).log_startup_mode()
+        _recorder().log_startup_mode()
+
+    emitted = [(r.levelno, r.getMessage()) for r in caplog.records]
+    assert any(lvl == logging.WARNING and "PASSTHROUGH" in m for lvl, m in emitted)
+    assert any(lvl == logging.INFO and "REPLAY mode" in m for lvl, m in emitted)
+
+
+class _UnreachableRedis:
+    def ping(self):
+        raise ConnectionError("redis offline")
+
+
+def test_log_startup_mode_warns_when_redis_configured_but_unreachable(caplog):
+    """A configured-but-dead redis must warn, not look like it will cache."""
+    with caplog.at_level(logging.WARNING, logger="openai_record_replay"):
+        _recorder(_UnreachableRedis()).log_startup_mode()
+
+    assert any(r.levelno == logging.WARNING and "DEGRADED" in r.getMessage() for r in caplog.records)