fix(proxy): dedup latest health checks in SQL and gate the DB save per window

litellm/constants.py

@@ -1619,6 +1619,7 @@ CLOUDZERO_EXPORT_USAGE_DATA_JOB_NAME: Final = "cloudzero_export_usage_data"
 MAVVRIK_FOCUS_EXPORT_JOB_NAME: Final = "mavvrik_focus_export_usage_data"
 CLOUDZERO_MAX_FETCHED_DATA_RECORDS: Final = int(os.getenv("CLOUDZERO_MAX_FETCHED_DATA_RECORDS", 50000))
 SPEND_LOG_CLEANUP_JOB_NAME: Final = "spend_log_cleanup"
+BACKGROUND_HEALTH_CHECK_DB_SAVE_JOB_NAME: Final = "background_health_check_db_save"
 KEY_ROTATION_JOB_NAME: Final = "litellm_key_rotation_job"
 EXPIRED_UI_SESSION_KEY_CLEANUP_JOB_NAME: Final = "litellm_expired_ui_session_key_cleanup_job"
 WEEKLY_SPEND_REPORT_JOB_ID: Final = "weekly_spend_report_job"

litellm/proxy/db/health_check_latest.py

@@ -0,0 +1,96 @@
+"""
+Latest health-check row per model, deduplicated by Postgres.
+
+prisma-client-py's ``find_many(distinct=...)`` dedups client-side: the emitted
+SQL carries no DISTINCT, so the whole append-only history table streams to the
+worker on every call. ``SELECT DISTINCT ON`` keeps the transfer at one row per
+(model_id, model_name) and is served by the matching descending index.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Final
+
+from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, field_validator
+
+from litellm._logging import verbose_proxy_logger
+
+if TYPE_CHECKING:
+    from litellm.proxy.utils import PrismaClient
+
+LATEST_HEALTH_CHECKS_SQL: Final = """
+SELECT DISTINCT ON ("model_id", "model_name")
+       "health_check_id", "model_name", "model_id", "status",
+       "healthy_count", "unhealthy_count", "error_message",
+       "response_time_ms", "details", "checked_by",
+       "checked_at", "created_at", "updated_at"
+FROM "LiteLLM_HealthCheckTable"
+ORDER BY "model_id" ASC, "model_name" ASC, "checked_at" DESC
+"""
+
+LATEST_HEALTH_CHECKS_FOR_MODELS_SQL: Final = """
+SELECT DISTINCT ON ("model_id", "model_name")
+       "health_check_id", "model_name", "model_id", "status",
+       "healthy_count", "unhealthy_count", "error_message",
+       "response_time_ms", "details", "checked_by",
+       "checked_at", "created_at", "updated_at"
+FROM "LiteLLM_HealthCheckTable"
+WHERE "model_name" = ANY($1)
+ORDER BY "model_id" ASC, "model_name" ASC, "checked_at" DESC
+"""
+
+
+class LatestHealthCheckRow(BaseModel):
+    model_config = ConfigDict(frozen=True, protected_namespaces=())
+
+    health_check_id: str
+    model_name: str
+    model_id: str | None = None
+    status: str
+    healthy_count: int = 0
+    unhealthy_count: int = 0
+    error_message: str | None = None
+    response_time_ms: float | None = None
+    details: JsonValue | None = None
+    checked_by: str | None = None
+    checked_at: datetime
+    created_at: datetime
+    updated_at: datetime
+
+    @field_validator("details", mode="before")
+    @classmethod
+    def _decode_json_text(cls, value: object) -> object:
+        return json.loads(value) if isinstance(value, str) else value
+
+    @field_validator("checked_at", "created_at", "updated_at")
+    @classmethod
+    def _assume_utc(cls, value: datetime) -> datetime:
+        return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value
+
+
+_ROWS_ADAPTER: Final = TypeAdapter(tuple[LatestHealthCheckRow, ...])
+
+
+async def fetch_latest_health_checks(prisma_client: PrismaClient) -> tuple[LatestHealthCheckRow, ...]:
+    try:
+        rows: Final = await prisma_client.db.query_raw(LATEST_HEALTH_CHECKS_SQL)
+        return _ROWS_ADAPTER.validate_python(rows)
+    except Exception as query_err:  # noqa: BLE001  # health decorates other reads; a driver error must not fail them
+        verbose_proxy_logger.error("Error getting all latest health checks: %s", query_err)
+        return ()
+
+
+async def fetch_latest_health_checks_for_models(
+    prisma_client: PrismaClient, model_names: Sequence[str]
+) -> tuple[LatestHealthCheckRow, ...]:
+    if not model_names:
+        return ()
+    try:
+        rows: Final = await prisma_client.db.query_raw(LATEST_HEALTH_CHECKS_FOR_MODELS_SQL, list(model_names))
+        return _ROWS_ADAPTER.validate_python(rows)
+    except Exception as query_err:  # noqa: BLE001  # a paged model list must not fail on its health decoration
+        verbose_proxy_logger.error("Error getting latest health checks for models: %s", query_err)
+        return ()

litellm/proxy/proxy_server.py

@@ -16,7 +16,16 @@ import threading
 import time
 import traceback
 import warnings
-from collections.abc import AsyncGenerator, AsyncIterator, Callable, Collection, Mapping, MutableMapping, Sequence
+from collections.abc import (
+    AsyncGenerator,
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Collection,
+    Mapping,
+    MutableMapping,
+    Sequence,
+)
 from datetime import datetime, timedelta, timezone
 from types import MappingProxyType, UnionType
 from typing import (
@@ -50,6 +59,7 @@ from litellm.constants import (
     AIOHTTP_NEEDS_CLEANUP_CLOSED,
     AIOHTTP_TTL_DNS_CACHE,
     AUDIO_SPEECH_CHUNK_SIZE,
+    BACKGROUND_HEALTH_CHECK_DB_SAVE_JOB_NAME,
     BASE_MCP_ROUTE,
     DAILY_TAG_SPEND_BATCH_MULTIPLIER,
     DEFAULT_MAX_RECURSE_DEPTH,
@@ -224,7 +234,7 @@ def generate_feedback_box():
 import contextlib
 from collections import defaultdict
 from contextlib import asynccontextmanager
-from functools import lru_cache
+from functools import lru_cache, partial
 
 import litellm
 import litellm._redis
@@ -402,6 +412,7 @@ from litellm.proxy.config_resolvers.alerting import (
 )
 from litellm.proxy.container_endpoints.endpoints import router as container_router
 from litellm.proxy.credential_endpoints.endpoints import router as credential_router
+from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
 from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import (
     SPEND_LOG_CLEANUP_BOUND_SETTINGS,
     SpendLogCleanup,
@@ -3580,12 +3591,35 @@ async def _run_direct_health_check_with_instrumentation(
     raise AssertionError("perform_health_check rejected every optional argument")
 
 
+async def _window_gated_health_check_db_save(
+    save: Callable[[], Awaitable[None]],
+    pod_lock_manager: PodLockManager | None,
+    lock_ttl: int | None,
+) -> None:
+    """
+    Persist at most once per window fleet-wide: the lock is the "this window's save is
+    done" marker, so it is deliberately never released and expires with the interval.
+    """
+    if pod_lock_manager is not None and pod_lock_manager.redis_cache is not None:
+        acquired: Final = await pod_lock_manager.acquire_lock(
+            cronjob_id=BACKGROUND_HEALTH_CHECK_DB_SAVE_JOB_NAME,
+            ttl=lock_ttl,
+            allow_reentrant=False,
+        )
+        if not acquired:
+            verbose_proxy_logger.debug("background_health_check_db_save_skipped another pod persisted this window")
+            return
+    await save()
+
+
 def _schedule_background_health_check_db_save(
     prisma_client,
     shared_health_manager,
     model_list: list,
     healthy_endpoints: list,
     unhealthy_endpoints: list,
+    pod_lock_manager: PodLockManager | None = None,
+    lock_ttl: int | None = None,
 ):
     """Fire-and-forget: persist health check results to DB if prisma is available."""
     if prisma_client is None:
@@ -3598,16 +3632,16 @@ def _schedule_background_health_check_db_save(
 
     checked_by: Final = shared_health_manager.pod_id if shared_health_manager is not None else "background_health_check"
     start_time: Final = time_module.time()
-    asyncio.create_task(
-        _save_background_health_checks_to_db(
-            prisma_client,
-            model_list,
-            healthy_endpoints,
-            unhealthy_endpoints,
-            start_time,
-            checked_by=checked_by,
-        )
+    save: Final = partial(
+        _save_background_health_checks_to_db,
+        prisma_client,
+        model_list,
+        healthy_endpoints,
+        unhealthy_endpoints,
+        start_time,
+        checked_by=checked_by,
     )
+    asyncio.create_task(_window_gated_health_check_db_save(save, pod_lock_manager, lock_ttl))
 
 
 def _get_endpoint_exception_status(endpoint: dict, exceptions: dict) -> int:
@@ -3914,6 +3948,8 @@ async def _run_background_health_check():
             _llm_model_list,
             healthy_endpoints,
             unhealthy_endpoints,
+            pod_lock_manager=proxy_logging_obj.db_spend_update_writer.pod_lock_manager,
+            lock_ttl=health_check_interval,
         )
 
         # Write health state to router cache for health-check-driven routing

litellm/proxy/utils.py

@@ -121,6 +121,11 @@ from litellm.proxy.db.exception_handler import (
     PrismaDBExceptionHandler,
     call_with_db_reconnect_retry,
 )
+from litellm.proxy.db.health_check_latest import (
+    LatestHealthCheckRow,
+    fetch_latest_health_checks,
+    fetch_latest_health_checks_for_models,
+)
 from litellm.proxy.db.log_db_metrics import log_db_metrics
 from litellm.proxy.db.prisma_client import (
     PrismaWrapper,
@@ -5962,48 +5967,13 @@ class PrismaClient:
             verbose_proxy_logger.error("Error getting health check history: %s", e)
             return []
 
-    async def get_all_latest_health_checks(self) -> "Sequence[prisma_models.LiteLLM_HealthCheckTable]":
-        """
-        Get the latest health check for each model.
+    async def get_all_latest_health_checks(self) -> tuple[LatestHealthCheckRow, ...]:
+        """Latest health check per (model_id, model_name), deduplicated in Postgres."""
+        return await fetch_latest_health_checks(self)
 
