[BUG CLIENT]: Severe event loop stall and ThreadPoolExecutor thread explosion per SSE line in WorkflowEncodingHook

[andriiorap]

Python -VV

Python 3.11.16 (main, Feb 2026, 12:00:00) [Clang 16.0.0]

Pip Freeze

mistralai>=2.10.0
cryptography>=43.0.0
httpx>=0.28.1
pydantic>=2.11.2

Reproduction Steps

1. Configure client workflow encoding / payload encryption using Mistral.configure_workflow_encoding().
2. Open a streaming workflow event subscription (get_stream_events or stream_workflow_execution).
3. Intercept the stream through _wrap_sse_response_with_decryption / _DecryptingAsyncByteStream.
4. Run concurrent streaming requests or background tasks on the same asyncio event loop.

Minimal Standalone Reproduction:

import asyncio
import concurrent.futures
import time

# Recreating the exact pattern in workflow_encoding_hook.py (lines 140-151 and 236-237)
def _run_async(coro):
    try:
        asyncio.get_running_loop()
        # Already in async context - run in a separate thread with new loop
        with concurrent.futures.ThreadPoolExecutor() as pool:
            future = pool.submit(asyncio.run, coro)
            return future.result()  # <-- Blocks the main event loop thread!
    except RuntimeError:
        return asyncio.run(coro)

async def mock_decrypt_attributes():
    await asyncio.sleep(0.002)  # Simulating payload attribute decryption
    return True

def decrypt_sse_line(line: bytes):
    _run_async(mock_decrypt_attributes())
    return line

async def decrypting_stream(n_lines=15):
    for _ in range(n_lines):
        decrypt_sse_line(b"data: {...}\n")
        await asyncio.sleep(0)

# Heartbeat monitor measuring event-loop latency stalls
async def heartbeat_monitor():
    delays = []
    for _ in range(30):
        t0 = time.perf_counter()
        await asyncio.sleep(0.01)  # 10ms target
        delays.append((time.perf_counter() - t0) * 1000)
    return delays

async def main():
    monitor = asyncio.create_task(heartbeat_monitor())
    streams = [asyncio.create_task(decrypting_stream()) for _ in range(10)]
    delays = await monitor
    await asyncio.gather(*streams)
    print(f"10ms Heartbeat Delay: avg={sum(delays)/len(delays):.1f}ms, max={max(delays):.1f}ms")

asyncio.run(main())

Expected Behavior

• _DecryptingAsyncByteStream should decrypt and yield SSE frames asynchronously within the existing event loop without blocking other tasks.
• Streaming consumption should not allocate, start, join, and tear down a new ThreadPoolExecutor, a new OS pthread, and a secondary asyncio event loop on every incoming SSE text line.

Actual Behavior

• Synchronous Event Loop Freezing: In src/mistralai/client/_hooks/workflow_encoding_hook.py, _DecryptingAsyncByteStream.__aiter__ runs on the event loop, but calls _decrypt_sse_line synchronously. That calls _run_async, which calls future.result() (threading.Event.wait()), freezing the event loop thread on every single SSE line while waiting for the worker thread.
• Severe Latency Spikes: In our benchmarks, under 10 concurrent streams, an expected 10ms event-loop tick was delayed up to 185.9ms.
• Thread Exhaustion: Spawning a fresh ThreadPoolExecutor and calling asyncio.run() (which registers kernel epoll/kqueue selectors and pipe descriptors) per line rapidly hits container process limits (pids.max or ulimit -u), raising RuntimeError: can't start new thread.

Additional Context

Code References:

• src/mistralai/client/_hooks/workflow_encoding_hook.py:
  • Lines 140–151 (_run_async implementation creating ThreadPoolExecutor)
  • Lines 236–237 (_decrypt_sse_line calling _run_async on every SSE line)
  • Lines 260–271 (_DecryptingAsyncByteStream._process_chunk iterating lines and yielding decrypted output)

Because basesdk.py dispatches hooks via asyncio.to_thread (run_sync_in_thread), WorkflowEncodingHook relies on _run_async to bridge sync to async. However, using this pattern inside a streaming byte generator (_DecryptingAsyncByteStream) multiplies this overhead by the number of streaming frames.

Suggested Solutions

1. Native Async Decryption in Stream Generator:
   Make _DecryptingAsyncByteStream native async:

   async def _process_chunk_async(self, chunk: bytes):
       self._buffer += chunk
       lines = self._buffer.split(b"\n")
       self._buffer = lines[-1]
       for line in lines[:-1]:
           decrypted = await self._decrypt_sse_line_async(line, self._payload_encoder)
           yield decrypted + b"\n"

   Awaiting _decrypt_event_attributes directly eliminates _run_async, thread allocation, and event-loop stalls.

2. Persistent Worker Executor:
   If a sync-to-async bridge is strictly required, reuse a single process-wide / client-wide worker pool or asyncio.run_coroutine_threadsafe instead of constructing and destroying a ThreadPoolExecutor and event loop per line.