fix(proxy): keep a team's empty logging list as the disabled state instead of falling back to config callbacks

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/litellm_pre_call_utils.py

@@ -905,8 +905,9 @@ class KeyAndTeamLoggingSettings:
     """
     Helper class to get the dynamic logging settings for the key and team
 
-    An empty ``logging`` list is the same as no ``logging`` key: both return ``None`` so the
-    caller falls through to the next level. Disabling a callback is ``litellm_disabled_callbacks``.
+    A key's empty ``logging`` list is unset (``None``) and falls through to the team; disabling a
+    callback on a key is ``litellm_disabled_callbacks``. A team's ``logging: []`` is the state
+    ``POST /team/{team_id}/disable_logging`` persists, so it is kept and stops the fallthrough.
     """
 
     @staticmethod
@@ -918,7 +919,7 @@ class KeyAndTeamLoggingSettings:
     @staticmethod
     def get_team_dynamic_logging_settings(user_api_key_dict: UserAPIKeyAuth):
         if user_api_key_dict.team_metadata is not None and "logging" in user_api_key_dict.team_metadata:
-            return decrypt_callback_vars(user_api_key_dict.team_metadata).get("logging") or None
+            return decrypt_callback_vars(user_api_key_dict.team_metadata).get("logging")
         return None
 
 

tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py

@@ -18223,6 +18223,30 @@ async def test_key_health_tests_the_team_callbacks_an_empty_key_logging_list_fal
     assert test_logging.await_args.kwargs["logging_callbacks"] == ("gcs_bucket",)
 
 
+@pytest.mark.asyncio
+async def test_key_health_tests_a_valid_key_callback_instead_of_the_team_default():
+    from litellm.proxy.management_endpoints.key_management_endpoints import key_health
+
+    caller: Final = UserAPIKeyAuth(
+        api_key="sk-1",
+        team_id="team-gcs",
+        metadata={"logging": [{"callback_name": "langfuse", "callback_vars": {"langfuse_public_key": "pk"}}]},
+        team_metadata={},
+    )
+    logging_status: Final = LoggingCallbackStatus(callbacks=("langfuse",), status="healthy", details="ok")
+    with (
+        patch("litellm.proxy.proxy_server.proxy_config", _default_team_gcs_proxy_config("team-gcs")),  # test-quality-ok: key_health reads the module-level proxy config
+        patch(  # test-quality-ok: the mock completion behind test_key_logging needs a running proxy
+            "litellm.proxy.management_endpoints.key_management_endpoints.test_key_logging",
+            AsyncMock(return_value=logging_status),
+        ) as test_logging,
+    ):
+        response = await key_health(request=MagicMock(), user_api_key_dict=caller)
+
+    assert response == KeyHealthResponse(key="healthy", logging_callbacks=logging_status)
+    assert test_logging.await_args.kwargs["logging_callbacks"] == ("langfuse",)
+
+
 @pytest.mark.asyncio
 async def test_key_health_without_any_effective_callbacks_reports_healthy_and_sends_no_test_log():
     from litellm.proxy.management_endpoints.key_management_endpoints import key_health
@@ -18287,21 +18311,53 @@ async def test_key_health_rejects_key_logging_entries_without_a_callback_name():
     assert "callback_name is required" in exc.value.message
 
 
-@pytest.mark.asyncio
-async def test_flush_gcs_reports_the_failed_upload_count_from_the_registered_logger():
+def _gcs_logger_whose_flush_reports(sent: int, failed: int):
     from litellm.integrations.gcs_bucket.gcs_bucket import GCSBucketLogger
-    from litellm.proxy.management_endpoints.key_management_endpoints import flush_gcs_and_describe_failures
     from litellm.types.integrations.gcs_bucket import GCSFlushResult
 
-    class _StuckUploadGCSLogger(GCSBucketLogger):
+    class _FixedFlushGCSLogger(GCSBucketLogger):
         def __init__(self) -> None:
             with patch("litellm.proxy.proxy_server.premium_user", True):  # test-quality-ok: GCS logging is premium-gated
                 super().__init__(bucket_name="test-bucket")
 
         async def flush_queue_and_report(self) -> GCSFlushResult:
-            return GCSFlushResult(sent=0, failed=3)
+            return GCSFlushResult(sent=sent, failed=failed)
 
