Make the image-gen record/replay proxy report cache mode and per-request HIT/MISS

tests/_openai_record_replay_proxy.py

@@ -19,16 +19,25 @@
 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
 
 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 @@ async def handle(self, method: str, path: str, body: bytes, fetch_upstream: Fetc
         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 @@ def _cache_get(self, key: str) -> Optional[UpstreamResult]:
             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 @@ def _cache_set(self, key: str, status: int, headers: Headers, body: bytes) -> No
         )
         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 @@ async def fetch_upstream() -> UpstreamResult:
 
     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)