fix(logging_worker): carry queued tasks across event-loop change instead of dropping them

[mateo-berri]

TLDR

Problem this solves:

• A loop change silently drops every queued spend-logging coroutine
• Missing spend rows and observability events, no error raised
• flush() reports success on a queue whose contents were discarded

How it solves it:

• Drain the stale queue, move pending tasks onto the new loop's queue
• Warn with the carried-over count instead of a silent debug log
• flush()/join() stay honest since nothing is thrown away

User Flow

Before: a developer using the LiteLLM Python SDK who runs each request in its own asyncio.run() loop loses the background spend/callback logging for earlier requests

1. They make an async LiteLLM completion inside asyncio.run() for request A; its success callback is queued for background logging, but the loop closes before the queue drains
2. They make another async LiteLLM completion inside a fresh asyncio.run() for request B
3. Python prints RuntimeWarning: coroutine 'Logging.async_success_handler' was never awaited to stderr
4. Request A's spend row and observability event never reach their logging backend, and nothing errors, so the loss is invisible

After: the same two requests both log

1. They make the async LiteLLM completion inside asyncio.run() for request A; its success callback is queued
2. They make the async LiteLLM completion inside a fresh asyncio.run() for request B
3. Instead of the warning they see LoggingWorker: event loop changed; carried N pending logging task(s) onto the new loop, and no never awaited warning appears
4. Request A's spend row and observability event are delivered; both requests show up in their logging backend

Relevant issues

Linear ticket

Resolves LIT-6028

Pre-Submission checklist

Please complete all items before asking a LiteLLM maintainer to review your PR

• I have added meaningful tests
• The handful of test files covering my change pass locally, e.g. uv run pytest tests/test_litellm/<your_test_file>.py -v. Leave the suites (make test-unit-*, make test-unit) to CI: it finishes in ~15 minutes where a laptop takes an hour or more
• My PR passes all required CI/CD checks (e.g., lint, schema.d.ts sync check, etc.)
• My PR's scope is as isolated as possible; it only solves 1 specific problem
• I have received a Greptile Confidence Score of at least 4/5 before requesting a maintainer review (Greptile reviews automatically once the PR is opened; only comment @greptileai to re-request a review after pushing changes)

Screenshots / Proof of Fix

Shared setup: a standalone script enqueues 5 spend-logging coroutines on one asyncio.run() loop, then triggers a second asyncio.run() loop (the exact "fresh loop per request" trigger from the ticket) and reports how many survived, how many ran, and how many never awaited warnings fired.

# repro_logging_worker.py
import asyncio
import gc
import warnings

from litellm.litellm_core_utils.logging_worker import LoggingWorker

executed: list[int] = []


async def spend_log(i: int) -> None:
    executed.append(i)


worker = LoggingWorker()
N = 5


def enqueue_on_fresh_loop() -> int:
    async def run() -> int:
        worker._ensure_queue()
        for i in range(N):
            worker.enqueue(spend_log(i))
        return worker._queue.qsize()

    return asyncio.run(run())


def rebind_on_fresh_loop_and_drain() -> int:
    async def run() -> int:
        worker._ensure_queue()
        carried = worker._queue.qsize()
        while not worker._queue.empty():
            task = worker._queue.get_nowait()
            await task["context"].run(asyncio.create_task, task["coroutine"])
        return carried

    return asyncio.run(run())


with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    queued = enqueue_on_fresh_loop()
    print(f"queued on loop A: {queued}")
    carried = rebind_on_fresh_loop_and_drain()
    gc.collect()
    print(f"carried onto loop B: {carried}")
    print(f"executed spend-log coroutines: {sorted(executed)}")
    never_awaited = [str(w.message) for w in caught
                     if issubclass(w.category, RuntimeWarning) and "never awaited" in str(w.message)]
    print(f"'never awaited' RuntimeWarnings: {len(never_awaited)}")

lost = N - len(executed)
print("VERDICT:", f"{lost} of {N} queued spend-log coroutines were DROPPED" if lost else "all queued coroutines survived the loop change")

Before (85d5ac2)

1. python repro_logging_worker.py
2. Observed output:

queued on loop A: 5
carried onto loop B: 0
executed spend-log coroutines: []
'never awaited' RuntimeWarnings: 5
VERDICT: 5 of 5 queued spend-log coroutines were DROPPED