-    assert await flush_gcs_and_describe_failures(_StuckUploadGCSLogger()) == "GCS upload failed for 3 event(s), 0 uploaded"
+    return _FixedFlushGCSLogger()
+
+
+@pytest.mark.asyncio
+async def test_flush_gcs_reports_the_failed_upload_count_from_the_registered_logger():
+    from litellm.proxy.management_endpoints.key_management_endpoints import flush_gcs_and_describe_failures
+
+    assert (
+        await flush_gcs_and_describe_failures(_gcs_logger_whose_flush_reports(sent=0, failed=3))
+        == "GCS upload failed for 3 event(s), 0 uploaded"
+    )
+    assert await flush_gcs_and_describe_failures(_gcs_logger_whose_flush_reports(sent=2, failed=0)) is None
+
+
+@pytest.mark.asyncio
+async def test_key_logging_marks_the_key_unhealthy_when_the_gcs_flush_leaves_events_undelivered():
+    from starlette.requests import Request as StarletteRequest
+
+    from litellm.proxy.management_endpoints.key_management_endpoints import test_key_logging
+
+    request: Final = StarletteRequest(
+        {"type": "http", "method": "POST", "path": "/key/health", "headers": [], "query_string": b""}
+    )
+    caller: Final = UserAPIKeyAuth(api_key="sk-1", team_id="team-gcs", metadata={"logging": []}, team_metadata={})
+    with (
+        patch("litellm.proxy.proxy_server.proxy_config", _default_team_gcs_proxy_config("team-gcs")),  # test-quality-ok: test_key_logging reads the module-level proxy config
+        patch(  # test-quality-ok: the registered logger is a process-wide registry, not an injectable
+            "litellm.litellm_core_utils.litellm_logging.get_custom_logger_compatible_class",
+            return_value=_gcs_logger_whose_flush_reports(sent=1, failed=3),
+        ),
+    ):
+        status = await test_key_logging(user_api_key_dict=caller, request=request, logging_callbacks=("gcs_bucket",))
+
+    assert status["status"] == "unhealthy"
+    assert "GCS upload failed for 3 event(s), 1 uploaded" in (status["details"] or "")
 
 
 @pytest.mark.asyncio

tests/test_litellm/proxy/test_litellm_pre_call_utils.py

@@ -2170,12 +2170,12 @@ def test_key_dynamic_logging_settings():
     assert result is None
 
 
-def test_empty_logging_list_on_key_and_team_is_unset():
-    """A UI-generated `logging: []` is the same as no logging metadata, not an explicit override"""
+def test_empty_key_logging_list_is_unset_while_empty_team_logging_list_is_kept():
+    """A UI-generated key `logging: []` is no override; a team's `logging: []` is /disable_logging's state"""
     auth = UserAPIKeyAuth(api_key="test-key", metadata={"logging": []}, team_metadata={"logging": []})
 
     assert KeyAndTeamLoggingSettings.get_key_dynamic_logging_settings(auth) is None
-    assert KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(auth) is None
+    assert KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings(auth) == []
 
 
 def test_empty_key_logging_falls_back_to_team_logging():
@@ -2202,7 +2202,7 @@ def test_empty_key_logging_falls_back_to_team_logging():
     assert result.callback_vars == {"gcs_bucket_name": "team-bucket"}
 
 
-def test_empty_key_and_team_logging_falls_back_to_default_team_settings():
+def _default_team_gcs_config():
     from litellm.proxy.proxy_server import ProxyConfig
 
     pc = ProxyConfig()
@@ -2218,14 +2218,13 @@ def test_empty_key_and_team_logging_falls_back_to_default_team_settings():
             ]
         }
     }
-    auth = UserAPIKeyAuth(
-        api_key="test-key",
-        team_id="team-gcs",
-        metadata={"logging": []},
-        team_metadata={"logging": []},
-    )
+    return pc
 
-    result = _get_dynamic_logging_metadata(user_api_key_dict=auth, proxy_config=pc)
+
+def test_empty_key_logging_without_team_logging_falls_back_to_default_team_settings():
+    auth = UserAPIKeyAuth(api_key="test-key", team_id="team-gcs", metadata={"logging": []}, team_metadata={})
+
+    result = _get_dynamic_logging_metadata(user_api_key_dict=auth, proxy_config=_default_team_gcs_config())
 
     assert result is not None
     assert result.success_callback == ["gcs_bucket"]
@@ -2233,6 +2232,25 @@ def test_empty_key_and_team_logging_falls_back_to_default_team_settings():
     assert result.callback_vars == {"turn_off_message_logging": "True"}
 
 
+@pytest.mark.parametrize(
+    "team_metadata",
+    [
+        {"logging": []},
+        {"logging": [], "callback_settings": {"success_callback": [], "failure_callback": []}},
+    ],
+    ids=["last team callback removed", "POST /team/{team_id}/disable_logging"],
+)
+def test_team_with_logging_disabled_does_not_inherit_default_team_settings(team_metadata: dict):
+    auth = UserAPIKeyAuth(
+        api_key="test-key", team_id="team-gcs", metadata={"logging": []}, team_metadata=team_metadata
+    )
+
+    result = _get_dynamic_logging_metadata(user_api_key_dict=auth, proxy_config=_default_team_gcs_config())
+
+    effective = () if result is None else (*(result.success_callback or ()), *(result.failure_callback or ()))
+    assert effective == ()
+
+
 def test_team_dynamic_logging_settings():
     """
     Test KeyAndTeamLoggingSettings.get_team_dynamic_logging_settings method with arize and langfuse callbacks