Streaming: correctness & resource-safety bugs in the token/delta pipeline

Summary

An audit of the token/event streaming pipeline (llm.py → on_token/on_event → PubSub → WebSocket) surfaced several correctness and resource-safety bugs. The durable record is correct — the final MessageEvent is rebuilt server-side via litellm.stream_chunk_builder, so persisted state never ends up wrong. All the issues below are in the live/transient streaming layer. Line numbers reference the current main.

The most impactful is OpenHands/agent-canvas#1 (a plain, unconditional code path — not timing-dependent).

---

1. HIGH — Retry re-fires on_token from scratch, duplicating every streamed token on the client

Tokens are emitted inside the retry boundary, so any retry re-streams and re-emits all tokens.

• openhands-sdk/openhands/sdk/llm/llm.py:2117-2120 — on_token(chunk) is called inside the stream loop of _transport_call.
• openhands-sdk/openhands/sdk/llm/llm.py:1429-1439 — _transport_call is invoked from _one_attempt, which is wrapped by @self._make_retry_decorator().
• openhands-sdk/openhands/sdk/llm/llm.py:1351-1354 — _validate_chat_response raises LLMNoResponseError after the full stream has already fired.
• openhands-sdk/openhands/sdk/llm/llm.py:131-138 — LLMNoResponseError (plus APIConnectionError, RateLimitError, LiteLLMTimeout, etc.) is in LLM_RETRY_EXCEPTIONS.

Failure: model streams "The answer is 4"; provider drops the connection (retryable) or _validate_chat_response raises LLMNoResponseError. Tenacity re-runs _one_attempt, re-opens the stream, and calls on_token for every chunk again → the client renders the text 2× (N× for N retries). Same defect in responses()/aresponses().

Fix idea: emit through a per-attempt buffering wrapper that only forwards new deltas, move on_token outside the retry boundary, or send the callback a "reset/restart" signal on each new attempt.

---

2. MEDIUM — Empty stream reassembles to None → un-retried AssertionError, diverging from the non-streaming path

• openhands-sdk/openhands/sdk/llm/llm.py:2120-2124 — ret = litellm.stream_chunk_builder(chunks, ...) then assert isinstance(ret, ModelResponse). For an empty stream stream_chunk_builder([]) returns None, so the assert raises AssertionError, which is not in LLM_RETRY_EXCEPTIONS and surfaces raw. Same at :2165-2169 (async).
• Contrast: the non-streaming path treats an empty response as a retryable LLMNoResponseError (:1351-1354).

Failure: provider returns HTTP 200 with an empty SSE body (content filter, proxy hiccup) → confusing un-retried AssertionError on the streaming path where the non-streaming path would retry.

Fix idea: if ret is None: raise LLMNoResponseError(...) before the assert.

---

3. MEDIUM — No backpressure on the delta path; the delivery future is discarded

• openhands-agent-server/openhands/agent_server/event_service.py:804-805 — _publish_stream_delta fire-and-forgets: asyncio.run_coroutine_threadsafe(self._pub_sub(event), self._main_loop) with the future discarded and never awaited. The LLM worker thread never blocks on delivery.
• openhands-agent-server/openhands/agent_server/sockets.py:479-481 — the underlying send has no timeout: await websocket.send_json(dumped).

Failure: one slow/stuck-but-connected WebSocket (full TCP send buffer) during a multi-thousand-token response → every token schedules a pub_sub(delta) task that suspends in send_json and never completes; pending tasks + StreamingDeltaEvent objects accumulate unbounded on the shared loop, affecting all conversations.

Fix idea: bound deltas per subscriber (bounded asyncio.Queue that drops/coalesces oldest when full) instead of fire-and-forget scheduling.

---

4. MEDIUM — No send-timeout on _publish_state_update (pause / interrupt / finalize); one wedged client can hang the control plane

