fix(proxy): run the remaining inline token counts off the event loop

Wrap the context-management editors, the end-of-stream chunk builder, acount_tokens, the compression interception hook, the passthrough interrupted-stream recovery, the A2A usage counters, and the semantic cache embedding truncation in asyncify so a multi-megabyte payload no longer stalls the worker's event loop while it is tokenized The pass-through suite now drains the process-global logging worker from an autouse conftest fixture so work queued on one test's loop cannot fire against the next test's callbacks Resolves LIT-7190 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-09 22:31:41 +00:00 · 2026-09-08 18:42:38 +00:00 · 2026-09-08 18:42:38 +00:00 · b5a7032eb4
commit b5a7032eb4
parent d202885f8b
22 changed files with 472 additions and 22 deletions
--- a/litellm/a2a_protocol/main.py
+++ b/litellm/a2a_protocol/main.py
@ -21,6 +21,7 @@ from litellm._logging import verbose_logger, verbose_proxy_logger
 from litellm.a2a_protocol.streaming_iterator import A2AStreamingIterator
 from litellm.a2a_protocol.utils import A2ARequestUtils
 from litellm.constants import DEFAULT_A2A_AGENT_TIMEOUT
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.llms.custom_httpx.http_handler import (
    get_async_httpx_client,
@ -507,7 +508,7 @@ async def asend_message(
        prompt_tokens,
        completion_tokens,
        _,
-    ) = A2ARequestUtils.calculate_usage_from_request_response(
+    ) = await asyncify(A2ARequestUtils.calculate_usage_from_request_response)(
        request=request,
        response_dict=response_dict,
    )
--- a/litellm/a2a_protocol/streaming_iterator.py
+++ b/litellm/a2a_protocol/streaming_iterator.py
@ -11,6 +11,7 @@ import litellm
 from litellm._logging import verbose_logger
 from litellm.a2a_protocol.cost_calculator import A2ACostCalculator
 from litellm.a2a_protocol.utils import A2ARequestUtils
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj

 if TYPE_CHECKING:
@ -99,11 +100,11 @@ class A2AStreamingIterator:
            # Calculate tokens from collected text
            input_message: Final = A2ARequestUtils.get_input_message_from_request(self.request)
            input_text: Final = A2ARequestUtils.extract_text_from_message(input_message)
-            prompt_tokens: Final = A2ARequestUtils.count_tokens(input_text)
+            prompt_tokens: Final = await asyncify(A2ARequestUtils.count_tokens)(input_text)

            # Use the last (most complete) text from chunks
            output_text: Final = self.collected_text_parts[-1] if self.collected_text_parts else ""
-            completion_tokens: Final = A2ARequestUtils.count_tokens(output_text)
+            completion_tokens: Final = await asyncify(A2ARequestUtils.count_tokens)(output_text)

            total_tokens: Final = prompt_tokens + completion_tokens

--- a/litellm/caching/qdrant_semantic_cache.py
+++ b/litellm/caching/qdrant_semantic_cache.py
@ -21,6 +21,7 @@ from litellm.constants import (
    QDRANT_VECTOR_SIZE,
    SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS,
 )
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
    get_str_from_messages,
 )
@ -255,7 +256,7 @@ class QdrantSemanticCache(BaseCache):
            llm_router = None

        router: Final = resolve_embedding_router(self.embedding_model, llm_router, llm_model_list)
