fix: stop dropping session host responses - #6781
Conversation
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
A request whose response never reaches the client is indistinguishable from a hung agent: the caller blocks until its timeout with a complete transcript and a healthy agent log. Three paths dropped one silently. `RoomSessionTransport.send_message` swallowed send failures into a warning and returned early when the room was gone, so a handler believed it had replied. Both now raise — only the caller knows whether anyone is waiting on that message. Events keep their fire-and-forget semantics through a wrapper that logs instead of raising. `SessionHost.aclose` cancelled in-flight handlers outright, discarding the response of a request that had already done its work. It now gives them a grace period to finish before cancelling. Both transports also serialize their sends. The room transport opens a stream per message on a shared topic, and the TCP transport can yield between writing a frame header and its payload, so concurrent senders could interleave. Addresses #6661. Co-authored-by: samanyugoyal2010 <202565741+samanyugoyal2010@users.noreply.github.com>
`_send_event` spawned a task per event, which under the new send lock means N tasks queueing on it rather than one writer working through them. Events are emitted from sync callbacks and nobody awaits them, so they queue on a channel that a single long-lived task drains. This also gives them a stricter order than before: emission order rather than task scheduling order. Responses stay on the direct path — a handler awaits the transport and gets the failure, so no per-message future is needed to hand the result back. `aclose` closes the channel so the writer flushes what the handlers queued on their way out, sharing one drain budget with them. Co-authored-by: samanyugoyal2010 <202565741+samanyugoyal2010@users.noreply.github.com>
4837df8 to
fff594f
Compare
Every inbound message spawned a task to read its byte stream, and every outbound event spawned one to send it. Task creation is not free, and these are the two paths that scale with traffic rather than with turns. Both now queue to a single long-lived drainer. Reading sequentially also fixes an ordering hazard: two per-stream tasks could finish out of order and deliver their messages swapped. Request handlers keep a task each — they have to run concurrently with the receive loop, or a long `run_input` would stall audio and every request behind it. Those are client-driven and serialized in practice, so it is roughly one per turn rather than one per message. `aclose` names them after their request so abandoning one at shutdown says which caller is left waiting, and both drains share one helper and one deadline. Co-authored-by: samanyugoyal2010 <202565741+samanyugoyal2010@users.noreply.github.com>
The read and write drainers are long-lived tasks: an exception that escapes either one ends the loop silently and the session stops delivering in that direction. Route both through `utils.log_exceptions` like the rest of the voice package, and name them as a `_read_loop` / `_write_loop` pair. `CancelledError` derives from BaseException, so shutdown still passes through unlogged. Co-authored-by: samanyugoyal2010 <202565741+samanyugoyal2010@users.noreply.github.com>
b4bab98 to
94ff50f
Compare
94ff50f to
63ce142
Compare
| # closing the channel lets the writer flush what the handlers queued on | ||
| # their way out, then exit | ||
| self._event_ch.close() | ||
| if self._write_task is not None: | ||
| await _drain({self._write_task}, deadline) | ||
| self._write_task = None |
There was a problem hiding this comment.
🟡 Queued session updates can be thrown away during shutdown when a long request is still running
The queued outgoing updates are given whatever is left of a single three-second shutdown budget already spent on waiting for in-flight requests (_drain({self._write_task}, deadline) at livekit-agents/livekit/agents/voice/remote_session.py:512), so when that budget is exhausted they are silently discarded instead of being delivered.
Impact: On shutdown of a busy session, the last status updates never reach the client and nothing in the logs says they were dropped.
Shared deadline leaves the writer zero time to flush
aclose computes one deadline = now + _SHUTDOWN_DRAIN_TIMEOUT (livekit-agents/livekit/agents/voice/remote_session.py:493) and first uses it to wait for in-flight request handlers (:498). If those handlers take the full 3 s (e.g. a slow run_input), the remaining budget for the event writer is 0, so _drain computes remaining = 0, asyncio.wait(..., timeout=0) returns the writer as pending, and it is cancelled immediately (:79-83). Everything still queued on _event_ch — including events the handlers emitted on their way out, which the comment at :508-509 says the close is meant to flush — is dropped. Unlike the handler drain, the writer cancellation is not logged, so the loss is invisible.
A small dedicated budget for the writer (or at least a warning when it is cancelled with items still queued) would keep the stated behaviour.
Was this helpful? React with 👍 or 👎 to provide feedback.
63ce142 to
157363f
Compare
| # closing the channel lets the writer flush what the handlers queued on | ||
| # their way out, then exit | ||
| self._event_ch.close() | ||
| if self._write_task is not None: | ||
| await _drain({self._write_task}, deadline) | ||
| self._write_task = None |
There was a problem hiding this comment.
🟡 Restarting a closed session host silently stops all status updates from reaching the client
The queue that status updates flow through is shut for good when the session host closes (self._event_ch.close() at livekit-agents/livekit/agents/voice/remote_session.py:511), and nothing re-creates it if the host is started again, so every update emitted after a restart is discarded without a trace.
Impact: A host that is closed and started again keeps serving requests but never reports state changes, transcripts or usage to the remote client.
Why a restart leaves the writer dead
aclose resets self._started = False (livekit-agents/livekit/agents/voice/remote_session.py:471), so start() (livekit-agents/livekit/agents/voice/remote_session.py:458-466) can legitimately run again — it re-creates _write_task, but _event_ch was built once in __init__ (livekit-agents/livekit/agents/voice/remote_session.py:439) and is permanently closed by aclose. The new _write_loop therefore sees StopAsyncIteration immediately and exits, while _send_event swallows the ChanClosed raised by send_nowait (livekit-agents/livekit/agents/voice/remote_session.py:550-551), so every subsequent event disappears silently. Before this PR each event was dispatched as its own task, so a restarted host still sent events. Re-creating the channel in start() (or on close) restores the previous contract.
Prompt for agents
SessionHost supports being closed and started again (aclose resets _started to False and unregisters the session handlers, start() re-creates the receive/write tasks). However _event_ch is created once in __init__ and closed permanently in aclose. After a restart, _write_loop exits immediately on the closed channel and _send_event swallows ChanClosed, so all events are dropped silently. Consider re-creating the event channel at the start of start() (or resetting it at the end of aclose) so a restarted host has a fresh, open queue, while keeping the current behaviour that events emitted between register_session() and start() are still queued.
Was this helpful? React with 👍 or 👎 to provide feedback.
|
looks good to merge @theomonnom feel free to do so thanks for including me! |
The event channel was built in start(), so an event emitted between register_session() and start() had nowhere to go and was dropped. Build it in the constructor instead: events queue from the moment the session is registered and flush once the writer starts. It is also no longer optional, which drops the None checks around every use. Co-authored-by: samanyugoyal2010 <202565741+samanyugoyal2010@users.noreply.github.com>
157363f to
f96cbb1
Compare
|
@samanyugoyal2010 Thanks for reporting the issue! |
| @utils.log_exceptions(logger=logger) | ||
| async def _read_loop(self) -> None: | ||
| # sequential, so messages reach _recv_ch in the order they were sent | ||
| async for reader in self._incoming_ch: | ||
| await self._read_stream(reader) |
There was a problem hiding this comment.
🟡 One stalled incoming message can silently block every later message on the session channel
Incoming session messages are now read one at a time by a single reader (await self._read_stream(reader) at livekit-agents/livekit/agents/voice/remote_session.py:139) with no time limit on any one of them, so a single message whose delivery stalls holds up every message queued behind it for as long as it stalls.
Impact: If one message from the remote peer never finishes arriving (e.g. the peer disappears mid-transfer), the agent stops seeing all further requests and events on that channel for the rest of the session, which looks exactly like the hung agent this change is meant to eliminate.
Head-of-line blocking introduced by the single sequential reader loop
Previously _on_byte_stream spawned a task per stream (asyncio.create_task(self._read_stream(reader))), so a stream that never terminated only stranded its own task. Now _on_byte_stream (livekit-agents/livekit/agents/voice/remote_session.py:129-133) queues readers into _incoming_ch and _read_loop drains them strictly sequentially. _read_stream (livekit-agents/livekit/agents/voice/remote_session.py:141-153) iterates the rtc.ByteStreamReader to completion with no timeout and only handles exceptions, not indefinite waiting. All subsequent readers, including responses/requests that were already fully received, sit in _incoming_ch untouched. The PR notes the concurrency/ordering trade-off but does not bound the per-stream wait; a per-read timeout (after which the stream is abandoned with a warning) or a bounded reader pool would preserve ordering guarantees for well-behaved peers while keeping a stalled stream from wedging the channel.
Prompt for agents
In RoomSessionTransport._read_loop (livekit-agents/livekit/agents/voice/remote_session.py), inbound byte streams are now read strictly one at a time with no upper bound on how long a single stream may take. _read_stream iterates the rtc.ByteStreamReader until exhaustion and only guards against exceptions, so a peer that opens a stream and stops sending (crash, network partition, half-open stream) blocks the loop forever and every later inbound message — requests, responses, events — is stuck in _incoming_ch and never delivered to _recv_ch. The previous per-stream task design isolated that failure. Consider bounding each _read_stream call with a timeout (abandoning that stream with a warning on expiry) so ordering is preserved for healthy peers without letting one stalled stream wedge the whole session channel; an alternative is a small bounded reader pool with explicit resequencing.
Was this helpful? React with 👍 or 👎 to provide feedback.
3 participants
Alternative to #6665, addressing #6661. Diagnosis is @samanyugoyal2010's — commits are co-authored to them, happy to close in favour of theirs.
What was happening
A
run_inputresponse that never reached the client looked exactly like a hung agent: the caller blocked for its full 60s timeout while the transcript was complete and the agent log showed a clean shutdown.Three paths dropped one silently:
RoomSessionTransport.send_messageswallowed send failures into a warning and returned.SessionHost.aclosecancelled in-flight handlers, discarding the response of a request that had already done its work.Nothing serialized writes either, and every inbound message spawned a task to read it — which could finish out of order and deliver messages swapped.
The fix
aclosedrains handlers before cancelling, and logs the request types it gave up on.Responses stay direct — a handler awaits the transport and gets the failure — so no per-message future is needed to hand results back. Events are the only thing that needs a queue.
14 tests, each verified to fail without its change.
Caveats