• openhands-agent-server/openhands/agent_server/event_service.py:1573 — await self._pub_sub(state_update_event) inline, no timeout.
• openhands-agent-server/openhands/agent_server/pub_sub.py:98 — fans out with await asyncio.gather(...), which never returns while any one subscriber's notify never completes.
• openhands-agent-server/openhands/agent_server/sockets.py:479-481 — send_json with no timeout is where a wedged client stalls.
• Contrast: the initial state push is bounded — openhands-agent-server/openhands/agent_server/event_service.py:590-594 wraps it in asyncio.wait_for(..., timeout=INITIAL_STATE_PUSH_TIMEOUT_SECONDS). Subsequent publishes are not.

Failure: a client whose receive window is full and never drains makes gather wait forever; every _publish_state_update (run finalization at :997, pause/interrupt) blocks indefinitely, so user pause/interrupt appears frozen even though other clients already got the update.

Fix idea: wrap each subscriber notify in asyncio.wait_for(...) like the initial push.

Nit (same area): the comment at event_service.py:1570-1572 says pub_sub "iterates through subscribers sequentially" — that is stale; pub_sub.py:98 fans out concurrently via asyncio.gather. The bug above stands regardless.

---

5. MEDIUM — The wait_for_pending ordering barrier ignores delta futures

• openhands-sdk/openhands/sdk/utils/async_utils.py:45-55 — only AsyncCallbackWrapper.__call__ records futures into _pending_futures.
• openhands-agent-server/openhands/agent_server/event_service.py:790-805 — deltas bypass the wrapper and go straight to _pub_sub, so their futures are never tracked.
• openhands-agent-server/openhands/agent_server/event_service.py:990-993 — the run's finally calls _callback_wrapper.wait_for_pending(30.0) before publishing the final state update, but that provably does not wait for trailing deltas.

Failure (timing-dependent): a trailing StreamingDeltaEvent can be delivered to a WS client after the superseding MessageEvent/FINISHED, producing duplicated/garbled streamed text. Usually masked by incidental FIFO task scheduling, so the observable reorder is plausible rather than guaranteed — but the barrier gap is real.

Fix idea: track delta futures and include them in wait_for_pending, or route deltas through the same ordered per-subscriber queue as regular events (ties into OpenHands/agent-canvas#3).

---

6. LOW / fragility — resend_mode=since relies on naive local timestamps that aren't monotonic

• openhands-sdk/openhands/sdk/event/base.py:28-29 — timestamp: str = Field(default_factory=lambda: datetime.now().isoformat()) — naive local wall-clock, no tz.
• The since filter (event_service.py:217-219) and TIMESTAMP_DESC search assume these strings are monotonically non-decreasing in append order.

Failure (needs clock step): an NTP backward step makes a newly appended event's timestamp less than the reconnect anchor → it's excluded from the >= resend window and never replayed. Format is confirmed; the dropped-event outcome requires an actual skew event.

Fix idea: anchor/paginate the since boundary on a monotonic cursor (append index / ULID) rather than wall-clock.

---

What was verified as robust (not bugs)

• PubSub fault isolation — pub_sub.py:89-98: each subscriber runs in its own try/except inside _notify, then gathered; a raising subscriber doesn't break the broadcast (a hanging one is OpenHands/agent-canvas#4).
• Unsubscribe on disconnect — sockets.py:469-470 removes the subscriber in a finally.
• Late-joiner correctness — deltas are purely additive UX; the final MessageEvent is rebuilt from the full response via litellm.stream_chunk_builder, so a client that misses deltas still gets the complete message from history.

---

Related

Client-side counterparts (mid-stream disconnect leaves a permanently holey/duplicated message because deltas aren't persisted and can't be replayed) are filed in OpenHands/agent-canvas#1979. The root enabler is that StreamingDeltaEvent carries no completion/message id, forcing the client to reconcile by positional adjacency — adding an id here would let the client key on it instead.

Start by tracing the streaming entry points in openhands-sdk/openhands/sdk/llm/llm.py and the publish flow through event_service.py, pub_sub.py, sockets.py, and async_utils.py. Review base.py and the resend filter for timestamp behavior. Done means the listed retry, empty-stream, backpressure, timeout, ordering, and resend cases are handled without duplicating live output or blocking control updates.