All 5 queued spend-logging coroutines are discarded on the loop change and Python reports each as coroutine 'spend_log' was never awaited.

After (12a34a1)

1. python repro_logging_worker.py
2. Observed output:

LiteLLM:WARNING: logging_worker.py:94 - LoggingWorker: event loop changed; carried 5 pending logging task(s) onto the new loop
queued on loop A: 5
carried onto loop B: 5
executed spend-log coroutines: [0, 1, 2, 3, 4]
'never awaited' RuntimeWarnings: 0
VERDICT: all queued coroutines survived the loop change

All 5 tasks are carried onto the new loop and execute there, with zero never awaited warnings and a single warning-level line reporting the carried-over count.

Type

🐛 Bug Fix

Caveats (if any)

• Only the loop-change path changes; steady-state single-loop behavior is unchanged

Final Attestation

• The tests check the right things, including the edge cases, and regressions in the respective real-world customer use-cases are not possible after this PR

---

Note

Medium Risk
Touches hot-path queue initialization for background logging; behavior change is scoped to event-loop changes, but incorrect carry-over could affect spend/callback delivery or shutdown semantics.

Overview
Fixes silent loss of background spend/observability logging when the global LoggingWorker sees a new asyncio event loop (e.g. repeated asyncio.run() per request).

On loop rebind, _ensure_queue no longer discards the old queue. It drains pending LoggingTask entries via new _drain_pending, re-enqueues them on a fresh loop-bound queue, resets semaphore/worker state, and logs a warning with the carried-over count when anything was pending.

Adds regression test test_event_loop_change_carries_pending_tasks_over (LIT-6028) so queued coroutines still run after rebind instead of triggering never awaited warnings.

^{Reviewed by Cursor Bugbot for commit 12a34a1. Bugbot is set up for automated code reviews on this repo. Configure here.}

[mateo-berri]

bugbot run

[greptile-apps]

Greptile Summary

The PR preserves pending logging coroutines when a shared LoggingWorker encounters a new event loop instead of discarding its queue.

• Drains pending entries from the stale queue and transfers them to a replacement queue bound to the current loop.
• Reinitializes loop-bound worker state and reports the number of carried tasks.
• Adds a regression test covering sequential asyncio.run() loops and successful execution of all transferred coroutines.

Confidence Score: 5/5

The PR appears safe to merge because the replacement queue retains pending logging work and the normal start path recreates its loop-bound worker state.

The changed implementation transfers all bounded pending entries to the new queue, and production enqueue paths initialize a worker on the current loop before adding work; the regression test verifies the intended sequential-loop scenario.

Important Files Changed

Filename	Overview
litellm/litellm_core_utils/logging_worker.py	Migrates queued logging work during event-loop rebinding while preserving the existing worker initialization flow; no actionable changed-code defect was established.
tests/test_litellm/litellm_core_utils/test_logging_worker.py	Adds focused regression coverage showing pending coroutines survive a transition between sequential event loops.

_{Reviews (1): Last reviewed commit: "fix(logging_worker): carry queued tasks ..." | Re-trigger Greptile}

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit 12a34a1. Configure here.}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[tin-berri]

Approving.

Checked the three things that could have gone wrong with the early return:

• It doesn't skip worker startup — start() calls _ensure_queue() and then sets up _sem / _worker_task after it, so the loop-change branch returning early still lands in the same initialization.
• put_nowait can't raise QueueFull on the carry-over: carried_over is drained from a queue with the same maxsize, so the new queue can hold all of it.
• contextvars.Context isn't loop-bound, so task["context"].run(...) on the new loop is fine — the carried tasks keep their original context.

The regression test is the right shape: it proves the coroutines actually execute on the second loop rather than just asserting qsize().

code-quality is red on recursive_detector flagging _flatten_form_field / _flatten_form_data_field in litellm/litellm_core_utils/llm_request_utils.py — untouched by this PR, so that's base drift, not you.

[codspeed-hq]

Merging this PR will not alter performance

✅ 31 untouched benchmarks

---

_{Comparing litellm_fix_logging_worker_loop_drop (12a34a1) with litellm_internal_staging (a91cac7)¹}

Footnotes

1. No successful run was found on litellm_internal_staging (12a34a1) during the generation of this report, so a91cac7 was used instead as the comparison base. There might be some changes unrelated to this pull request in this report. ↩