-        embedding_input: Final = self._embedding_input(prompt, router)
+        embedding_input: Final = await asyncify(self._embedding_input)(prompt, router)
        embedding_call: Final = (
            router.aembedding(
                model=self.embedding_model,
--- a/litellm/caching/redis_semantic_cache.py
+++ b/litellm/caching/redis_semantic_cache.py
@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, Any, Final, cast
 import litellm
 from litellm._logging import print_verbose, verbose_logger
 from litellm.constants import SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
    get_str_from_messages,
 )
@ -522,7 +523,7 @@ class RedisSemanticCache(BaseCache):
            llm_router = None

        router: Final = resolve_embedding_router(self.embedding_model, llm_router, llm_model_list)
-        embedding_input: Final = self._embedding_input(prompt, router)
+        embedding_input: Final = await asyncify(self._embedding_input)(prompt, router)
        embedding_call: Final = (
            router.aembedding(
                model=self.embedding_model,
--- a/litellm/integrations/compression_interception/handler.py
+++ b/litellm/integrations/compression_interception/handler.py
@ -15,6 +15,7 @@ from typing_extensions import ReadOnly, TypedDict
 from litellm._logging import verbose_logger
 from litellm.compression import compress
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.types.integrations.compression_interception import (
    CompressionInterceptionConfig,
    CompressionSavingsMetadata,
@ -153,7 +154,7 @@ class CompressionInterceptionLogger(CustomLogger):

        self._prune_expired_cache()

-        compressed: Final = compress(
+        compressed: Final = await asyncify(compress)(
            messages=messages,
            model=model,
            call_type=CallTypes.anthropic_messages,
--- a/litellm/litellm_core_utils/streaming_handler.py
+++ b/litellm/litellm_core_utils/streaming_handler.py
@ -19,6 +19,7 @@ from typing_extensions import NotRequired, TypedDict
 import litellm
 from litellm import verbose_logger
 from litellm._uuid import uuid
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.model_response_utils import (
    is_model_response_stream_empty,
 )
@ -2247,7 +2248,7 @@ class CustomStreamWrapper:
        if self.sent_last_chunk is True:
            # log the final chunk with accurate streaming values
            try:
-                complete_streaming_response = litellm.stream_chunk_builder(
+                complete_streaming_response = await asyncify(litellm.stream_chunk_builder)(
                    chunks=self.chunks,
                    messages=self.messages,
                    logging_obj=self.logging_obj,
--- a/litellm/llms/anthropic/experimental_pass_through/context_management/dispatcher.py
+++ b/litellm/llms/anthropic/experimental_pass_through/context_management/dispatcher.py
@ -5,6 +5,7 @@ from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Final, TypeAlias

 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.types.llms.anthropic import AppliedEdit

 from .constants import CLEAR_TOOL_USES_EDIT_TYPE, COMPACT_EDIT_TYPE
@ -82,9 +83,9 @@ async def apply_context_management(
    """Run edits in order; return a single ``PolyfillResult``.

    The dispatcher is async so async editors (``compact_20260112``) can
-    ``await`` the configured summarization model. Sync editors are called
-    inline — ``inspect.iscoroutinefunction`` decides how each editor is
-    invoked.
+    ``await`` the configured summarization model. Sync editors run in a
+    worker thread so their token counts stay off the event loop;
+    ``inspect.iscoroutinefunction`` decides how each editor is invoked.
    """
    edits: Final = _normalize_spec(context_management_spec)
    if not edits:
@ -121,7 +122,7 @@ async def apply_context_management(
                user_api_key_auth=user_api_key_auth,
            )
            if editor_is_async
-            else editor(
+            else await asyncify(editor)(
                model=model,
                messages=current_messages,
                tools=tools,
--- a/litellm/llms/anthropic/experimental_pass_through/context_management/editors/compact.py
+++ b/litellm/llms/anthropic/experimental_pass_through/context_management/editors/compact.py
@ -20,6 +20,7 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack

 import litellm
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.types.llms.anthropic import (
    AppliedEdit,
    CompactionBlock,
@ -1157,7 +1158,7 @@ async def apply_compact_20260112(

    # Phase B: threshold check.
    try:
-        current_tokens = _count_effective_tokens(
+        current_tokens = await asyncify(_count_effective_tokens)(
            model=model,
            effective_messages=effective_messages,
            # ``augmented_system`` already carries the prior compaction summary
--- a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py
+++ b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py
@ -678,7 +678,7 @@ class BaseAnthropicMessagesStreamingIterator:
        """
        from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler

-        PassThroughStreamingHandler.schedule_stream_failure_logging(
+        await PassThroughStreamingHandler.schedule_stream_failure_logging(
            litellm_logging_obj=self.litellm_logging_obj,
            endpoint_type=EndpointType.ANTHROPIC,
            request_body=self.request_body,
--- a/litellm/main.py
+++ b/litellm/main.py
@ -67,7 +67,7 @@ from litellm.constants import (
 )
 from litellm.exceptions import LiteLLMUnknownProvider
 from litellm.integrations.custom_logger import CustomLogger
-from litellm.litellm_core_utils.asyncify import run_async_function
+from litellm.litellm_core_utils.asyncify import asyncify, run_async_function
 from litellm.litellm_core_utils.audio_utils.utils import (
    calculate_request_duration,
    get_audio_file_for_health_check,
@ -9076,7 +9076,7 @@ async def acount_tokens(
    fallback_messages = messages or []
    if system and fallback_messages:
        fallback_messages = [{"role": "system", "content": system}] + fallback_messages
-    local_count: Final = litellm.token_counter(
+    local_count: Final = await asyncify(litellm.token_counter)(
        model=model,
        messages=fallback_messages,
        tools=tools,
--- a/litellm/proxy/pass_through_endpoints/streaming_handler.py
+++ b/litellm/proxy/pass_through_endpoints/streaming_handler.py
@ -8,6 +8,7 @@ import httpx

 import litellm
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.asyncify import asyncify
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.proxy._types import PassThroughEndpointLoggingResultValues
@ -60,7 +61,7 @@ class PassThroughStreamingHandler:
            litellm_logging_obj._update_completion_start_time(completion_start_time=datetime.now())

    @staticmethod
-    def schedule_stream_failure_logging(
+    async def schedule_stream_failure_logging(
        litellm_logging_obj: LiteLLMLoggingObj,
        endpoint_type: EndpointType,
        request_body: dict[str, object],
@ -68,7 +69,7 @@ class PassThroughStreamingHandler:
        exception: Exception,
        stream_context: PassThroughStreamContext | None = None,
    ) -> None:
-        PassThroughStreamingHandler._record_partial_usage_for_failure(
+        await asyncify(PassThroughStreamingHandler._record_partial_usage_for_failure)(
            litellm_logging_obj=litellm_logging_obj,
            endpoint_type=endpoint_type,
            request_body=request_body,
@ -222,7 +223,7 @@ class PassThroughStreamingHandler:
            verbose_proxy_logger.error("Error in chunk_processor: %s", e)
            if response.status_code < 400:
                logging_scheduled = True
-                PassThroughStreamingHandler.schedule_stream_failure_logging(
+                await PassThroughStreamingHandler.schedule_stream_failure_logging(
                    litellm_logging_obj=litellm_logging_obj,
                    endpoint_type=endpoint_type,
                    request_body=resolved_request_body,
@ -274,7 +275,7 @@ class PassThroughStreamingHandler:
            (
                standard_logging_response_object,
                kwargs,
-            ) = PassThroughStreamingHandler._build_passthrough_logging_result(
+            ) = await asyncify(PassThroughStreamingHandler._build_passthrough_logging_result)(
                litellm_logging_obj=litellm_logging_obj,
                passthrough_success_handler_obj=passthrough_success_handler_obj,
                url_route=url_route,
@ -316,8 +317,8 @@ class PassThroughStreamingHandler:
        Synchronous, CPU-bound reconstruction of the standard logging payload
        from collected raw SSE bytes. Extracted from
        _route_streaming_logging_to_handler so the per-endpoint dispatch can
-        be unit-tested in isolation. Still invoked synchronously on the event
-        loop; an off-loop dispatch is a future change, not part of this PR.
+        be unit-tested in isolation. The async callers run it in a worker
+        thread so the token counts inside stay off the event loop.
        """
        all_chunks: Final = PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(raw_bytes)
        standard_logging_response_object: PassThroughEndpointLoggingResultValues | None = None
--- a/tests/pass_through_unit_tests/conftest.py
+++ b/tests/pass_through_unit_tests/conftest.py
@ -1,6 +1,8 @@
+import asyncio

 import pytest

+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER

 from tests._vcr_conftest_common import (  # noqa: E402,F401
    VerboseReporterState,
@ -45,6 +47,21 @@ def _vcr_outcome_gate(request, vcr):
    record_vcr_outcome(request, vcr)


+@pytest.fixture(autouse=True)
+async def _drain_logging_worker():
+    """
+    The logging queue is bound to the running loop, so anything left queued when a test's loop
+    goes away is carried onto the next loop and fires against that test's callbacks.
+    """
+    GLOBAL_LOGGING_WORKER.start()
+    try:
+        await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10)
+    except asyncio.TimeoutError:
+        pass
+    await GLOBAL_LOGGING_WORKER.stop()
+    yield
+
+
 def pytest_configure(config):
    _verbose_state.remember_pluginmanager(config)
    reset_vcr_diag_dir()
--- a/tests/test_litellm/a2a_protocol/test_a2a_streaming_iterator.py
+++ b/tests/test_litellm/a2a_protocol/test_a2a_streaming_iterator.py
@ -100,3 +100,57 @@ async def test_custom_logger_only_never_submits_sync_success_handler(monkeypatch

    assert recorder.async_hook_fired is True
    assert recording_executor.submitted_for(logging_obj) == []
+
+
+class _AgentChunk:
+    def __init__(self, text: str):
+        self._text = text
+
+    def model_dump(self, mode: str, exclude_none: bool) -> dict:
+        return {"result": {"kind": "message", "role": "agent", "parts": [{"kind": "text", "text": self._text}]}}
+
+
+@pytest.mark.asyncio
+async def test_stream_completion_counts_tokens_off_the_event_loop(monkeypatch):
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    warm_tokenizer("gpt-5.6-luna")
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "_async_success_callback", [])
+    logging_obj = LitellmLogging(
+        model="a2a/test-agent",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=True,
+        call_type="a2a_send_message_streaming",
+        start_time=time.time(),
+        litellm_call_id="lit-7190-test",
+        function_id="lit-7190-test",
+    )
+
+    async def _stream():
+        yield _AgentChunk(text * 100)
+
+    iterator = A2AStreamingIterator(
+        stream=_stream(),
+        request=SimpleNamespace(
+            params=SimpleNamespace(message={"role": "user", "parts": [{"kind": "text", "text": text * 100}]})
+        ),
+        logging_obj=logging_obj,
+        agent_name="test-agent",
+    )
+
+    async def drain() -> int:
+        return len([chunk async for chunk in iterator])
+
+    yielded, took, lags = await timed_with_loop_lags(drain)
+
+    assert yielded == 1
+    usage = logging_obj.model_call_details["usage"]
+    assert usage.prompt_tokens > 100_000
+    assert usage.completion_tokens > 100_000
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/a2a_protocol/test_main.py
+++ b/tests/test_litellm/a2a_protocol/test_main.py
@ -1,5 +1,7 @@
 """Tests for litellm/a2a_protocol/main.py non-streaming send behavior."""

+import asyncio
+
 import httpx
 import pytest

@ -13,7 +15,8 @@ from a2a.compat.v0_3.types import (
 )

 import litellm
-from litellm.a2a_protocol.main import _send_message, _stream_messages, create_a2a_client
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.a2a_protocol.main import _send_message, _stream_messages, asend_message, create_a2a_client
 from litellm.caching.llm_caching_handler import LLMClientCache
 from litellm.constants import DEFAULT_A2A_AGENT_TIMEOUT
 from litellm.llms.custom_httpx.http_handler import (
@ -413,3 +416,51 @@ async def test_the_pooled_a2a_client_arrives_with_cookie_persistence_disabled(is

    assert dict(handler.client.cookies) == {}, "the pooled A2A client kept an upstream's cookie"
    await handler.close()
+
+
+class _UsageRecorder(CustomLogger):
+    def __init__(self):
+        super().__init__()
+        self.logged = asyncio.Event()
+        self.payload = None
+
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        self.payload = kwargs["standard_logging_object"]
+        self.logged.set()
+
+
+@pytest.mark.asyncio
+async def test_asend_message_counts_usage_off_the_event_loop(monkeypatch):
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    warm_tokenizer("gpt-5.6-luna")
+    recorder = _UsageRecorder()
+    monkeypatch.setattr(litellm, "callbacks", [recorder])
+    monkeypatch.setattr(litellm, "success_callback", [recorder])
+    monkeypatch.setattr(litellm, "_async_success_callback", [recorder])
+
+    reply = _conv.pb2_v10.StreamResponse()
+    reply.message.message_id = "reply-1"
+    reply.message.role = _conv.pb2_v10.Role.ROLE_AGENT
+    reply.message.parts.add().text = text * 100
+    request = SendMessageRequest(
+        id="r1",
+        params=MessageSendParams(
+            message={"messageId": "m1", "role": "user", "parts": [{"kind": "text", "text": text * 100}]}
+        ),
+    )
+
+    response, took, lags = await timed_with_loop_lags(
+        lambda: asend_message(a2a_client=_FakeClient(reply), request=request)
+    )
+
+    assert response.id == "r1"
+    await asyncio.wait_for(recorder.logged.wait(), timeout=10)
+    assert recorder.payload["prompt_tokens"] > 100_000
+    assert recorder.payload["completion_tokens"] > 100_000
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/caching/test_qdrant_semantic_cache.py
+++ b/tests/test_litellm/caching/test_qdrant_semantic_cache.py
@ -1026,3 +1026,36 @@ def test_qdrant_semantic_cache_defaults_embedding_timeout():
    cache = QdrantSemanticCache.__new__(QdrantSemanticCache)
    assert cache.embedding_timeout == SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS
    assert SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS < 60
+
+
+@pytest.mark.asyncio
+async def test_qdrant_async_embedding_truncates_off_the_event_loop(monkeypatch):
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    from litellm.caching.qdrant_semantic_cache import QdrantSemanticCache
+
+    warm_tokenizer("sem-embed")
+    cache = QdrantSemanticCache.__new__(QdrantSemanticCache)
+    cache.embedding_model = "sem-embed"
+    cache.embedding_max_input_tokens = 5
+    cache.embedding_timeout = 5
+
+    router = MagicMock()
+    router.get_configured_token_limits.return_value = (8191, None)
+    router.aembedding = AsyncMock(return_value={"data": [{"embedding": [0.1, 0.2]}]})
+    monkeypatch.setitem(
+        sys.modules,
+        "litellm.proxy.proxy_server",
+        _router_proxy_module(router, "sem-embed"),
+    )
+
+    response, took, lags = await timed_with_loop_lags(lambda: cache._get_async_embedding(text * 100))
+
+    assert response["data"][0]["embedding"] == [0.1, 0.2]
+    assert _token_count("sem-embed", router.aembedding.call_args.kwargs["input"]) == 5
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/caching/test_redis_semantic_cache.py
+++ b/tests/test_litellm/caching/test_redis_semantic_cache.py
@ -1472,3 +1472,32 @@ def test_redis_semantic_cache_defaults_embedding_timeout():
    cache = RedisSemanticCache.__new__(RedisSemanticCache)
    assert cache.embedding_timeout == SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS
    assert SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS < 60
+
+
+@pytest.mark.asyncio
+async def test_redis_async_embedding_truncates_off_the_event_loop(monkeypatch):
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    from litellm.caching.redis_semantic_cache import RedisSemanticCache
+
+    warm_tokenizer("sem-embed")
+    cache = RedisSemanticCache.__new__(RedisSemanticCache)
+    cache.embedding_model = "sem-embed"
+    cache.embedding_max_input_tokens = 5
+    cache.embedding_timeout = 5
+
+    router = MagicMock()
+    router.get_configured_token_limits.return_value = (8191, None)
+    router.aembedding = AsyncMock(return_value={"data": [{"embedding": [0.1, 0.2]}]})
+    _proxy_with_router(monkeypatch, router, "sem-embed")
+
+    embedding, took, lags = await timed_with_loop_lags(lambda: cache._get_async_embedding(text * 100))
+
+    assert embedding == [0.1, 0.2]
+    assert _token_count("sem-embed", router.aembedding.call_args.kwargs["input"]) == 5
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/integrations/compression_interception/test_compression_interception_handler.py
+++ b/tests/test_litellm/integrations/compression_interception/test_compression_interception_handler.py
@ -523,3 +523,28 @@ async def test_pre_call_hook_no_compression_records_no_savings(monkeypatch):
    await logger.async_pre_call_deployment_hook(kwargs=kwargs, call_type=CallTypes.anthropic_messages)

    assert "compression_savings" not in litellm_metadata
+
+
+@pytest.mark.asyncio
+async def test_pre_call_hook_counts_tokens_off_the_event_loop():
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    model = "anthropic/claude-fable-5"
+    warm_tokenizer(model)
+    logger = CompressionInterceptionLogger(compression_trigger=10_000_000)
+    messages = [{"role": "user", "content": text * 100}]
+    kwargs = {"model": model, "messages": messages}
+
+    result, took, lags = await timed_with_loop_lags(
+        lambda: logger.async_pre_call_deployment_hook(kwargs=kwargs, call_type=CallTypes.anthropic_messages)
+    )
+
+    assert result is not None
+    assert result["messages"] is messages
+    assert "tools" not in result
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/litellm_core_utils/test_streaming_handler.py
+++ b/tests/test_litellm/litellm_core_utils/test_streaming_handler.py
@ -4875,3 +4875,50 @@ class TestStableStreamingResponseId:
        )
        wrapper.response_id = "chatcmpl-from-provider"
        assert wrapper.model_response_creator().id == "chatcmpl-from-provider"
+
+
+@pytest.mark.asyncio
+async def test_async_stream_without_usage_counts_tokens_off_the_event_loop():
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    model = "gpt-5.6-luna"
+    warm_tokenizer(model)
+    messages = [{"role": "user", "content": text * 100}]
+    content_chunks = [_make_chunk(text) for _ in range(100)]
+    stop_chunk = ModelResponseStream(
+        id="test",
+        created=1741037890,
+        model=model,
+        choices=[StreamingChoices(index=0, delta=Delta(content=""), finish_reason="stop")],
+    )
+    logging_obj = Logging(
+        model=model,
+        messages=messages,
+        stream=True,
+        call_type="acompletion",
+        start_time=time.time(),
+        litellm_call_id="12345",
+        function_id="1245",
+    )
+    wrapper = CustomStreamWrapper(
+        completion_stream=ModelResponseListIterator(model_responses=content_chunks + [stop_chunk]),
+        model=model,
+        custom_llm_provider="openai",
+        logging_obj=logging_obj,
+        stream_options={"include_usage": True},
+    )
+
+    async def consume() -> list[ModelResponseStream]:
+        return [chunk async for chunk in wrapper]
+
+    chunks, took, lags = await timed_with_loop_lags(consume)
+
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks) == text * 100
+    assert chunks[-1].usage.prompt_tokens > 100_000
+    assert chunks[-1].usage.completion_tokens > 100_000
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/llms/anthropic/experimental_pass_through/context_management/test_compact.py
+++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/context_management/test_compact.py
@ -2605,3 +2605,34 @@ def test_build_summary_messages_keeps_midturn_system_correction_in_place():
    assert summary_messages[0]["content"] == "caller system prompt"
    assert summary_messages[2]["content"] == "use the corrected result"
    assert summary_messages[-1]["content"] == "summarize the conversation"
+
+
+async def test_threshold_check_counts_tokens_off_the_event_loop(monkeypatch):
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    from litellm.llms.anthropic.experimental_pass_through.context_management.constants import (
+        COMPACT_SUMMARY_MODEL_SETTING_KEY,
+    )
+    from litellm.proxy.proxy_server import general_settings
+
+    monkeypatch.setitem(general_settings, COMPACT_SUMMARY_MODEL_SETTING_KEY, "claude-haiku-4-5")
+    warm_tokenizer(MODEL)
+    messages = [{"role": "user", "content": text * 100}, *_simple_messages()]
+    result, took, lags = await timed_with_loop_lags(
+        lambda: apply_compact_20260112(
+            model=MODEL,
+            messages=messages,
+            tools=None,
+            system=None,
+            edit_spec={"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 10_000_000}},
+        )
+    )
+
+    assert result.messages == messages
+    assert result.compaction_block is None
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/llms/anthropic/experimental_pass_through/context_management/test_dispatcher.py
+++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/context_management/test_dispatcher.py
@ -129,3 +129,35 @@ async def test_malformed_edit_entries_are_skipped():
    )
    assert result.applied_edits == []
    assert result.messages == messages
+
+
+async def test_sync_editor_counts_tokens_off_the_event_loop():
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    warm_tokenizer(MODEL)
+    messages = [{"role": "user", "content": text * 100}, *_history_with_two_tool_pairs()]
+
+    result, took, lags = await timed_with_loop_lags(
+        lambda: apply_context_management(
+            model=MODEL,
+            messages=messages,
+            tools=None,
+            system=None,
+            context_management_spec={
+                "edits": [
+                    {
+                        "type": "clear_tool_uses_20250919",
+                        "trigger": {"type": "input_tokens", "value": 10_000_000},
+                    }
+                ]
+            },
+        )
+    )
+
+    assert result.messages == messages
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/proxy/pass_through_endpoints/test_streaming_handler.py
+++ b/tests/test_litellm/proxy/pass_through_endpoints/test_streaming_handler.py
@ -7,6 +7,7 @@ import pytest

 import litellm
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.proxy.pass_through_endpoints.llm_provider_handlers.vertex_passthrough_logging_handler import (
    VertexPassthroughLoggingHandler,
 )
@ -128,3 +129,104 @@ def test_vertex_generate_content_payload_prices_gemini_urls_at_gemini_rates():

    assert result["kwargs"]["response_cost"] == pytest.approx(GEMINI_COST)
    assert logging_obj.model_call_details["custom_llm_provider"] == "gemini"
+
+
+def _interrupted_anthropic_stream(model: str, output_text: str) -> list[bytes]:
+    def sse(event: str, data: dict) -> bytes:
+        return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()
+
+    message_start = {
+        "type": "message_start",
+        "message": {
+            "id": "msg_interrupted",
+            "type": "message",
+            "role": "assistant",
+            "model": model,
+            "content": [],
+            "stop_reason": None,
+            "stop_sequence": None,
+            "usage": {"input_tokens": 29, "output_tokens": 2},
+        },
+    }
+    block_start = {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}
+    delta = {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": output_text}}
+    return [
+        sse("message_start", message_start),
+        sse("content_block_start", block_start),
+        sse("content_block_delta", delta),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_interrupted_anthropic_stream_recovers_output_tokens_off_the_event_loop():
+    from unittest.mock import AsyncMock
+
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    model = "claude-fable-5"
+    warm_tokenizer(model)
+    logging_obj = _logging_obj()
+    logging_obj.model_call_details = {"model": model, "stream": True}
+    logging_obj.litellm_params = {}
+    logging_obj.get_router_model_id.return_value = None
+    logging_obj.dispatch_success_handlers = AsyncMock()
+
+    _, took, lags = await timed_with_loop_lags(
+        lambda: PassThroughStreamingHandler._route_streaming_logging_to_handler(
+            litellm_logging_obj=logging_obj,
+            passthrough_success_handler_obj=PassThroughEndpointLogging(),
+            url_route="/anthropic/v1/messages",
+            request_body={"model": model, "stream": True},
+            endpoint_type=EndpointType.ANTHROPIC,
+            start_time=datetime.now(),
+            raw_bytes=_interrupted_anthropic_stream(model, text * 100),
+            end_time=datetime.now(),
+            model=model,
+        )
+    )
+
+    logging_obj.dispatch_success_handlers.assert_awaited_once()
+    logged_usage = logging_obj.dispatch_success_handlers.await_args.kwargs["result"].usage
+    assert logged_usage.completion_tokens > 100_000
+    assert_loop_stayed_free(took, lags)
+
+
+@pytest.mark.asyncio
+async def test_failed_anthropic_stream_records_partial_usage_off_the_event_loop():
+    from unittest.mock import AsyncMock
+
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    model = "claude-fable-5"
+    warm_tokenizer(model)
+    logging_obj = _logging_obj()
+    logging_obj.model_call_details = {"model": model, "stream": True}
+    logging_obj.litellm_params = {}
+    logging_obj.get_router_model_id.return_value = None
+    logging_obj.dispatch_failure_handlers = AsyncMock()
+
+    _, took, lags = await timed_with_loop_lags(
+        lambda: PassThroughStreamingHandler.schedule_stream_failure_logging(
+            litellm_logging_obj=logging_obj,
+            endpoint_type=EndpointType.ANTHROPIC,
+            request_body={"model": model, "stream": True},
+            raw_bytes=_interrupted_anthropic_stream(model, text * 100),
+            exception=RuntimeError("upstream closed the stream"),
+        )
+    )
+    await GLOBAL_LOGGING_WORKER.flush()
+
+    logging_obj.dispatch_failure_handlers.assert_awaited_once()
+    partial_usage = logging_obj.record_partial_usage_for_failure.call_args.kwargs["usage"]
+    assert partial_usage.completion_tokens > 100_000
+    assert_loop_stayed_free(took, lags)
--- a/tests/test_litellm/test_count_tokens_public_api.py
+++ b/tests/test_litellm/test_count_tokens_public_api.py
@ -155,3 +155,23 @@ def test_acount_tokens_no_api_key_falls_back(monkeypatch):
    # Should fall back to local tokenizer since no API key
    assert result.total_tokens > 0
    assert result.tokenizer_type == "local_tokenizer"
+
+
+async def test_acount_tokens_local_fallback_counts_off_the_event_loop():
+    from tests.large_text import text
+    from tests.test_litellm.litellm_core_utils.event_loop_lag import (
+        assert_loop_stayed_free,
+        timed_with_loop_lags,
+        warm_tokenizer,
+    )
+
+    model = "together_ai/meta-llama/Llama-3-8b-chat-hf"
+    warm_tokenizer(model)
+
+    result, took, lags = await timed_with_loop_lags(
+        lambda: litellm.acount_tokens(model=model, messages=[{"role": "user", "content": text * 100}])
+    )
+
+    assert result.tokenizer_type == "local_tokenizer"
+    assert result.total_tokens > 100_000
+    assert_loop_stayed_free(took, lags)