[Bug] _async_gen_to_sync_gen deadlocks container when SSE consumer disconnects (queue blocks worker loop thread)

[kedazhu1-ai]

In _async_gen_to_sync_gen:

q: queue.Queue = queue.Queue(maxsize=100)

async def _produce() -> None:
async for chunk in async_gen:
q.put((True, chunk)) # ← BLOCKING put, no timeout
q.put((True, _DONE))

Sync consumer

while True:
ok, value = q.get() # ← stops iterating when SSE disconnects
yield value

Here's the deadlock chain:
SSE client disconnects → Starlette stops iterating the sync generator
_produce() is still running on the worker loop thread, producing chunks
Queue fills up to 100 items → q.put() blocks the worker loop thread
Worker loop thread is now frozen → container can't process ANY new requests
Container stays deadlocked until idle timeout

This is not something you can work around on your side — q.put() is a blocking call with no timeout or cancellation mechanism. The fix needs to be in the SDK.

Code ref: app.py L737-L750

[bthivierge-jg]

Can confirm this in production. We're running a long-running streaming agent (responses >2.5 min) with Lambda rotation — Lambda1 streams until soft timeout, disconnects, Lambda2 reconnects to the same session with a new invocation.

Observed: 78-second gap between AgentCore returning HTTP 200 to Lambda2 and our @app.entrypoint generator actually receiving the payload. During this time the microVM sends SSE heartbeat comments (:) but no data events. Then all buffered events dump at once.

Timeline from CloudWatch (run 5013d4dd):

21:35:09.691  AgentCore returns 200 OK to Lambda2
              ─── 78 seconds, worker loop blocked ───
21:36:27.763  Entrypoint receives payload (buffer has 3740 events, done=True)
**Reproduction** (no AWS needed):

import asyncio, queue, threading, time

q = queue.Queue(maxsize=5)
loop = asyncio.new_event_loop()
threading.Thread(target=loop.run_forever, daemon=True).start()

async def producer():
    for i in range(20):
        q.put(("item", i))  # blocks at item 5

async def second_task():
    print("second task ran")  # never prints while producer is blocked

asyncio.run_coroutine_threadsafe(producer(), loop)
time.sleep(0.5)
asyncio.run_coroutine_threadsafe(second_task(), loop)
time.sleep(3)
# second_task never runs — loop thread is stuck in q.put

Suggested fix: Replace queue.Queue with asyncio.Queue and await q.put(...):

async def _produce() -> None:
    async for chunk in async_gen:
        await q.put((True, chunk))  # yields to scheduler when full
    await q.put((True, _DONE))

The sync consumer side would need asyncio.run_coroutine_threadsafe(q.get(), worker_loop).result() instead of q.get().

We're monkey-patching this locally for now but would love an upstream fix. Happy to submit a PR if that's welcome.