diff --git a/litellm/__init__.py b/litellm/__init__.py
index f020ed9293..3365abe325 100644
--- a/litellm/__init__.py
+++ b/litellm/__init__.py
@@ -413,6 +413,12 @@ internal_user_budget_duration: Optional[str] = None
 tag_budget_config: Optional[Dict[str, "BudgetConfig"]] = None
 max_end_user_budget: Optional[float] = None
 max_end_user_budget_id: Optional[str] = None
+# When True, end-user IDs extracted from requests are validated against
+# LiteLLM_EndUserTable / LiteLLM_UserTable. Values that do not resolve to a
+# known row are dropped before reaching spend logs. Defaults to False for
+# backwards compatibility — arbitrary client-supplied identifiers still
+# pass through unchanged.
+validate_end_user_id_in_db: bool = False
 disable_end_user_cost_tracking: Optional[bool] = None
 disable_end_user_cost_tracking_prometheus_only: Optional[bool] = None
 enable_end_user_cost_tracking_prometheus_only: Optional[bool] = None
diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py
index 13381c7a6c..09bb805720 100644
--- a/litellm/proxy/auth/auth_checks.py
+++ b/litellm/proxy/auth/auth_checks.py
@@ -1187,6 +1187,127 @@ async def get_end_user_object(
         return None
 
 
+_END_USER_VALIDATION_NEGATIVE_TTL = 60
+_END_USER_VALIDATION_POSITIVE_TTL = 300
+
+
+async def resolve_and_validate_end_user_id(
+    raw_end_user_id: Optional[str],
+    prisma_client: Optional[PrismaClient],
+    user_api_key_cache: UserApiKeyCache,
+    parent_otel_span: Optional[Span] = None,
+    proxy_logging_obj: Optional[ProxyLogging] = None,
+    route: str = "",
+) -> Optional[str]:
+    """Optionally drop end-user ids that don't resolve to a known DB row.
+
+    Default: pass-through. LiteLLM's documented pattern is that the `user`
+    field is an arbitrary caller-supplied identifier, so validation is
+    opt-in behind ``litellm.validate_end_user_id_in_db`` to preserve
+    backwards compatibility.
+
+    When the flag is set: accept the id when it matches any of
+      - LiteLLM_EndUserTable.user_id
+      - LiteLLM_UserTable.user_id
+      - LiteLLM_UserTable.user_email (case-insensitive)
+
+    If the id doesn't match but ``litellm.max_end_user_budget_id`` is set,
+    we still preserve the id so the default end-user budget is applied
+    downstream; otherwise we return None.
+
+    DB lookups reuse ``get_end_user_object`` / ``get_user_object`` so they
+    share the same cache as the rest of the auth path instead of adding new
+    raw Prisma queries.
+    """
+    if raw_end_user_id is None:
+        return None
+    if not litellm.validate_end_user_id_in_db:
+        return raw_end_user_id
+    if prisma_client is None:
+        return raw_end_user_id
+
+    cache_key = f"end_user_validation:{raw_end_user_id}"
+    cached = await user_api_key_cache.async_get_cache(key=cache_key)
+    if cached == "valid":
+        return raw_end_user_id
+    if cached == "invalid":
+        return raw_end_user_id if litellm.max_end_user_budget_id else None
+
+    is_valid = await _end_user_id_exists_in_db(
+        end_user_id=raw_end_user_id,
+        prisma_client=prisma_client,
+        user_api_key_cache=user_api_key_cache,
+        parent_otel_span=parent_otel_span,
+        proxy_logging_obj=proxy_logging_obj,
+        route=route,
+    )
+
+    await user_api_key_cache.async_set_cache(
+        key=cache_key,
+        value="valid" if is_valid else "invalid",
+        ttl=(
+            _END_USER_VALIDATION_POSITIVE_TTL
+            if is_valid
+            else _END_USER_VALIDATION_NEGATIVE_TTL
+        ),
+    )
+
+    if is_valid:
+        return raw_end_user_id
+    # Preserve id so the caller can still apply litellm.max_end_user_budget_id.
+    if litellm.max_end_user_budget_id:
+        return raw_end_user_id
+    return None
+
+
+async def _end_user_id_exists_in_db(
+    end_user_id: str,
+    prisma_client: PrismaClient,
+    user_api_key_cache: UserApiKeyCache,
+    parent_otel_span: Optional[Span] = None,
+    proxy_logging_obj: Optional[ProxyLogging] = None,
+    route: str = "",
+) -> bool:
+    """True when the id matches an EndUser, User, or user_email row."""
+    try:
+        end_user_obj = await get_end_user_object(
+            end_user_id=end_user_id,
+            prisma_client=prisma_client,
+            user_api_key_cache=user_api_key_cache,
+            parent_otel_span=parent_otel_span,
+            proxy_logging_obj=proxy_logging_obj,
+            route=route,
+        )
+        if end_user_obj is not None:
+            return True
+    except litellm.BudgetExceededError:
+        raise
+    except Exception as e:
+        verbose_proxy_logger.debug(
+            f"end_user validation: get_end_user_object lookup failed: {e}"
+        )
+
+    try:
+        user_obj = await get_user_object(
+            user_id=end_user_id,
+            prisma_client=prisma_client,
+            user_api_key_cache=user_api_key_cache,
+            user_id_upsert=False,
+            parent_otel_span=parent_otel_span,
+            proxy_logging_obj=proxy_logging_obj,
+            check_db_only=False,
+            user_email=end_user_id if "@" in end_user_id else None,
+        )
+        if user_obj is not None:
+            return True
+    except Exception as e:
+        verbose_proxy_logger.debug(
+            f"end_user validation: get_user_object lookup failed: {e}"
+        )
+
+    return False
+
+
 @log_db_metrics
 async def get_tag_objects_batch(
     tag_names: List[str],
diff --git a/litellm/proxy/auth/auth_utils.py b/litellm/proxy/auth/auth_utils.py
index 637a4a070c..c4dcca764b 100644
--- a/litellm/proxy/auth/auth_utils.py
+++ b/litellm/proxy/auth/auth_utils.py
@@ -10,6 +10,7 @@ import litellm
 from litellm import Router, provider_list
 from litellm._logging import verbose_proxy_logger
 from litellm.constants import STANDARD_CUSTOMER_ID_HEADERS
+from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
 from litellm.litellm_core_utils.url_utils import SSRFError, validate_url
 from litellm.proxy._types import *
 from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS
@@ -1008,12 +1009,47 @@ def _get_customer_id_from_standard_headers(
     for standard_header in STANDARD_CUSTOMER_ID_HEADERS:
         for header_name, header_value in request_headers.items():
             if header_name.lower() == standard_header.lower():
-                user_id_str = str(header_value) if header_value is not None else ""
-                if user_id_str.strip():
+                user_id_str = _coerce_user_id_to_str(header_value)
+                if user_id_str:
                     return user_id_str
     return None
 
 
+def _coerce_user_id_to_str(value: Any) -> Optional[str]:
+    """Return a usable end-user identifier string, or None if the value isn't one.
+
+    Always drops non-string structured values (dict/list/tuple/set) because
+    stringifying them produces garbage spend-log rows like
+    ``"{'device_id': ...}"``. Strings that *decode* to a structured payload
+    are only rejected when ``litellm.validate_end_user_id_in_db`` is enabled
+    — operators who currently pass JSON-encoded identifiers keep their
+    existing behavior until they opt in. See
+    auth_utils.py:get_end_user_id_from_request_body for the extraction chain.
+    """
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        # bool is an int subclass; handle explicitly to avoid "True"/"False".
+        return None
+    if isinstance(value, (int, float)):
+        return str(value)
+    if isinstance(value, str):
+        stripped = value.strip()
+        if not stripped:
+            return None
+        # Reject strings that decode to a structured payload (JSON object/array)
+        # only when the operator has opted into end-user validation. Gating
+        # behind the flag preserves backwards compatibility for deployments
+        # that intentionally pass JSON-encoded user identifiers.
+        if litellm.validate_end_user_id_in_db and stripped[:1] in ("{", "["):
+            parsed = safe_json_loads(stripped)
+            if isinstance(parsed, (dict, list)):
+                return None
+        return stripped
+    # dict, list, tuple, set, arbitrary objects -> drop.
+    return None
+
+
 def get_end_user_id_from_request_body(
     request_body: dict, request_headers: Optional[dict] = None
 ) -> Optional[str]:
@@ -1052,23 +1088,22 @@ def get_end_user_id_from_request_body(
         if isinstance(custom_header_name_to_check, list):
             headers_lower = {k.lower(): v for k, v in request_headers.items()}
             for expected_header in custom_header_name_to_check:
-                header_value = headers_lower.get(expected_header)
-                if header_value is not None:
-                    user_id_str = str(header_value)
-                    if user_id_str.strip():
-                        return user_id_str
+                user_id_str = _coerce_user_id_to_str(headers_lower.get(expected_header))
+                if user_id_str:
+                    return user_id_str
 
         elif isinstance(custom_header_name_to_check, str):
             for header_name, header_value in request_headers.items():
                 if header_name.lower() == custom_header_name_to_check.lower():
-                    user_id_str = str(header_value) if header_value is not None else ""
-                    if user_id_str.strip():
+                    user_id_str = _coerce_user_id_to_str(header_value)
+                    if user_id_str:
                         return user_id_str
 
     # Check 3: 'user' field in request_body (commonly OpenAI)
-    if "user" in request_body and request_body["user"] is not None:
-        user_from_body_user_field = request_body["user"]
-        return str(user_from_body_user_field)
+    if "user" in request_body:
+        user_id_str = _coerce_user_id_to_str(request_body["user"])
+        if user_id_str:
+            return user_id_str
 
     def _as_dict(value: Any) -> dict:
         # metadata / litellm_metadata can arrive as JSON strings from
@@ -1077,32 +1112,30 @@ def get_end_user_id_from_request_body(
         if isinstance(value, dict):
             return value
         if isinstance(value, str):
-            from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
-
             parsed = safe_json_loads(value)
             return parsed if isinstance(parsed, dict) else {}
         return {}
 
     # Check 4: 'litellm_metadata.user' in request_body (commonly Anthropic)
     litellm_metadata = _as_dict(request_body.get("litellm_metadata"))
-    user_from_litellm_metadata = litellm_metadata.get("user")
-    if user_from_litellm_metadata is not None:
-        return str(user_from_litellm_metadata)
+    user_id_str = _coerce_user_id_to_str(litellm_metadata.get("user"))
+    if user_id_str:
+        return user_id_str
 
     # Check 5: 'metadata.user_id' in request_body (another common pattern)
     metadata_dict = _as_dict(request_body.get("metadata"))
-    user_id_from_metadata_field = metadata_dict.get("user_id")
-    if user_id_from_metadata_field is not None:
-        return str(user_id_from_metadata_field)
+    user_id_str = _coerce_user_id_to_str(metadata_dict.get("user_id"))
+    if user_id_str:
+        return user_id_str
 
     # Check 6: 'safety_identifier' in request body (OpenAI Responses API parameter)
     # SECURITY NOTE: safety_identifier can be set by any caller in the request body.
     # Only use this for end-user identification in trusted environments where you control
     # the calling application. For untrusted callers, prefer using headers or server-side
     # middleware to set the end_user_id to prevent impersonation.
-    if request_body.get("safety_identifier") is not None:
-        user_from_body_user_field = request_body["safety_identifier"]
-        return str(user_from_body_user_field)
+    user_id_str = _coerce_user_id_to_str(request_body.get("safety_identifier"))
+    if user_id_str:
+        return user_id_str
 
     return None
 
diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py
index 0cca9414b2..6974860a22 100644
--- a/litellm/proxy/auth/user_api_key_auth.py
+++ b/litellm/proxy/auth/user_api_key_auth.py
@@ -44,6 +44,7 @@ from litellm.proxy.auth.auth_checks import (
     get_team_object,
     get_user_object,
     is_valid_fallback_model,
+    resolve_and_validate_end_user_id,
 )
 from litellm.proxy.auth.auth_exception_handler import UserAPIKeyAuthExceptionHandler
 from litellm.proxy.auth.auth_utils import (
@@ -1071,9 +1072,17 @@ async def _user_api_key_auth_builder(  # noqa: PLR0915
         _end_user_object = None
         end_user_params = {}
 
-        end_user_id = get_end_user_id_from_request_body(
+        raw_end_user_id = get_end_user_id_from_request_body(
             request_data, _safe_get_request_headers(request)
         )
+        end_user_id = await resolve_and_validate_end_user_id(
+            raw_end_user_id=raw_end_user_id,
+            prisma_client=prisma_client,
+            user_api_key_cache=user_api_key_cache,
+            parent_otel_span=parent_otel_span,
+            proxy_logging_obj=proxy_logging_obj,
+            route=route,
+        )
         if end_user_id:
             try:
                 end_user_params["end_user_id"] = end_user_id
@@ -1759,7 +1768,7 @@ def _team_obj_from_token(valid_token: UserAPIKeyAuth) -> LiteLLM_TeamTableCached
 
 
 @tracer.wrap()
-async def _run_centralized_common_checks(
+async def _run_centralized_common_checks(  # noqa: PLR0915
     user_api_key_auth_obj: UserAPIKeyAuth,
     request: Request,
     request_data: dict,
@@ -1837,9 +1846,23 @@ async def _run_centralized_common_checks(
         return
 
     parent_otel_span = user_api_key_auth_obj.parent_otel_span
-    end_user_id = get_end_user_id_from_request_body(
-        request_data, _safe_get_request_headers(request)
-    )
+    # In the integrated auth flow ``_user_api_key_auth_builder`` has already
+    # resolved the end-user id and attached it here. Reuse that to avoid a
+    # second extraction pass; fall back to extracting locally when the
+    # function is invoked in isolation (e.g. in direct unit tests).
+    end_user_id = user_api_key_auth_obj.end_user_id
+    if end_user_id is None:
+        raw_end_user_id = get_end_user_id_from_request_body(
+            request_data, _safe_get_request_headers(request)
+        )
+        end_user_id = await resolve_and_validate_end_user_id(
+            raw_end_user_id=raw_end_user_id,
+            prisma_client=prisma_client,
+            user_api_key_cache=user_api_key_cache,
+            parent_otel_span=parent_otel_span,
+            proxy_logging_obj=proxy_logging_obj,
+            route=route,
+        )
 
     fetch_coros = []
     if user_api_key_auth_obj.team_id is not None:
@@ -2170,11 +2193,33 @@ async def user_api_key_auth(
             api_key=api_key,
         )
 
-    end_user_id = get_end_user_id_from_request_body(
-        request_data, _safe_get_request_headers(request)
-    )
-    if end_user_id is not None:
-        user_api_key_auth_obj.end_user_id = end_user_id
+    # Defense-in-depth: ``_user_api_key_auth_builder`` has multiple early-return
+    # paths (no master key, /user/auth route, JWT short-circuits) that bypass
+    # the end-user resolution block. If those paths produced an auth obj
+    # without an ``end_user_id`` set, fall back to extracting from the request
+    # body so spend logs are still attributed correctly. Validation honours
+    # ``litellm.validate_end_user_id_in_db``.
+    if user_api_key_auth_obj.end_user_id is None:
+        from litellm.proxy.proxy_server import (
+            prisma_client,
+            proxy_logging_obj,
+            user_api_key_cache,
+        )
+
+        raw_end_user_id = get_end_user_id_from_request_body(
+            request_data, _safe_get_request_headers(request)
+        )
+        if raw_end_user_id is not None:
+            resolved_end_user_id = await resolve_and_validate_end_user_id(
+                raw_end_user_id=raw_end_user_id,
+                prisma_client=prisma_client,
+                user_api_key_cache=user_api_key_cache,
+                parent_otel_span=user_api_key_auth_obj.parent_otel_span,
+                proxy_logging_obj=proxy_logging_obj,
+                route=route,
+            )
+            if resolved_end_user_id is not None:
+                user_api_key_auth_obj.end_user_id = resolved_end_user_id
 
     user_api_key_auth_obj.request_route = normalize_request_route(route)
     return user_api_key_auth_obj
diff --git a/tests/test_litellm/proxy/auth/test_auth_checks.py b/tests/test_litellm/proxy/auth/test_auth_checks.py
index 26f04a4abc..35a3bd7f65 100644
--- a/tests/test_litellm/proxy/auth/test_auth_checks.py
+++ b/tests/test_litellm/proxy/auth/test_auth_checks.py
@@ -3016,3 +3016,340 @@ async def test_team_member_budget_check_zero_per_member_row_still_blocks():
                 proxy_logging_obj=proxy_logging_obj,
             )
     assert exc_info.value.max_budget == 0.0
+
+
+# --- resolve_and_validate_end_user_id ---------------------------------------
+
+
+@pytest.fixture
+def _validate_flag_on(monkeypatch):
+    """Enable opt-in DB validation for the duration of a test."""
+    import litellm
+
+    monkeypatch.setattr(litellm, "validate_end_user_id_in_db", True)
+    monkeypatch.setattr(litellm, "max_end_user_budget_id", None)
+
+
+def _validation_cache():
+    cache = MagicMock()
+    cache.async_get_cache = AsyncMock(return_value=None)
+    cache.async_set_cache = AsyncMock()
+    return cache
+
+
+def _patch_validation_helpers(monkeypatch, *, end_user=None, user=None, fuzzy=None):
+    """Stub out the DB helpers resolve_and_validate_end_user_id delegates to."""
+    from litellm.proxy.auth import auth_checks
+
+    monkeypatch.setattr(
+        auth_checks, "get_end_user_object", AsyncMock(return_value=end_user)
+    )
+    monkeypatch.setattr(auth_checks, "get_user_object", AsyncMock(return_value=user))
+    monkeypatch.setattr(
+        auth_checks, "_get_fuzzy_user_object", AsyncMock(return_value=fuzzy)
+    )
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_returns_none_for_none_input(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch)
+    cache = _validation_cache()
+    assert (
+        await resolve_and_validate_end_user_id(
+            raw_end_user_id=None,
+            prisma_client=MagicMock(),
+            user_api_key_cache=cache,
+        )
+        is None
+    )
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_passes_through_when_flag_disabled(monkeypatch):
+    """Default behaviour: flag is off, arbitrary ids pass through untouched."""
+    import litellm
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    monkeypatch.setattr(litellm, "validate_end_user_id_in_db", False)
+    _patch_validation_helpers(monkeypatch)
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="codex-session-abc",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result == "codex-session-abc"
+    cache.async_set_cache.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_passes_through_when_no_prisma_client(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch)
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="alice@example.com",
+        prisma_client=None,
+        user_api_key_cache=cache,
+    )
+    assert result == "alice@example.com"
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_matches_end_user_table(_validate_flag_on, monkeypatch):
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch, end_user=MagicMock())
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="customer-123",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result == "customer-123"
+    cache.async_set_cache.assert_awaited_once()
+    kwargs = cache.async_set_cache.await_args.kwargs
+    assert kwargs["key"] == "end_user_validation:customer-123"
+    assert kwargs["value"] == "valid"
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_matches_user_table_by_user_id(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch, user=MagicMock())
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="user-xyz",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result == "user-xyz"
+    # email fallback should not run for a non-email input
+    auth_checks._get_fuzzy_user_object.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_matches_user_table_by_email(
+    _validate_flag_on, monkeypatch
+):
+    """Email-shaped ids route through get_user_object with user_email set.
+
+    The fuzzy lookup must happen inside get_user_object so it shares the
+    _should_check_db throttle and user_api_key_cache — no direct raw
+    Prisma calls on the auth path.
+    """
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch, user=MagicMock())
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="Alice@Example.com",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result == "Alice@Example.com"
+    auth_checks.get_user_object.assert_awaited_once()
+    user_kwargs = auth_checks.get_user_object.await_args.kwargs
+    assert user_kwargs["user_id"] == "Alice@Example.com"
+    assert user_kwargs["user_email"] == "Alice@Example.com"
+    # email branch must not bypass the cached helper with a raw fuzzy call
+    auth_checks._get_fuzzy_user_object.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_non_email_id_does_not_pass_user_email(
+    _validate_flag_on, monkeypatch
+):
+    """Non-email ids skip the email fuzzy path to avoid a pointless DB hit."""
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch, user=MagicMock())
+    cache = _validation_cache()
+
+    await resolve_and_validate_end_user_id(
+        raw_end_user_id="user-xyz",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    auth_checks.get_user_object.assert_awaited_once()
+    user_kwargs = auth_checks.get_user_object.await_args.kwargs
+    assert user_kwargs["user_email"] is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_drops_codex_opaque_identifier(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch)  # all helpers return None
+    cache = _validation_cache()
+
+    codex_id = (
+        "user_8a4a360c36621665b341e06fb76041d9b6def732bb183eea148d4abc9d97c1de"
+        "_account__session_a2bce4a5-8887-44ef-b491-fbf0a55c6569"
+    )
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id=codex_id,
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result is None
+    cache.async_set_cache.assert_awaited_once()
+    kwargs = cache.async_set_cache.await_args.kwargs
+    assert kwargs["value"] == "invalid"
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_preserves_id_when_default_budget_configured(
+    _validate_flag_on, monkeypatch
+):
+    """Don't drop unregistered ids when litellm.max_end_user_budget_id is set.
+
+    The default end-user budget is applied downstream when the id is present
+    but not found in the db — dropping the id here would bypass those limits.
+    """
+    import litellm
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    monkeypatch.setattr(litellm, "max_end_user_budget_id", "default-budget")
+    _patch_validation_helpers(monkeypatch)
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="new-customer",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result == "new-customer"
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_drops_unknown_email(_validate_flag_on, monkeypatch):
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch)
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="stranger@example.com",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_uses_cached_valid_result(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch)
+    cache = _validation_cache()
+    cache.async_get_cache = AsyncMock(return_value="valid")
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="alice@example.com",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result == "alice@example.com"
+    auth_checks.get_end_user_object.assert_not_awaited()
+    auth_checks.get_user_object.assert_not_awaited()
+    auth_checks._get_fuzzy_user_object.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_uses_cached_invalid_result(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    _patch_validation_helpers(monkeypatch, end_user=MagicMock())
+    cache = _validation_cache()
+    cache.async_get_cache = AsyncMock(return_value="invalid")
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="bogus",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    assert result is None
+    # Despite a matching row configured, helpers aren't called — cache wins.
+    auth_checks.get_end_user_object.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_swallows_db_errors_and_returns_none(
+    _validate_flag_on, monkeypatch
+):
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    monkeypatch.setattr(
+        auth_checks,
+        "get_end_user_object",
+        AsyncMock(side_effect=Exception("db down")),
+    )
+    monkeypatch.setattr(
+        auth_checks,
+        "get_user_object",
+        AsyncMock(side_effect=Exception("db down")),
+    )
+    cache = _validation_cache()
+
+    result = await resolve_and_validate_end_user_id(
+        raw_end_user_id="alice@example.com",
+        prisma_client=MagicMock(),
+        user_api_key_cache=cache,
+    )
+    # DB errors shouldn't raise through the auth path — treat as unknown.
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_end_user_reraises_budget_exceeded(
+    _validate_flag_on, monkeypatch
+):
+    """BudgetExceededError from get_end_user_object must bubble up so the
+    auth path enforces spend limits instead of silently dropping the id."""
+    import litellm
+    from litellm.proxy.auth import auth_checks
+    from litellm.proxy.auth.auth_checks import resolve_and_validate_end_user_id
+
+    monkeypatch.setattr(
+        auth_checks,
+        "get_end_user_object",
+        AsyncMock(
+            side_effect=litellm.BudgetExceededError(current_cost=10.0, max_budget=5.0)
+        ),
+    )
+    cache = _validation_cache()
+
+    with pytest.raises(litellm.BudgetExceededError):
+        await resolve_and_validate_end_user_id(
+            raw_end_user_id="customer-over-budget",
+            prisma_client=MagicMock(),
+            user_api_key_cache=cache,
+        )
diff --git a/tests/test_litellm/proxy/auth/test_auth_utils.py b/tests/test_litellm/proxy/auth/test_auth_utils.py
index 08035fb717..68e1636d38 100644
--- a/tests/test_litellm/proxy/auth/test_auth_utils.py
+++ b/tests/test_litellm/proxy/auth/test_auth_utils.py
@@ -597,6 +597,315 @@ def test_get_end_user_id_falls_back_to_deprecated_user_header_name():
     assert result == "user-legacy"
 
 
+class TestCoerceUserIdToStr:
+    """Unit tests for the _coerce_user_id_to_str helper."""
+
+    def test_plain_string_is_returned_verbatim(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str("alice@example.com") == "alice@example.com"
+
+    def test_string_is_stripped(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str("  bob  ") == "bob"
+
+    def test_codex_opaque_identifier_is_preserved(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        codex_id = (
+            "user_8a4a360c36621665b341e06fb76041d9b6def732bb183eea148d4abc9d97c1de"
+            "_account__session_a2bce4a5-8887-44ef-b491-fbf0a55c6569"
+        )
+        assert _coerce_user_id_to_str(codex_id) == codex_id
+
+    def test_none_returns_none(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str(None) is None
+
+    def test_empty_string_returns_none(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str("") is None
+        assert _coerce_user_id_to_str("   ") is None
+
+    def test_dict_returns_none(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        payload = {
+            "device_id": "abc",
+            "account_uuid": "",
+            "session_id": "c284b8cb",
+        }
+        assert _coerce_user_id_to_str(payload) is None
+
+    def test_list_returns_none(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str(["a", "b"]) is None
+
+    def test_json_encoded_dict_string_passes_through_by_default(self):
+        """JSON-encoded dict strings are preserved unless opt-in flag is on.
+
+        This preserves backwards compatibility: existing deployments that
+        intentionally pass JSON-encoded user identifiers keep working.
+        """
+        import litellm
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        blob = (
+            '{"device_id":"d5abe9199ee7759a0558974e9371e78c7b38d7621aae26d6609c1de61af6afb0",'
+            '"account_uuid":"","session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
+        )
+        original = litellm.validate_end_user_id_in_db
+        litellm.validate_end_user_id_in_db = False
+        try:
+            assert _coerce_user_id_to_str(blob) == blob
+        finally:
+            litellm.validate_end_user_id_in_db = original
+
+    def test_json_encoded_dict_string_returns_none_when_validation_enabled(self):
+        import litellm
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        # Same broken shape we saw in spend logs, but pre-stringified to JSON.
+        blob = (
+            '{"device_id":"d5abe9199ee7759a0558974e9371e78c7b38d7621aae26d6609c1de61af6afb0",'
+            '"account_uuid":"","session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
+        )
+        original = litellm.validate_end_user_id_in_db
+        litellm.validate_end_user_id_in_db = True
+        try:
+            assert _coerce_user_id_to_str(blob) is None
+        finally:
+            litellm.validate_end_user_id_in_db = original
+
+    def test_json_encoded_list_string_passes_through_by_default(self):
+        import litellm
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        original = litellm.validate_end_user_id_in_db
+        litellm.validate_end_user_id_in_db = False
+        try:
+            assert _coerce_user_id_to_str('["a","b"]') == '["a","b"]'
+        finally:
+            litellm.validate_end_user_id_in_db = original
+
+    def test_json_encoded_list_string_returns_none_when_validation_enabled(self):
+        import litellm
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        original = litellm.validate_end_user_id_in_db
+        litellm.validate_end_user_id_in_db = True
+        try:
+            assert _coerce_user_id_to_str('["a","b"]') is None
+        finally:
+            litellm.validate_end_user_id_in_db = original
+
+    def test_int_returns_str(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str(12345) == "12345"
+
+    def test_bool_returns_none(self):
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        # bool is an int subclass — reject explicitly, never produce "True"/"False".
+        assert _coerce_user_id_to_str(True) is None
+        assert _coerce_user_id_to_str(False) is None
+
+    def test_brace_string_that_isnt_json_is_kept(self):
+        """A string starting with `{` but failing to parse stays as-is."""
+        from litellm.proxy.auth.auth_utils import _coerce_user_id_to_str
+
+        assert _coerce_user_id_to_str("{not json") == "{not json"
+
+
+class TestGetEndUserIdDropsMalformedBodyValues:
+    """Tests that get_end_user_id_from_request_body drops dict-shaped values
+    rather than stringifying them into spend logs."""
+
+    def test_dict_user_falls_through_to_litellm_metadata(self):
+        request_body = {
+            "user": {
+                "device_id": "abc",
+                "session_id": "c284b8cb",
+            },
+            "litellm_metadata": {"user": "alice@example.com"},
+        }
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result == "alice@example.com"
+
+    def test_dict_user_with_no_other_sources_returns_none(self):
+        request_body = {
+            "user": {"device_id": "abc", "session_id": "xyz"},
+        }
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result is None
+
+    def test_json_encoded_user_string_passes_through_by_default(self):
+        """JSON-encoded user strings pass through unless validation is opted in.
+
+        Gating behind ``litellm.validate_end_user_id_in_db`` keeps existing
+        deployments that send JSON-encoded identifiers working until they
+        explicitly opt into the stricter extraction.
+        """
+        import litellm
+
+        blob = (
+            '{"device_id":"d5abe9199ee7759a","account_uuid":"",'
+            '"session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
+        )
+        request_body = {"user": blob}
+
+        original = litellm.validate_end_user_id_in_db
+        litellm.validate_end_user_id_in_db = False
+        try:
+            with patch("litellm.proxy.proxy_server.general_settings", {}):
+                result = get_end_user_id_from_request_body(
+                    request_body=request_body, request_headers={}
+                )
+        finally:
+            litellm.validate_end_user_id_in_db = original
+
+        assert result == blob
+
+    def test_json_encoded_user_string_returns_none_when_validation_enabled(self):
+        import litellm
+
+        request_body = {
+            "user": (
+                '{"device_id":"d5abe9199ee7759a","account_uuid":"",'
+                '"session_id":"c284b8cb-a050-4278-8599-cc4e016a10ab"}'
+            ),
+        }
+
+        original = litellm.validate_end_user_id_in_db
+        litellm.validate_end_user_id_in_db = True
+        try:
+            with patch("litellm.proxy.proxy_server.general_settings", {}):
+                result = get_end_user_id_from_request_body(
+                    request_body=request_body, request_headers={}
+                )
+        finally:
+            litellm.validate_end_user_id_in_db = original
+
+        assert result is None
+
+    def test_plain_string_user_is_preserved(self):
+        request_body = {"user": "alice@example.com"}
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result == "alice@example.com"
+
+    def test_codex_opaque_user_is_preserved(self):
+        codex_id = (
+            "user_8a4a360c36621665b341e06fb76041d9b6def732bb183eea148d4abc9d97c1de"
+            "_account__session_a2bce4a5-8887-44ef-b491-fbf0a55c6569"
+        )
+        request_body = {"user": codex_id}
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result == codex_id
+
+    def test_int_user_is_coerced_to_string(self):
+        request_body = {"user": 12345}
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result == "12345"
+
+    def test_list_user_falls_through(self):
+        request_body = {
+            "user": ["a", "b"],
+            "safety_identifier": "alice@example.com",
+        }
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result == "alice@example.com"
+
+    def test_dict_safety_identifier_returns_none(self):
+        request_body = {
+            "safety_identifier": {"device_id": "abc"},
+        }
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result is None
+
+    def test_dict_metadata_user_id_returns_none(self):
+        request_body = {
+            "metadata": {"user_id": {"device_id": "abc"}},
+        }
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result is None
+
+    def test_whitespace_user_falls_through(self):
+        request_body = {"user": "   ", "safety_identifier": "alice@example.com"}
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers={}
+            )
+
+        assert result == "alice@example.com"
+
+    def test_dict_user_header_falls_through_to_body(self):
+        """A dict-shaped value in a configured user-id header is dropped, not stringified."""
+        general_settings = {"user_header_name": "x-custom-user-id"}
+        # A header value will normally be a str, but be defensive: the coercion
+        # must drop anything that isn't a usable identifier.
+        headers = {"x-custom-user-id": {"device_id": "abc"}}
+        request_body = {"user": "alice@example.com"}
+
+        with (
+            patch(
+                "litellm.proxy.auth.auth_utils._get_customer_id_from_standard_headers",
+                return_value=None,
+            ),
+            patch("litellm.proxy.proxy_server.general_settings", general_settings),
+        ):
+            result = get_end_user_id_from_request_body(
+                request_body=request_body, request_headers=headers
+            )
+
+        assert result == "alice@example.com"
+
+
 def _make_deployment_dict(
     model_name: str, tpm: Optional[int] = None, rpm: Optional[int] = None
 ) -> dict:
diff --git a/tests/test_litellm/proxy/auth/test_user_api_key_auth.py b/tests/test_litellm/proxy/auth/test_user_api_key_auth.py
index 442625c75a..defd3bbcdc 100644
--- a/tests/test_litellm/proxy/auth/test_user_api_key_auth.py
+++ b/tests/test_litellm/proxy/auth/test_user_api_key_auth.py
@@ -3335,3 +3335,125 @@ async def test_master_key_auth_substitutes_alias_for_api_key():
     finally:
         for k, v in _orig.items():
             setattr(_proxy_server_mod, k, v)
+
+
+@pytest.mark.asyncio
+async def test_user_api_key_auth_sets_end_user_id_when_builder_skips_it():
+    """Defense-in-depth: ``_user_api_key_auth_builder`` has multiple
+    early-return paths (master_key=None, /user/auth route, JWT
+    short-circuits) that bypass the end-user resolution block. The wrapper
+    must still attribute spend logs to the request-supplied end-user when
+    none of those paths set it.
+
+    Krrish flagged the removal of this fallback as a regression risk; this
+    test pins the behaviour so future refactors don't silently drop it.
+    """
+    from fastapi import Request
+    from starlette.datastructures import URL
+
+    import litellm.proxy.proxy_server as _proxy_server_mod
+
+    builder_token = UserAPIKeyAuth(api_key="sk-test", user_id="u1")
+    # builder did NOT set end_user_id (e.g. master_key=None early return)
+    assert builder_token.end_user_id is None
+
+    request = Request(
+        scope={
+            "type": "http",
+            "headers": [(b"content-type", b"application/json")],
+            "method": "POST",
+        }
+    )
+    request._url = URL(url="/chat/completions")
+    request._body = json.dumps(
+        {"model": "gpt-4o", "user": "alice@example.com"}
+    ).encode()
+
+    attrs = _proxy_attrs_for_centralized_checks(user_custom_auth=None)
+    originals = {a: getattr(_proxy_server_mod, a, None) for a in attrs}
+    try:
+        for k, v in attrs.items():
+            setattr(_proxy_server_mod, k, v)
+        # Stub the builder so the test doesn't have to traverse the full
+        # auth state machine; we only care about the wrapper's safety net.
+        with (
+            patch(
+                "litellm.proxy.auth.user_api_key_auth._user_api_key_auth_builder",
+                new_callable=AsyncMock,
+                return_value=builder_token,
+            ),
+            patch(
+                "litellm.proxy.auth.user_api_key_auth._run_centralized_common_checks",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "litellm.proxy.auth.user_api_key_auth.RouteChecks.should_call_route",
+            ),
+        ):
+            result = await user_api_key_auth(request=request, api_key="Bearer sk-test")
+
+        # Validation flag is False by default → pass-through, raw value lands
+        # on the auth obj instead of being silently dropped.
+        assert result.end_user_id == "alice@example.com"
+    finally:
+        for k, v in originals.items():
+            setattr(_proxy_server_mod, k, v)
+
+
+@pytest.mark.asyncio
+async def test_user_api_key_auth_does_not_overwrite_end_user_id_set_by_builder():
+    """When the builder already resolved the end-user id (the primary
+    path), the wrapper-level safety net must not run a second resolution
+    pass — that would re-extract from the request body and could
+    overwrite a value the builder explicitly chose to set."""
+    from fastapi import Request
+    from starlette.datastructures import URL
+
+    import litellm.proxy.proxy_server as _proxy_server_mod
+
+    builder_token = UserAPIKeyAuth(
+        api_key="sk-test", user_id="u1", end_user_id="builder-resolved-id"
+    )
+
+    request = Request(
+        scope={
+            "type": "http",
+            "headers": [(b"content-type", b"application/json")],
+            "method": "POST",
+        }
+    )
+    request._url = URL(url="/chat/completions")
+    request._body = json.dumps(
+        {"model": "gpt-4o", "user": "different-id-from-body"}
+    ).encode()
+
+    attrs = _proxy_attrs_for_centralized_checks(user_custom_auth=None)
+    originals = {a: getattr(_proxy_server_mod, a, None) for a in attrs}
+    try:
+        for k, v in attrs.items():
+            setattr(_proxy_server_mod, k, v)
+        with (
+            patch(
+                "litellm.proxy.auth.user_api_key_auth._user_api_key_auth_builder",
+                new_callable=AsyncMock,
+                return_value=builder_token,
+            ),
+            patch(
+                "litellm.proxy.auth.user_api_key_auth._run_centralized_common_checks",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "litellm.proxy.auth.user_api_key_auth.RouteChecks.should_call_route",
+            ),
+            patch(
+                "litellm.proxy.auth.user_api_key_auth.resolve_and_validate_end_user_id",
+                new_callable=AsyncMock,
+            ) as mock_resolve,
+        ):
+            result = await user_api_key_auth(request=request, api_key="Bearer sk-test")
+
+        assert result.end_user_id == "builder-resolved-id"
+        mock_resolve.assert_not_awaited()
+    finally:
+        for k, v in originals.items():
+            setattr(_proxy_server_mod, k, v)