fix: registration owns the shadow-eval lifecycle loop; populate verdict row's real-side columns

litellm/integrations/shadow_eval_logger.py

@@ -179,12 +179,13 @@ class _CallFailure:
 
 @dataclass(frozen=True, slots=True)
 class _ShadowResponse:
-    """A successful shadow call: (text, routed model, tier, completion tokens)."""
+    """A successful shadow call, with the ids and token counts the verdict row records."""
 
     text: str
     model: str
     tier: str | None
     completion_tokens: int | None
+    request_id: str | None
 
 
 @dataclass(frozen=True, slots=True)
@@ -320,10 +321,20 @@ class ShadowEvalLogger(CustomLogger):
         self._lifecycle_task: asyncio.Task[None] | None = None
 
     def start_lifecycle_loop(self) -> None:
-        """Idempotently start the periodic loop that owns job lifecycle off the request path."""
+        """Idempotently start the periodic loop that owns job lifecycle off the request path.
+
+        Without the loop, jobs never finalize and counters never flush, so a caller
+        registering this logger outside a running event loop gets a warning rather
+        than a silent no-op.
+        """
         if self._lifecycle_task is not None and not self._lifecycle_task.done():
             return
-        self._lifecycle_task = asyncio.create_task(self._lifecycle_loop())
+        try:
+            self._lifecycle_task = asyncio.create_task(self._lifecycle_loop())
+        except RuntimeError:
+            verbose_logger.warning(
+                "shadow_eval: no running event loop; lifecycle loop not started, jobs will not finalize on this process"
+            )
 
     async def _lifecycle_loop(self) -> None:
         while True:
@@ -406,6 +417,7 @@ class ShadowEvalLogger(CustomLogger):
                     else (),
                     response_obj=response_obj,
                     real_model=payload.get("model") or "",
+                    real_response_tokens=payload.get("completion_tokens"),
                     model_parameters=MappingProxyType(
                         dict(payload.get("model_parameters") or {})  # mutable-ok: frozen snapshot
                     ),
@@ -507,6 +519,7 @@ class ShadowEvalLogger(CustomLogger):
         messages: Sequence[Mapping[str, object]],
         response_obj: object,
         real_model: str,
+        real_response_tokens: int | None,
         model_parameters: Mapping[str, object],
         parent_metadata: Mapping[str, object],
         budget_metadata: Mapping[str, object] = _EMPTY_METADATA,
