test(mcp): drain the logging worker after each test so queued callbacks cannot leak into the next test (#38228)

LoggingWorker now carries still-queued coroutines onto the next event loop (12a34a10d8). Under xdist,
a success-logging coroutine queued by test_acompletion_mcp_respects_manual_approval ran nine seconds
later inside test_mcp_tool_call_hook on the same worker, resolved litellm.callbacks at run time and
overwrote that test's captured payload with a gpt-4o-mini completion (assert 1.35e-05 == 1.42).

Run clear_queue() in the suite's autouse teardown so every coroutine a test enqueues finishes before the
next test registers its callbacks, and add a subprocess regression test that runs the real conftest
against a stopped worker with work still queued.

tests/mcp_tests/conftest.py

@@ -7,6 +7,7 @@ import pytest
 
 import litellm
 import asyncio
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 
 
 @pytest.fixture(scope="session")
@@ -38,6 +39,8 @@ def setup_and_teardown():
     yield
 
     # Teardown code (executes after the yield point)
+    # LoggingWorker carries still-queued coroutines onto the next test's loop, where they'd log into that test's callbacks
+    asyncio.run(GLOBAL_LOGGING_WORKER.clear_queue())
     loop.close()  # Close the loop created earlier
     asyncio.set_event_loop(None)  # Remove the reference to the loop
 

tests/mcp_tests/test_mcp_logging.py

@@ -1,6 +1,9 @@
 import os
 import pytest
 import asyncio
+import subprocess
+import sys
+from pathlib import Path
 from typing import Optional
 from unittest.mock import AsyncMock, patch
 
@@ -458,3 +461,45 @@ async def test_mcp_tool_call_hook():
                 logged_standard_logging_payload is not None
             ), "Standard logging payload should not be None"
             assert logged_standard_logging_payload["response_cost"] == 1.42
+
+
+_QUEUED_LOGGING_OUTLIVES_TEST = '''
+import time
+
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+
+ran_at = []
+
+
+async def _record_run():
+    ran_at.append(time.monotonic())
+
+
+async def test_1_leaves_logging_queued_behind_a_stopped_worker():
+    GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(_record_run())
+    await GLOBAL_LOGGING_WORKER.stop()
+    assert ran_at == []
+
+
+async def test_2_starts_after_the_previous_tests_logging_ran():
+    started_at = time.monotonic()
+    GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(_record_run())
+    await GLOBAL_LOGGING_WORKER.flush()
+    assert [t < started_at for t in ran_at] == [True, False]
+'''
+
+
+def test_logging_queued_by_one_test_is_drained_before_the_next(tmp_path: Path):
+    """Regression: a logging coroutine queued by one test must not run inside a later test (it would log into that
+    test's callbacks, which is how test_mcp_tool_call_hook captured a gpt-4o-mini payload under xdist)."""
+    (tmp_path / "conftest.py").write_text((Path(__file__).parent / "conftest.py").read_text())
+    (tmp_path / "pyproject.toml").write_text('[tool.pytest.ini_options]\nasyncio_mode = "auto"\n')
+    (tmp_path / "test_queued_logging.py").write_text(_QUEUED_LOGGING_OUTLIVES_TEST)
+    result = subprocess.run(
+        [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", "test_queued_logging.py"],
+        cwd=tmp_path,
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr