fix(responses): stop scheduling sync success_handler concurrently with async_success_handler

[yassin-berriai]

Relevant issues

Linear ticket

Resolves LIT-4210

Pre-Submission checklist

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

• I have added meaningful tests
• My PR passes all CI/CD checks (e.g., lint, format, unit tests)
• My PR's scope is as isolated as possible; it only solves 1 specific problem
• I have requested a Greptile review by commenting @greptileai and received a Confidence Score of at least 4/5 before requesting a maintainer review

Delays in PR merge?

If you're seeing a delay in your PR being merged, ping the LiteLLM Team on Slack (#pr-review).

Screenshots / Proof of Fix

A customer's proxy pods were segfaulting (exit 139) roughly 18 times a day. They isolated it to callback registration: a bare no-op CustomLogger() with zero overrides crashed pods within 30 to 90 minutes, while a config with only prometheus ran 19 hours clean. Their faulthandler dumps from 9 crashed pods all show the main thread inside pydantic __setattr__ / _check_frozen while thread-pool workers run litellm's sync success_handler

The root cause is that several streaming iterators fire both success handlers at end of stream, unconditionally and with the same response object

asyncio.create_task(self.logging_obj.async_success_handler(result=logging_response, ...))
executor.submit(self.logging_obj.success_handler, result=logging_response, ...)

The asyncio task and the executor thread then both run the full logging pipeline (helper mutations, cost calculation, hidden-params merge, redaction, standard logging payload) against one pydantic model and one shared model_call_details dict. Concurrent mutation of a pydantic-core model from two threads is unsound in the Rust core and eventually segfaults with no Python traceback. Chat completions streaming had this exact pair until v1.88.0 routed it through dispatch_success_handlers (#29089); the responses API, interactions API, and A2A iterators kept the old pattern. Note the executor-side submit does no useful CustomLogger work on async calls anyway, since success_handler gates log_success_event on _is_sync_litellm_request

Live proxy repro on this branch's base, run with a user callback module identical to the customer's no-op handler, except it also logs which executor worker threads exist at the moment the async success hook runs:

$ cat custom_callbacks.py
class NoopHandler(CustomLogger):
    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        pool_threads = sorted(t.name for t in threading.enumerate() if t.name.startswith("ThreadPoolExecutor"))
        logger.info("NOOP_CALLBACK success. executor_threads=%s", pool_threads)

$ python litellm/proxy/proxy_cli.py --config config.yaml --port 14210
$ curl -s http://127.0.0.1:14210/v1/responses \
    -H "Authorization: Bearer sk-lit4210" -H "Content-Type: application/json" \
    -d '{"model":"gpt-5.4-nano","input":"Say hello in one word","stream":true}'
data: {"type":"response.completed","response":{"id":"resp_ATrXFkqJ...","model":"gpt-5.4-nano-2026-03-17","usage":{"input_tokens":11,"output_tokens":5,"total_tokens":16},...}}
data: [DONE]

Before (base): the litellm logging executor spawns a worker to run the sync success_handler concurrently with the async hook that is mid-flight on the event loop

INFO:noop_handler:NOOP_CALLBACK success. executor_threads=['ThreadPoolExecutor-0_0', 'ThreadPoolExecutor-4_0']

After (this branch, same command repeated 3 times): no worker is ever submitted for the sync handler; the only pool thread is the unrelated one created during proxy startup

req1: 200
req2: 200
req3: 200
INFO:noop_handler:NOOP_CALLBACK success. executor_threads=['ThreadPoolExecutor-4_0']
INFO:noop_handler:NOOP_CALLBACK success. executor_threads=['ThreadPoolExecutor-4_0']
INFO:noop_handler:NOOP_CALLBACK success. executor_threads=['ThreadPoolExecutor-4_0']

The customer's production faulthandler dumps are the evidence for the segfault itself; a 10 minute local stress run (300k streamed responses through litellm.aresponses with the no-op logger) exercised the race continuously but did not reproduce the crash on this machine, which is expected for a pydantic-core memory-safety bug that needs their exact interpreter and payload profile to trigger. The fix removes the cross-thread concurrency outright, so the unsound interleaving can no longer occur

A screen recording of the same before and after reproduction is included below. The run exercises the native responses iterator through an outer proxy configured as openai/gpt-5.4-nano whose upstream is a second local proxy, with the CustomLogger-only config attached, so it drives the exact end-of-stream path this PR changes. The recording shows the proxy starting, the streaming /v1/responses output, the before and after NOOP_CALLBACK executor thread lines, and the four regression test files passing with 105 tests

Before, on the base branch, the litellm logging executor ThreadPoolExecutor-0 has a live worker running the sync success_handler during the in-flight async hook

After, on this branch head, the logging executor has no live worker across three streamed requests and the regression suite passes

Type

🐛 Bug Fix

Changes

The async completion paths of litellm/responses/streaming_iterator.py (including the responses websocket bridge), litellm/interactions/streaming_iterator.py, litellm/a2a_protocol/streaming_iterator.py, and litellm/litellm_core_utils/realtime_streaming.py now route end-of-stream logging through the existing dispatch_success_handlers(prefer_async_handlers=True) helper instead of firing async_success_handler and executor.submit(success_handler) side by side. The helper awaits the async handler first and then submits the sync handler only when _should_run_sync_callbacks_for_async_calls() finds a genuine sync-only callback such as langfuse or s3, so a CustomLogger-only config never touches the thread pool at all. Sync SDK paths keep their existing behavior. The realtime bridge's old submit also had a latent bug where success_handler ran inline and its return value was submitted to the pool; routing through the dispatch helper removes that too

Regression tests in tests/test_litellm/responses/test_responses_streaming_iterator.py, tests/test_litellm/interactions/test_interactions_streaming_iterator.py, and tests/test_litellm/a2a_protocol/test_a2a_streaming_iterator.py drive each iterator's completion handler with a recording executor and assert that a CustomLogger-only config never submits the sync handler to the pool, and that when a sync callback is configured the submit happens only after the async handler has fully completed. All of them fail on the base branch and pass with this change

Link to Devin session: https://app.devin.ai/sessions/e5a77af3ab5c4e498f81b9e9f84db82e

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[yassin-berriai]

@greptileai

[greptile-apps]

Greptile Summary

This PR eliminates a thread-safety hazard in the streaming iterators for the responses API, interactions API, A2A protocol, and realtime bridge. Previously each of these async paths fired both asyncio.create_task(async_success_handler(...)) and executor.submit(success_handler, ...) unconditionally, causing concurrent mutation of a single pydantic model from the asyncio event loop thread and a thread-pool worker — an unsound pattern in pydantic-core's Rust core that produced exit 139 segfaults on customer pods.

• Core fix — the async path in all four iterators now delegates to dispatch_success_handlers(prefer_async_handlers=True), which awaits the async handler first and only submits the sync handler via the executor when _should_run_sync_callbacks_for_async_calls() finds a genuine sync-only callback; a CustomLogger-only config never touches the thread pool.
• Latent bug also fixed — realtime_streaming.py had executor.submit(self.logging_obj.success_handler(self.messages)), calling the handler inline and submitting its return value to the pool instead of the callable; the new code routes through dispatch_success_handlers and removes the module-level executor.
• Regression tests — three new test files use a RecordingExecutor to assert no concurrent executor submission for CustomLogger-only configs, and that when a sync callback is present the executor submission happens only after the async handler has fully completed; all would fail on the base branch.

Confidence Score: 5/5

Safe to merge — all async streaming paths now serialize the async and sync logging handlers, removing a confirmed concurrent-mutation hazard against a shared pydantic model.

The fix is narrowly scoped to end-of-stream logging dispatch in four async iterators. Each changed site replaces a two-statement fire-and-forget pattern with a single await-then-conditionally-submit pattern that was already proven safe in the chat-completions streaming path. New regression tests use a recording executor to confirm the old concurrent submission cannot reoccur. Sync SDK paths are untouched. No backward-incompatible API surface changes are introduced.

No files require special attention.

Important Files Changed

Filename	Overview
litellm/responses/streaming_iterator.py	Async path now routes through dispatch_success_handlers(prefer_async_handlers=True) instead of concurrent create_task + executor.submit; sync path keeps executor.submit for backward compatibility. Correct fix.
litellm/interactions/streaming_iterator.py	Async path uses dispatch_success_handlers; sync path preserves run_async_function + executor.submit (sequential, not concurrent). executor import is still needed for the SyncInteractionsAPIStreamingIterator at line 251.
litellm/litellm_core_utils/realtime_streaming.py	Removes module-level thread pool executor and fixes a latent bug where success_handler was called inline (not submitted as a callable) before routing through dispatch_success_handlers.
litellm/a2a_protocol/streaming_iterator.py	Removes concurrent executor.submit + async_success_handler pattern; routes through dispatch_success_handlers with prefer_async_handlers=True. Import removed cleanly.
tests/test_litellm/responses/test_responses_streaming_iterator.py	New regression test using RecordingExecutor to assert no concurrent executor submission for CustomLogger-only config, and that sync callbacks are only submitted after async handler completes.
tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py	Assertion updated from assert_called_once() to assert_not_called() on the mock executor, correctly reflecting the fixed behavior where async path no longer submits the sync handler concurrently.

_{Reviews (3): Last reviewed commit: "fix(realtime): route realtime success lo..." | Re-trigger Greptile}

[greptile-apps]

Greptile Summary

This PR fixes a race condition that caused pydantic-core segfaults (exit 139) in pods using custom loggers: streaming iterators for the responses API, interactions API, and A2A protocol were firing both async_success_handler and executor.submit(success_handler) concurrently against the same response object. The fix routes end-of-stream logging through the existing dispatch_success_handlers helper (with prefer_async_handlers=True), which awaits the async handler first and only submits to the thread pool when a genuinely sync-only callback (e.g., langfuse, s3) is configured.

• litellm/responses/streaming_iterator.py, litellm/interactions/streaming_iterator.py, and litellm/a2a_protocol/streaming_iterator.py async paths now use dispatch_success_handlers(prefer_async_handlers=True) instead of racing async_success_handler + executor.submit(success_handler).
• Three new regression test files verify that a CustomLogger-only config never submits to the thread pool, and that sync callbacks are submitted only after the async handler completes.
• The sync iteration path in BaseResponsesAPIStreamingIterator is preserved unchanged.

Confidence Score: 4/5

Safe to merge; the core fix is mechanically correct and the three new test files all fail on base and pass on this branch.

The change touches the end-of-stream logging path in three separate iterators. All three async paths now go through dispatch_success_handlers, which serializes the async handler before conditionally submitting the sync handler. The sync path in BaseResponsesAPIStreamingIterator is left intact. The only minor gap is that the A2A regression test passes request=object() as a dummy and relies on A2ARequestUtils.get_input_message_from_request returning None gracefully rather than raising; if that helper ever tightens its types, the test would silently fail instead of catching a real regression.

The A2A test (test_a2a_streaming_iterator.py) uses a bare object() as the request sentinel — worth confirming it fires the task on the current and future versions of the helper.

Important Files Changed

Filename	Overview
litellm/responses/streaming_iterator.py	Async path now routes through dispatch_success_handlers(prefer_async_handlers=True); sync path preserved with run_async_function + executor.submit. WebSocket path updated correctly. executor import retained for sync path usage.
litellm/interactions/streaming_iterator.py	Async iterator _handle_logging_completed_response now uses dispatch_success_handlers; sync iterator at line 251 still uses executor.submit directly (correct). The executor import at line 21 is still needed for the sync path.
litellm/a2a_protocol/streaming_iterator.py	Removed executor import and the executor.submit(success_handler) call; _handle_stream_complete now uses dispatch_success_handlers(prefer_async_handlers=True) wrapped in asyncio.create_task.
tests/test_litellm/responses/test_responses_streaming_iterator.py	New regression test: verifies CustomLogger-only config never submits to the thread pool, and that sync callbacks are submitted only after async handler finishes (timing assertion with monotonic clock).
tests/test_litellm/interactions/test_interactions_streaming_iterator.py	New regression test for interactions API async path; correctly patches both thread_pool_executor_module.executor and interactions_streaming_iterator_module.executor.
tests/test_litellm/a2a_protocol/test_a2a_streaming_iterator.py	New regression test; uses raising=False in monkeypatch since executor is no longer imported by the a2a module. The test relies on A2ARequestUtils.get_input_message_from_request(object()) returning None gracefully, which it does.
tests/test_litellm/responses/test_responses_websocket_all_providers.py	Six mock-based tests updated to mock dispatch_success_handlers instead of async_success_handler; correctly reflects the renamed call path in ResponsesWebSocketStreaming._log_messages.

_{Reviews (2): Last reviewed commit: "fix(responses): stop scheduling sync suc..." | Re-trigger Greptile}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[yassin-berriai]

@greptileai