[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.

[galuszkm]

Worth splitting this into two cases, because the impact is very different depending on where the app runs.

Deployed on AgentCore Runtime

Here you don't actually have to depend on the SSE disconnect to stop a session. The platform exposes a control-plane API, StopRuntimeSession, that terminates the session out-of-band and stops any ongoing streaming response. So even while this deadlock exists, a disconnected or runaway session can still be stopped by calling the control plane directly:

import boto3

# Stop a running session out-of-band (deployed AgentCore Runtime only).
client = boto3.client("bedrock-agentcore", region_name="us-east-1")

resp = client.stop_runtime_session(
    agentRuntimeArn="arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/my-agent-xxxxxxxx",
    runtimeSessionId="<the 33-256 char session id you invoked with>",
    # qualifier="DEFAULT",   # optional endpoint alias
    # clientToken="...",     # optional idempotency token
)
print(resp["statusCode"])

Two things to know:

• The calling principal needs bedrock-agentcore:StopRuntimeSession on the runtime ARN. This is not part of the default runtime execution role, so you have to add it to whatever identity makes the call (e.g. your backend/Lambda role).
• If the session already finished, the call raises a "session not found" error — safe to treat as "already stopped".

So on the managed runtime this is more of an efficiency/edge-case problem (e.g. the Lambda-rotation reconnect above) than a hard blocker, because you have an out-of-band kill switch.

Local dev (running the app as a plain ASGI server, e.g. uvicorn)

This is where it really hurts and there's no good workaround. There's no control plane locally, so StopRuntimeSession doesn't exist — the only way to stop a session is the client disconnecting. But that's exactly the path that deadlocks: the consumer stops draining, the queue fills to 100, q.put() blocks the single shared worker-loop thread, and from then on the whole process is wedged (every session, since there's one worker loop per process). Nothing on the caller side can unblock it because q.put() has no timeout or cancellation.

So for local development the fix really does have to be in the SDK. The asyncio.Queue + await q.put(...) change suggested above would also let the producer be cancelled cleanly when the consumer goes away, which fixes both the deadlock and the "session never stops on disconnect" behavior.