-        Uses DB-level DISTINCT ON (model_id, model_name) with ORDER BY checked_at DESC
-        (via Prisma ``distinct`` + ``order``) so we never load the full history into memory.
-        """
-        try:
-            return await HealthCheckRepository(self).table.find_many(
-                distinct=["model_id", "model_name"],
-                order=[
-                    {"model_id": "asc"},
-                    {"model_name": "asc"},
-                    {"checked_at": "desc"},
-                ],
-            )
-        except Exception as e:
-            verbose_proxy_logger.error("Error getting all latest health checks: %s", e)
-            return []
-
-    async def get_latest_health_checks_for_models(
-        self, model_names: "Sequence[str]"
-    ) -> "Sequence[prisma_models.LiteLLM_HealthCheckTable]":
-        """
-        Get the latest health check for each of the named models.
-
-        Same DISTINCT ON as ``get_all_latest_health_checks``, bounded to the models asked
-        about, so a paged caller reads health for its page instead of for the whole table.
-        """
-        if not model_names:
-            return ()
-        latest_first: Final = (("model_id", "asc"), ("model_name", "asc"), ("checked_at", "desc"))
-        order: Final = [{field: direction} for field, direction in latest_first]  # mutable-ok: prisma order is a list
-        try:
-            return await HealthCheckRepository(self).table.find_many(
-                where={"model_name": {"in": list(model_names)}},  # mutable-ok: prisma filters are dicts and lists
-                distinct=["model_id", "model_name"],  # mutable-ok: prisma distinct takes a list
-                order=order,
-            )
-        except Exception as e:  # noqa: BLE001  # health decorates a list; a driver error must not fail the page
-            verbose_proxy_logger.error("Error getting latest health checks for models: %s", e)
-            return ()
+    async def get_latest_health_checks_for_models(self, model_names: Sequence[str]) -> tuple[LatestHealthCheckRow, ...]:
+        """Same as ``get_all_latest_health_checks``, bounded to the named models."""
+        return await fetch_latest_health_checks_for_models(self, model_names)
 
 
 ### HELPER FUNCTIONS ###

tests/test_litellm/proxy/db/test_health_check_latest.py

@@ -0,0 +1,111 @@
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from litellm.proxy.db.health_check_latest import (
+    LATEST_HEALTH_CHECKS_FOR_MODELS_SQL,
+    LATEST_HEALTH_CHECKS_SQL,
+    fetch_latest_health_checks,
+    fetch_latest_health_checks_for_models,
+)
+
+
+def _prisma(rows):
+    prisma = MagicMock()
+    prisma.db.query_raw = AsyncMock(return_value=rows)
+    return prisma
+
+
+def _raw_row(**overrides):
+    row = {
+        "health_check_id": "hc-1",
+        "model_name": "gpt-4",
+        "model_id": "deployment-abc",
+        "status": "healthy",
+        "healthy_count": 1,
+        "unhealthy_count": 0,
+        "error_message": None,
+        "response_time_ms": 12.5,
+        "details": None,
+        "checked_by": "pod-1",
+        "checked_at": "2026-08-25T00:00:00+00:00",
+        "created_at": "2026-08-25T00:00:00+00:00",
+        "updated_at": "2026-08-25T00:00:00+00:00",
+    }
+    return {**row, **overrides}
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_runs_one_distinct_on_query_with_no_parameters():
+    """The dedup must be in the SQL: prisma find_many(distinct=...) streams the whole history table."""
+    prisma = _prisma([])
+    assert await fetch_latest_health_checks(prisma) == ()
+    assert prisma.db.query_raw.await_args.args == (LATEST_HEALTH_CHECKS_SQL,)
+    assert 'DISTINCT ON ("model_id", "model_name")' in LATEST_HEALTH_CHECKS_SQL
+    assert '"checked_at" DESC' in LATEST_HEALTH_CHECKS_SQL
+
+
+@pytest.mark.asyncio
+async def test_raw_datetimes_come_back_tz_aware_with_or_without_an_offset():
+    """The save path subtracts checked_at from datetime.now(timezone.utc); a naive value would TypeError."""
+    naive = _raw_row(health_check_id="hc-naive", model_id=None, checked_at="2026-08-25T00:00:00")
+    aware = _raw_row(health_check_id="hc-aware", checked_at="2026-08-25T01:00:00+02:00")
+    rows = await fetch_latest_health_checks(_prisma([naive, aware]))
+    assert {row.health_check_id: (row.model_id, row.checked_at) for row in rows} == {
+        "hc-naive": (None, datetime(2026, 8, 25, 0, 0, tzinfo=timezone.utc)),
+        "hc-aware": ("deployment-abc", datetime(2026, 8, 24, 23, 0, tzinfo=timezone.utc)),
+    }
+
+
+@pytest.mark.asyncio
+async def test_json_details_decode_from_text_and_pass_through_as_dict():
+    rows = await fetch_latest_health_checks(
+        _prisma(
+            [
+                _raw_row(health_check_id="text", details='{"region": "eu"}'),
+                _raw_row(health_check_id="dict", details={"region": "us"}),
+                _raw_row(health_check_id="none", details=None),
+            ]
+        )
+    )
+    assert {row.health_check_id: row.details for row in rows} == {
+        "text": {"region": "eu"},
+        "dict": {"region": "us"},
+        "none": None,
+    }
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_degrades_to_no_rows_when_the_query_fails():
+    prisma = _prisma([])
+    prisma.db.query_raw.side_effect = RuntimeError("db down")
+    assert await fetch_latest_health_checks(prisma) == ()
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_degrades_to_no_rows_for_a_malformed_row():
+    assert await fetch_latest_health_checks(_prisma([{"unexpected": "shape"}])) == ()
+
+
+@pytest.mark.asyncio
+async def test_fetch_for_models_binds_the_page_as_the_only_parameter():
+    prisma = _prisma([_raw_row()])
+    rows = await fetch_latest_health_checks_for_models(prisma, ("gpt-4", "claude-opus"))
+    assert prisma.db.query_raw.await_args.args == (LATEST_HEALTH_CHECKS_FOR_MODELS_SQL, ["gpt-4", "claude-opus"])
+    assert [row.model_name for row in rows] == ["gpt-4"]
+    assert 'WHERE "model_name" = ANY($1)' in LATEST_HEALTH_CHECKS_FOR_MODELS_SQL
+
+
+@pytest.mark.asyncio
+async def test_fetch_for_models_skips_the_database_for_an_empty_page():
+    prisma = _prisma([])
+    assert await fetch_latest_health_checks_for_models(prisma, ()) == ()
+    prisma.db.query_raw.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fetch_for_models_degrades_to_no_rows_when_the_query_fails():
+    prisma = _prisma([])
+    prisma.db.query_raw.side_effect = RuntimeError("db down")
+    assert await fetch_latest_health_checks_for_models(prisma, ("gpt-4",)) == ()

tests/test_litellm/proxy/proxy_server/test_background_health.py

@@ -20,6 +20,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 import litellm.proxy.proxy_server as proxy_server
+from litellm.constants import BACKGROUND_HEALTH_CHECK_DB_SAVE_JOB_NAME
 from litellm.proxy.proxy_server import (
     _adaptive_router_flusher_loop,
     _get_endpoint_exception_status,
@@ -245,6 +246,86 @@ async def test_schedule_background_health_check_db_save_invalid_no_event_loop_ra
         )
 
 
+def _lock_manager(redis_cache, acquired):
+    manager = MagicMock()
+    manager.redis_cache = redis_cache
+    manager.acquire_lock = AsyncMock(return_value=acquired)
+    manager.release_lock = AsyncMock()
+    return manager
+
+
+def _capture_saves(monkeypatch):
+    saves = []
+
+    async def _fake_save(*_args, **kwargs):
+        saves.append(kwargs)
+
+    import litellm.proxy.health_endpoints._health_endpoints as he
+
+    monkeypatch.setattr(he, "_save_background_health_checks_to_db", _fake_save)
+    return saves
+
+
+def _schedule_with(lock_manager):
+    _schedule_background_health_check_db_save(
+        prisma_client=MagicMock(),
+        shared_health_manager=None,
+        model_list=[],
+        healthy_endpoints=[],
+        unhealthy_endpoints=[],
+        pod_lock_manager=lock_manager,
+        lock_ttl=300,
+    )
+
+
+@pytest.mark.asyncio
+async def test_schedule_background_health_check_db_save_skips_a_window_another_pod_persisted(monkeypatch):
+    saves = _capture_saves(monkeypatch)
+    lock_manager = _lock_manager(redis_cache=MagicMock(), acquired=False)
+
+    _schedule_with(lock_manager)
+    await asyncio.sleep(0)
+
+    assert saves == []
+
+
+@pytest.mark.asyncio
+async def test_schedule_background_health_check_db_save_holds_the_window_lock_for_the_whole_interval(monkeypatch):
+    """The lock is the "saved this window" marker: never reentrant, TTL = interval, and never released."""
+    saves = _capture_saves(monkeypatch)
+    lock_manager = _lock_manager(redis_cache=MagicMock(), acquired=True)
+
+    _schedule_with(lock_manager)
+    await asyncio.sleep(0)
+
+    assert normalize(
+        {
+            "saves": len(saves),
+            "lock_request": lock_manager.acquire_lock.await_args.kwargs,
+            "released": lock_manager.release_lock.await_count,
+        }
+    ) == {
+        "saves": 1,
+        "lock_request": {
+            "cronjob_id": BACKGROUND_HEALTH_CHECK_DB_SAVE_JOB_NAME,
+            "ttl": 300,
+            "allow_reentrant": False,
+        },
+        "released": 0,
+    }
+
+
+@pytest.mark.asyncio
+async def test_schedule_background_health_check_db_save_runs_ungated_without_redis(monkeypatch):
+    saves = _capture_saves(monkeypatch)
+    lock_manager = _lock_manager(redis_cache=None, acquired=True)
+
+    _schedule_with(lock_manager)
+    await asyncio.sleep(0)
+
+    assert (len(saves), lock_manager.acquire_lock.await_count) == (1, 0)
+
+
 # ---------------------------------------------------------------------------
 # _get_endpoint_exception_status
 # ---------------------------------------------------------------------------

tests/test_litellm/proxy/test_health_check_functions.py

@@ -6,6 +6,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.db.health_check_latest import LatestHealthCheckRow
 from litellm.proxy.health_endpoints._health_endpoints import (
     _aggregate_health_check_results,
     _build_model_param_to_info_mapping,
@@ -13,6 +15,7 @@ from litellm.proxy.health_endpoints._health_endpoints import (
     _save_background_health_checks_to_db,
     _save_health_check_results_if_changed,
     _save_health_check_to_db,
+    latest_health_checks_endpoint,
 )
 from litellm.proxy.utils import PrismaClient
 
@@ -451,96 +454,125 @@ async def test_save_background_health_checks_to_db_exception_handling():
     # Function should complete without raising
 
 
-@pytest.mark.asyncio
-async def test_get_all_latest_health_checks_with_model_id(mock_prisma):
-    """Test get_all_latest_health_checks properly groups by model_id"""
-    mock_check2 = MagicMock()
-    mock_check2.model_id = "model-456"
-    mock_check2.model_name = "gpt-3.5-turbo"
-    mock_check2.checked_at = datetime.now(timezone.utc) - timedelta(minutes=5)
-
-    mock_check3 = MagicMock()
-    mock_check3.model_id = "model-123"
-    mock_check3.model_name = "gpt-3.5-turbo"
-    mock_check3.checked_at = datetime.now(timezone.utc) - timedelta(
-        minutes=1
-    )  # Latest for model-123
-
-    # Order by checked_at desc
-    mock_prisma.db.litellm_healthchecktable.find_many = AsyncMock(
-        return_value=[mock_check3, mock_check2]
-    )
-
-    result = await mock_prisma.get_all_latest_health_checks()
-
-    # Should return 2 unique models (by model_id)
-    assert len(result) == 2
-
-    # Should have latest check for each model_id
-    model_ids = {check.model_id for check in result}
-    assert "model-123" in model_ids
-    assert "model-456" in model_ids
-
-    # model-123 should have the latest check (1 minute ago)
-    model123_check = next(c for c in result if c.model_id == "model-123")
-    assert model123_check.checked_at == mock_check3.checked_at
+def _raw_latest_row(model_name: str, model_id, checked_at: datetime) -> dict:
+    return {
+        "health_check_id": f"hc-{model_id or 'no-id'}-{model_name}",
+        "model_name": model_name,
+        "model_id": model_id,
+        "status": "healthy",
+        "healthy_count": 1,
+        "unhealthy_count": 0,
+        "error_message": None,
+        "response_time_ms": 10.0,
+        "details": None,
+        "checked_by": "pod-1",
+        "checked_at": checked_at.isoformat(),
+        "created_at": checked_at.isoformat(),
+        "updated_at": checked_at.isoformat(),
+    }
 
 
 @pytest.mark.asyncio
-async def test_get_all_latest_health_checks_without_model_id(mock_prisma):
-    """Test get_all_latest_health_checks groups by model_name when model_id is None"""
-    mock_check2 = MagicMock()
-    mock_check2.model_id = None
-    mock_check2.model_name = "gpt-3.5-turbo"
-    mock_check2.checked_at = datetime.now(timezone.utc) - timedelta(minutes=1)  # Latest
-
-    mock_prisma.db.litellm_healthchecktable.find_many = AsyncMock(
-        return_value=[mock_check2]
-    )
-
-    result = await mock_prisma.get_all_latest_health_checks()
-
-    # Should return 1 unique model (by model_name)
-    assert len(result) == 1
-    assert result[0].model_name == "gpt-3.5-turbo"
-    assert result[0].checked_at == mock_check2.checked_at  # Latest
-
-
-@pytest.mark.asyncio
-async def test_get_all_latest_health_checks_same_name_with_and_without_model_id(
-    mock_prisma,
-):
+async def test_get_all_latest_health_checks_keeps_every_distinct_group_with_its_own_checked_at(mock_prisma):
     """