@@ -547,8 +560,10 @@ class ShadowEvalLogger(CustomLogger):
                 data={  # mutable-ok: Prisma payload
                     "job_id": job.id,
                     "request_id": request_id,
+                    "shadow_request_id": shadow.request_id,
                     "tier_classification": shadow.tier,
                     "real_model": real_model,
+                    "real_response_tokens": real_response_tokens,
                     "shadow_model": shadow.model,
                     "shadow_response_tokens": shadow.completion_tokens,
                     "judge_preference": verdict.preference,
@@ -631,11 +646,13 @@ class ShadowEvalLogger(CustomLogger):
         raw_tier: Final = routing_decision.get("tier_label") or routing_decision.get("tier")
         usage: Final = getattr(response, "usage", None)
         raw_tokens: Final = getattr(usage, "completion_tokens", None) if usage is not None else None
+        raw_id: Final = getattr(response, "id", None)
         return _ShadowResponse(
             text=text,
             model=str(getattr(response, "model", None) or routing_decision.get("routed_model") or ""),
             tier=str(raw_tier) if raw_tier is not None else None,
             completion_tokens=int(raw_tokens) if isinstance(raw_tokens, int) else None,
+            request_id=str(raw_id) if raw_id else None,
         )
 
     async def _call_judge(

litellm/proxy/proxy_server.py

@@ -1184,8 +1184,6 @@ async def proxy_startup_event(app: FastAPI):
                 _tagged.strategy._state_loaded = True
     asyncio.create_task(_adaptive_router_flusher_loop())
 
-    _start_shadow_eval_lifecycle_loop()
-
     ## [Optional] Initialize dd tracer
     ProxyStartupEvent._init_dd_tracer()
 
@@ -2217,27 +2215,18 @@ def cost_tracking():
 
 
 def _register_shadow_eval_logger() -> None:
-    """Register the shadow-eval success hook.
+    """Register the shadow-eval success hook and start its lifecycle loop.
 
     Cheap when idle: with no active LiteLLM_ShadowEvalJob rows the hook is one
-    cached dict lookup per request. Registered alongside cost tracking because
-    it has the same hard dependency on prisma_client. Its lifecycle loop is
-    started separately at proxy startup, once an event loop exists.
+    dict lookup per request. Registered alongside cost tracking because it has
+    the same hard dependency on prisma_client. Registration owns the loop start
+    so the two cannot be reordered apart; the loop is started on whichever
+    instance is actually registered, since the callback manager dedups by class
+    and discards a second instance.
     """
     from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
 
     litellm.logging_callback_manager.add_litellm_callback(ShadowEvalLogger())
-
-
-def _start_shadow_eval_lifecycle_loop() -> None:
-    """Start the registered shadow-eval logger's periodic lifecycle loop.
-
-    The loop owns counter flushes, snapshot refreshes, and finishing jobs whose
-    window or spend cap has passed, so a job on a key that goes quiet still ends
-    on schedule with its final counters written.
-    """
-    from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
-
     for callback in litellm.callbacks:
         if isinstance(callback, ShadowEvalLogger):
             callback.start_lifecycle_loop()

tests/test_litellm/integrations/test_shadow_eval_logger.py

@@ -362,7 +362,7 @@ class TestStoppedJobCannotBeReactivated:
 
         logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma)
         logger._call_router_shadow = AsyncMock(
-            return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10)
+            return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1")
         )
         logger._call_judge = AsyncMock(
             return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01)
@@ -375,6 +375,7 @@ class TestStoppedJobCannotBeReactivated:
             messages=[{"role": "user", "content": "hi"}],
             response_obj={"choices": [{"message": {"content": "real text"}}]},
             real_model="gpt-4o",
+            real_response_tokens=42,
             model_parameters={},
             parent_metadata={},
         )
@@ -387,6 +388,44 @@ class TestStoppedJobCannotBeReactivated:
         assert call_kwargs["data"]["status"] == "running"
 
 
+@pytest.mark.asyncio
+class TestVerdictRowRecordsBothSides:
+    async def test_verdict_row_carries_ids_and_token_counts_for_both_arms(self):
+        """real_response_tokens next to shadow_response_tokens is what lets a reader
+        check the judge's verdicts for verbosity bias, and shadow_request_id joins the
+        verdict to the shadow call's own spend log for drill-down. Declared columns
+        that are never written are worse than absent ones."""
+        prisma = MagicMock()
+        prisma.db.litellm_shadowevalverdict.create = AsyncMock()
+        prisma.db.litellm_shadowevaljob.update_many = AsyncMock()
+        logger = ShadowEvalLogger(router_provider=_router_mock, prisma_provider=lambda: prisma)
+        logger._call_router_shadow = AsyncMock(
+            return_value=_ShadowResponse(
+                text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1"
+            )
+        )
+        logger._call_judge = AsyncMock(
+            return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01)
+        )
+
+        job = ActiveShadowEvalJob(id="j1", router_name="r", judge_model="m", shadow_percentage=100.0, status="running")
+        await logger._run_shadow_eval(
+            job=job,
+            request_id="req-1",
+            messages=[{"role": "user", "content": "hi"}],
+            response_obj={"choices": [{"message": {"content": "real text"}}]},
+            real_model="gpt-4o",
+            real_response_tokens=42,
+            model_parameters={},
+            parent_metadata={},
+        )
+
+        data = prisma.db.litellm_shadowevalverdict.create.call_args.kwargs["data"]
+        assert data["shadow_request_id"] == "shadow-req-1"
+        assert data["real_response_tokens"] == 42
+        assert data["shadow_response_tokens"] == 10
+
+
 @pytest.mark.asyncio
 class TestVerdictWriteAccumulatesCost:
     async def test_cost_actual_uses_increment_not_a_raw_set(self):
@@ -403,7 +442,7 @@ class TestVerdictWriteAccumulatesCost:
 
         logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma)
         logger._call_router_shadow = AsyncMock(
-            return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10)
+            return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1")
         )
         logger._call_judge = AsyncMock(
             return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.05)
@@ -416,6 +455,7 @@ class TestVerdictWriteAccumulatesCost:
             messages=[{"role": "user", "content": "hi"}],
             response_obj={"choices": [{"message": {"content": "real text"}}]},
             real_model="gpt-4o",
+            real_response_tokens=42,
             model_parameters={},
             parent_metadata={},
         )
@@ -534,7 +574,7 @@ class TestSubCallsAreAttributedToTheShadowedKey:
         logger, _, _ = _logger_with_mocks(job)
         logger._prisma_provider = lambda: prisma
         logger._call_router_shadow = AsyncMock(
-            return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10)
+            return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1")
         )
         logger._call_judge = AsyncMock(
             return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01)
@@ -1109,6 +1149,7 @@ class TestJudgeFailureModes:
             messages=({"role": "user", "content": "hi"},),
             response_obj={"choices": [{"message": {"content": "real says"}}]},
             real_model="gpt-4o",
+            real_response_tokens=42,
             model_parameters={},
             parent_metadata={},
         )

tests/test_litellm/proxy/proxy_server/test_lifecycle.py

@@ -512,19 +512,45 @@ def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch):
 
 
 @pytest.mark.asyncio
-async def test_startup_starts_the_shadow_eval_lifecycle_loop(monkeypatch):
-    """The loop owns counter flushes and finishing expired jobs, so a job on a key
-    that goes quiet still ends on schedule; it must be running after startup."""
+async def test_registration_starts_the_lifecycle_loop_on_the_registered_instance(monkeypatch):
+    """Registration owns the loop start, so there is no startup ordering to get
+    wrong, and the start lands on the instance the callback manager kept, which
+    is the ORIGINAL one when its class dedup discards a re-registration."""
     import litellm
     from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
 
-    logger = ShadowEvalLogger(router_provider=lambda: None, prisma_provider=lambda: None)
-    monkeypatch.setattr(litellm, "callbacks", [logger], raising=False)
+    fake_prisma = MagicMock()
+    monkeypatch.setattr(ps, "prisma_client", fake_prisma, raising=False)
+    monkeypatch.setattr(litellm, "callbacks", [], raising=False)
+    monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False)
 
-    ps._start_shadow_eval_lifecycle_loop()
+    cost_tracking()
+    cost_tracking()
 
-    assert logger._lifecycle_task is not None and not logger._lifecycle_task.done()
-    logger._lifecycle_task.cancel()
+    loggers = [cb for cb in litellm.callbacks if isinstance(cb, ShadowEvalLogger)]
+    assert len(loggers) == 1
+    task = loggers[0]._lifecycle_task
+    assert task is not None and not task.done()
+    task.cancel()
+
+
+def test_cost_tracking_twice_registers_one_shadow_eval_logger(monkeypatch):
+    """cost_tracking can run again (config reload); the callback manager dedups
+    CustomLogger instances by class, so a second call must not produce a second
+    ShadowEvalLogger with its own snapshot, buffer, and in-flight counter."""
+    import litellm
+    from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
+
+    fake_prisma = MagicMock()
+    monkeypatch.setattr(ps, "prisma_client", fake_prisma, raising=False)
+    monkeypatch.setattr(litellm, "callbacks", [], raising=False)
+    monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False)
+
+    cost_tracking()
+    cost_tracking()
+
+    shadow_loggers = [cb for cb in litellm.callbacks if isinstance(cb, ShadowEvalLogger)]
+    assert len(shadow_loggers) == 1
 
 
 def test_cost_tracking_no_op_when_prisma_missing(monkeypatch):