Repository navigation
fix(voice): keep overlapping realtime chat context pushes from deleting each other's items - #7904
Rehansanjay wants to merge 2 commits into
Conversation
…ng each other's items `update_chat_ctx` sends the difference between the context it is handed and the provider's mirror of the server conversation, and that mirror only catches up when the server acks the items. Three places took their snapshot of the mirror outside that wait: the text user turn, the interrupted tool outputs, and the new tool outputs. A snapshot taken while another push waits for its ack is missing that push's items, so they come back out of the diff as deletes. A text user turn queued during a tool output push therefore deleted the `function_call_output` from the server conversation while the agent's own chat context kept it: the function call stayed open with no output, and the tool reply and every later response were generated without the result, with the model reporting the operation as still running. The window is the ack round trip, so it closes on a fake server with an instant ack and stays open with real latency. Move the snapshot and the push into one helper that holds a per-activity lock across both, so a push diffs a mirror nobody else is about to change. This keeps it out of the RealtimeSession ABC and covers every provider whose `update_chat_ctx` diffs against a server mirror. Fixes livekit#7880
| async with self._rt_chat_ctx_push_lock: | ||
| chat_ctx = self._rt_session.chat_ctx.copy() | ||
| chat_ctx.items.extend(items) | ||
| await self._rt_session.update_chat_ctx(chat_ctx) |
There was a problem hiding this comment.
🔴 Concurrent context replacement deletes user turns
When Agent.update_chat_ctx overlaps a text push, _rt_chat_ctx_push_lock cannot protect the newly acknowledged user item. The replacement update bypasses that lock and can delete the item before the reply.
Learn more
The realtime provider diffs each requested context against its acknowledged server mirror. A text turn adds its user item to the agent context only after its push returns in _realtime_reply_task. Meanwhile, update_chat_ctx submits a replacement context directly without taking the new lock. If that replacement excludes the still-pending user item, its provider update runs after the text push and deletes the item. The interrupted-generation cleanup also updates the provider directly at _realtime_generation_task.
Example: A text user turn pushes “hello” and waits for server acknowledgment. Application code calls agent.update_chat_ctx(agent.chat_ctx.copy()) before “hello” enters the local context. After the first push is acknowledged, the replacement diff deletes “hello” from the realtime conversation, and the model answers without that user turn.
Recommended fix: Serialize all activity-owned calls to RealtimeSession.update_chat_ctx against the same activity lock, including replacement and cleanup calls. Determine replacement semantics explicitly: a caller-supplied context that predates an in-flight item must either retain that item or intentionally remove it. Add an interleaving test with an acknowledged push followed by a stale replacement.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
The race is real, and I had it wrong in a comment on #7905 — thanks for the push. But the recommended fix does not close it, so I would rather not apply it here.
The interleaving, with the lock added to update_chat_ctx:
- the text turn takes the activity lock, snapshots the mirror, pushes
hello, waits for the ack - application code calls
update_chat_ctx(agent.chat_ctx.copy()). That copy has nohelloin it, because the text turn only adds it to the agent context after its push returns.update_chat_ctxassignsself._agent._chat_ctx = chat_ctxstraight away and then blocks on the lock - the text turn's ack lands, it releases the lock and upserts
hellolocally - the replacement acquires the lock and diffs its own copy, which still has no
hello, against a mirror that now does — and deletes it
Step 4 is unchanged by the lock: what is stale is the caller's context object, and no amount of serialization refreshes it. That is why I left those two sites alone — not because the race does not exist, which is what I claimed on #7905 and was wrong about.
What does close it is ordering, and it is one line. The two tool-output paths already insert into self._agent._chat_ctx before pushing (agent_activity.py:4759 and :4807-4808), so a context copied mid-push contains their items and no diff can delete them. The text turn is the only push that does it the other way round, at :4198. Hoisting that upsert above the push makes it consistent with the other two.
It is not free, which is why it is not in this PR: _conversation_item_added fires from the same place, so hoisting changes when that event is emitted, and on the unexpected-exception path (speech_handle._mark_done(error=e); return) the message would now stay in the local context for a turn that failed. Both are answerable, but they are a different question from the one this PR fixes — overlapping pushes diffing stale mirror snapshots — and #7880 does not cover them.
I will file it separately with this interleaving so it does not get lost. Happy to fold it in here instead if a maintainer would rather have one change.
…ssion The plugin diffs, sends and waits for the ack while holding _update_chat_ctx_lock, so an overlapping push diffs the mirror the earlier push left behind rather than the one it saw when it built its context. Giving the fake that lock reproduces the delete through the same mechanism as the real one instead of through a diff deferred to the ack.
|
#7905 landed a minute after this one with the same fix — same lock, same three snapshot sites. Flagging it so this does not get reviewed twice; either is fine by me. Also pushed a follow-up to the test fake: it now holds a lock across the diff, the send and the ack the way the plugin's |
|
Closing in favour of #7905, which is the better version of this change. Epsilon006 found the ordering I had reasoned my way past. I only considered append-then-replace, where the replacement captures its context before blocking and still diffs a stale object afterwards — which a lock cannot fix, and which is why I left two call sites alone. The case that matters is the reverse: the replacement goes first, holds the provider lock, and the later append snapshots a mirror that does not yet contain its item. The activity lock does fix that, because the snapshot cannot be taken until the earlier push is acked. Same for never-played cleanup versus a late tool output recreating a deleted item. Locking all five sites is the right invariant. #7905 has it, with repros through a full AgentSession. Nothing here that isn't covered there. |
Fixes #7880.
The bug
RealtimeSession.update_chat_ctxsends the difference between the context it is handed and the provider's mirror of the server conversation, and that mirror only catches up when the server acks the items. Three call sites took their snapshot of the mirror outside that wait:_realtime_reply_taskgenerate_reply(user_input=...)/session.run(user_input=...)_realtime_generation_task_realtime_generation_taskA snapshot taken while another push is waiting for its ack has none of that push's items, so they come back out of the diff as deletes. A text user turn queued during a tool-output push therefore sent
conversation.item.deletefor thefunction_call_outputthe tool push had just created, while the agent's ownchat_ctxkept it. The function call stayed open with no output, and the tool reply and every later response were generated without the result — the model typically reports the operation as still running.The window is the ack round trip: it closes against a fake server that acks instantly, and stays open with real latency. As the issue notes, any two overlapping pushes can drop each other's items the same way, not just this pair.
The fix
One helper,
_extend_rt_chat_ctx, that holds a per-activity lock across the snapshot and the push, so a push diffs a mirror nobody else is about to change. The three sites call it instead of copying the mirror themselves; their error handling is unchanged, and it still raisesllm.RealtimeErrorthe wayupdate_chat_ctxdoes, so the best-effort paths behave as before.Fixing it in the activity rather than in the plugin was deliberate: the stale snapshot is created by the caller, all three callers are here, and this way it covers every provider whose
update_chat_ctxdiffs against a server mirror without adding anything to theRealtimeSessionABC.The user-turn site now builds its
ChatMessagedirectly instead of viachat_ctx.add_message(...)on the snapshot. With nocreated_atpassed,add_messageappends, so the resulting item and its position are the same.Test
tests/test_realtime_reply_chat_ctx.py::test_overlapping_pushes_keep_each_other_s_itemsdrives two overlapping pushes through a fake session with the plugin's semantics:chat_ctxanswers from a mirror, andupdate_chat_ctxonly applies its diff once an explicit ack event is set. The ack is a gate rather than a sleep, so the interleaving is deterministic instead of timing-dependent.Verified by mutation — dropping the lock from the helper and keeping everything else:
That item is the
function_call_output.What this doesn't cover
A push whose ack times out raises
RealtimeErrorand releases the lock, while its items may still be acked afterwards — so a following push can still diff a mirror that is about to change. That is the existing best-effort behaviour at these call sites (the items were sent; only the ack timed out), and narrowing it looked like a separate change.