-    Same model_name can appear twice after DISTINCT ON: once keyed by (model_id, name)
-    and once by (NULL, name) — different Postgres groups than a single row with id.
+    Postgres owns the dedup. (id, name), (other id, name) and (NULL, name) are distinct groups and each row
+    must arrive typed, with its own checked_at, for the 1h re-save compare and the id-or-name lookup key.
     """
     now = datetime.now(timezone.utc)
-    with_id = MagicMock()
-    with_id.model_id = "deployment-abc"
-    with_id.model_name = "gpt-4"
-    with_id.checked_at = now - timedelta(minutes=2)
-
-    without_id = MagicMock()
-    without_id.model_id = None
-    without_id.model_name = "gpt-4"
-    without_id.checked_at = now - timedelta(minutes=1)
-
-    mock_prisma.db.litellm_healthchecktable.find_many = AsyncMock(
-        return_value=[without_id, with_id]
+    mock_prisma.db.query_raw = AsyncMock(
+        return_value=[
+            _raw_latest_row("gpt-3.5-turbo", "model-123", now - timedelta(minutes=1)),
+            _raw_latest_row("gpt-3.5-turbo", "model-456", now - timedelta(minutes=5)),
+            _raw_latest_row("gpt-4", "deployment-abc", now - timedelta(minutes=2)),
+            _raw_latest_row("gpt-4", None, now - timedelta(minutes=3)),
+        ]
     )
 
     result = await mock_prisma.get_all_latest_health_checks()
 
-    assert len(result) == 2
-    names = {r.model_name for r in result}
-    assert names == {"gpt-4"}
-    ids = {r.model_id for r in result}
-    assert "deployment-abc" in ids
-    assert None in ids
+    assert {(check.model_id, check.model_name): check.checked_at for check in result} == {
+        ("model-123", "gpt-3.5-turbo"): now - timedelta(minutes=1),
+        ("model-456", "gpt-3.5-turbo"): now - timedelta(minutes=5),
+        ("deployment-abc", "gpt-4"): now - timedelta(minutes=2),
+        (None, "gpt-4"): now - timedelta(minutes=3),
+    }
 
-    by_key = {(r.model_id, r.model_name): r for r in result}
-    assert by_key[("deployment-abc", "gpt-4")].checked_at == with_id.checked_at
-    assert by_key[(None, "gpt-4")].checked_at == without_id.checked_at
+
+@pytest.mark.asyncio
+async def test_save_background_health_checks_compares_raw_checked_at_against_utc_now(mock_prisma):
+    """
+    Raw rows carry ISO strings and the engine may omit the offset. A naive checked_at would TypeError
+    inside the 1h compare, be swallowed, and silently stop every save; a stale row must still re-save.
+    """
+    stale = (datetime.now(timezone.utc) - timedelta(hours=2)).replace(tzinfo=None)
+    fresh = datetime.now(timezone.utc) - timedelta(minutes=5)
+    mock_prisma.db.query_raw = AsyncMock(
+        return_value=[
+            _raw_latest_row("stale-model", "stale-id", stale),
+            _raw_latest_row("fresh-model", "fresh-id", fresh),
+        ]
+    )
+    mock_prisma.save_health_check_result = AsyncMock()
+    model_list = [
+        {"model_name": "stale-model", "model_info": {"id": "stale-id"}, "litellm_params": {"model": "openai/stale"}},
+        {"model_name": "fresh-model", "model_info": {"id": "fresh-id"}, "litellm_params": {"model": "openai/fresh"}},
+    ]
+
+    await _save_background_health_checks_to_db(
+        mock_prisma,
+        model_list,
+        [{"model": "openai/stale"}, {"model": "openai/fresh"}],
+        [],
+        time.time(),
+        "pod-1",
+    )
+    await asyncio.sleep(0)
+
+    assert [call.kwargs["model_id"] for call in mock_prisma.save_health_check_result.await_args_list] == ["stale-id"]
+
+
+@pytest.mark.asyncio
+async def test_latest_health_checks_endpoint_serialises_raw_rows(monkeypatch):
+    row = LatestHealthCheckRow(
+        health_check_id="hc-1",
+        model_name="gpt-4",
+        model_id="deployment-abc",
+        status="healthy",
+        healthy_count=1,
+        unhealthy_count=0,
+        error_message=None,
+        response_time_ms=12.5,
+        details='{"region": "eu"}',
+        checked_by="pod-1",
+        checked_at=datetime(2026, 8, 25),
+        created_at=datetime(2026, 8, 25, tzinfo=timezone.utc),
+        updated_at=datetime(2026, 8, 25, tzinfo=timezone.utc),
+    )
+    prisma = MagicMock()
+    prisma.get_all_latest_health_checks = AsyncMock(return_value=(row,))
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma)
+
+    response = await latest_health_checks_endpoint(user_api_key_dict=UserAPIKeyAuth())
+
+    assert response == {
+        "latest_health_checks": {
+            "deployment-abc": {
+                "health_check_id": "hc-1",
+                "model_name": "gpt-4",
+                "model_id": "deployment-abc",
+                "status": "healthy",
+                "healthy_count": 1,
+                "unhealthy_count": 0,
+                "error_message": None,
+                "response_time_ms": 12.5,
+                "details": {"region": "eu"},
+                "checked_by": "pod-1",
+                "checked_at": "2026-08-25T00:00:00+00:00",
+                "created_at": "2026-08-25T00:00:00+00:00",
+            }
+        },
+        "total_models": 1,
+    }
 
 
 @pytest.mark.asyncio

tests/test_litellm/proxy/utils/prisma_and_spend/test_prisma_client_health.py

@@ -22,6 +22,10 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from litellm.proxy.db.health_check_latest import (
+    LATEST_HEALTH_CHECKS_FOR_MODELS_SQL,
+    LATEST_HEALTH_CHECKS_SQL,
+)
 from litellm.proxy.utils import PrismaClient
 
 
@@ -261,36 +265,49 @@ async def test_get_health_check_history_db_error_returns_empty_list(
     assert await prisma_client.get_health_check_history() == []
 
 
+def _raw_health_check_row(model_name: str = "gpt-4", model_id: str | None = "deployment-abc") -> dict[str, Any]:
+    return {
+        "health_check_id": f"hc-{model_name}",
+        "model_name": model_name,
+        "model_id": model_id,
+        "status": "healthy",
+        "healthy_count": 1,
+        "unhealthy_count": 0,
+        "error_message": None,
+        "response_time_ms": 12.5,
+        "details": None,
+        "checked_by": "pod-1",
+        "checked_at": "2026-08-25T00:00:00+00:00",
+        "created_at": "2026-08-25T00:00:00+00:00",
+        "updated_at": "2026-08-25T00:00:00+00:00",
+    }
+
+
 @pytest.mark.asyncio
-async def test_get_all_latest_health_checks_uses_distinct(
+async def test_get_all_latest_health_checks_dedups_in_postgres(
     prisma_client: PrismaClient,
 ) -> None:
-    rows = [MagicMock(name=f"row-{i}") for i in range(3)]
-    prisma_client.db.litellm_healthchecktable.find_many = AsyncMock(return_value=rows)
+    """A revert to prisma find_many(distinct=...) streams the whole history table; the SQL must own the DISTINCT."""
+    prisma_client.db.query_raw = AsyncMock(return_value=[_raw_health_check_row()])
     result = await prisma_client.get_all_latest_health_checks()
-    kwargs = prisma_client.db.litellm_healthchecktable.find_many.await_args.kwargs
     actual = {
-        "len": len(result),
-        "distinct": kwargs["distinct"],
-        "order_len": len(kwargs["order"]),
-        "first_order": kwargs["order"][0],
+        "query": prisma_client.db.query_raw.await_args.args,
+        "rows": [(row.model_id, row.model_name, row.status) for row in result],
+        "row_type": type(result[0]).__name__,
     }
     assert actual == {
-        "len": 3,
-        "distinct": ["model_id", "model_name"],
-        "order_len": 3,
-        "first_order": {"model_id": "asc"},
+        "query": (LATEST_HEALTH_CHECKS_SQL,),
+        "rows": [("deployment-abc", "gpt-4", "healthy")],
+        "row_type": "LatestHealthCheckRow",
     }
 
 
 @pytest.mark.asyncio
-async def test_get_all_latest_health_checks_db_error_returns_empty_list(
+async def test_get_all_latest_health_checks_db_error_returns_no_rows(
     prisma_client: PrismaClient,
 ) -> None:
-    prisma_client.db.litellm_healthchecktable.find_many = AsyncMock(
-        side_effect=RuntimeError("oops")
-    )
-    assert await prisma_client.get_all_latest_health_checks() == []
+    prisma_client.db.query_raw = AsyncMock(side_effect=RuntimeError("oops"))
+    assert await prisma_client.get_all_latest_health_checks() == ()
 
 
 @pytest.mark.asyncio
@@ -298,18 +315,15 @@ async def test_get_latest_health_checks_for_models_bounds_the_query_to_those_mod
     prisma_client: PrismaClient,
 ) -> None:
     """A paged caller reads health for its page; an unbounded read is the bug this exists to avoid."""
-    prisma_client.db.litellm_healthchecktable.find_many = AsyncMock(return_value=[])
-    await prisma_client.get_latest_health_checks_for_models(["gpt-5", "claude-opus"])
-    kwargs = prisma_client.db.litellm_healthchecktable.find_many.await_args.kwargs
+    prisma_client.db.query_raw = AsyncMock(return_value=[_raw_health_check_row(model_name="gpt-5")])
+    result = await prisma_client.get_latest_health_checks_for_models(["gpt-5", "claude-opus"])
     actual = {
-        "where": kwargs["where"],
-        "distinct": kwargs["distinct"],
-        "order": kwargs["order"],
+        "query": prisma_client.db.query_raw.await_args.args,
+        "rows": [row.model_name for row in result],
     }
     assert actual == {
-        "where": {"model_name": {"in": ["gpt-5", "claude-opus"]}},
-        "distinct": ["model_id", "model_name"],
-        "order": [{"model_id": "asc"}, {"model_name": "asc"}, {"checked_at": "desc"}],
+        "query": (LATEST_HEALTH_CHECKS_FOR_MODELS_SQL, ["gpt-5", "claude-opus"]),
+        "rows": ["gpt-5"],
     }
 
 
@@ -317,14 +331,14 @@ async def test_get_latest_health_checks_for_models_bounds_the_query_to_those_mod
 async def test_get_latest_health_checks_for_models_does_not_query_for_an_empty_page(
     prisma_client: PrismaClient,
 ) -> None:
-    prisma_client.db.litellm_healthchecktable.find_many = AsyncMock(return_value=[])
+    prisma_client.db.query_raw = AsyncMock(return_value=[])
     assert await prisma_client.get_latest_health_checks_for_models([]) == ()
-    assert prisma_client.db.litellm_healthchecktable.find_many.await_count == 0
+    assert prisma_client.db.query_raw.await_count == 0
 
 
 @pytest.mark.asyncio
-async def test_get_latest_health_checks_for_models_db_error_returns_empty_list(
+async def test_get_latest_health_checks_for_models_db_error_returns_no_rows(
     prisma_client: PrismaClient,
 ) -> None:
-    prisma_client.db.litellm_healthchecktable.find_many = AsyncMock(side_effect=RuntimeError("oops"))
+    prisma_client.db.query_raw = AsyncMock(side_effect=RuntimeError("oops"))
     assert await prisma_client.get_latest_health_checks_for_models(["gpt-5"]) == ()