feat(agents): serialize the inline AgentTasks awaited from one turn - #6862
feat(agents): serialize the inline AgentTasks awaited from one turn#6862u9g wants to merge 1 commit into
Conversation
Pausing an activity is a single slot, so concurrent handoffs left every agent switch but the last overwritten and those tasks awaiting a result nothing could produce - function calls that never returned, a speech that never finished, and a session wedged until close timed out. The inline tasks of a turn's parallel tool calls now queue for the slot and take it in turn. AgentActivity._inline_task_slot owns the protocol, because its three steps are each ordered against the queue and invert into a different hang: - the interruption hold is taken before the queue and counted, since the queued tasks share one speech handle; releasing it per task let the user turns of one task's sub-conversation interrupt the speech the rest were anchored to, and an interrupted handle can no longer disallow interruptions - drain registration stays before the queue, or session close waits on a task for the slot it is still queued for - run watching moves past the queue, or a run waits for the user input that the task ahead of the watched one needs AgentTask.__await_impl keeps only the handoff itself. A nested task pauses a different activity, so it never queues behind the task it is nested in, and a queue that outlives its activity still refuses.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| try: | ||
| # before the queue: a queued task absent from the drain set makes session close | ||
| # wait on the slot it is still queued for | ||
| self._add_drain_blocked_tasks(blocked_tasks) |
There was a problem hiding this comment.
🔴 Session shutdown can hang when two tasks from one turn take turns pausing the agent
The waiting task is registered as "do not wait for me" only once before it queues (_add_drain_blocked_tasks(blocked_tasks) at livekit-agents/livekit/agents/voice/agent_activity.py:1229), and that registration is wiped when the task ahead of it hands the agent back, so a shutdown starting in that gap waits forever for the still-waiting task.
Impact: Closing the session (or any other agent switch) in that moment never finishes, wedging the conversation until an external timeout.
How the drain exclusion is lost between the first task's resume and the second task's pause
Flow for two inline AgentTasks awaited from one turn's parallel tool calls:
- Both tool tasks call
_inline_task_slot, both add themselves toAgentActivity._drain_blocked_tasksbefore awaiting_inline_task_lock. - Task 1 gets the slot, pauses the activity, and finally resumes it via
session._update_activity(old_agent, new_activity="resume")(livekit-agents/livekit/agents/voice/agent.py:1028-1030).AgentActivity._resume_scheduling_taskcallsself._drain_blocked_tasks.clear()(livekit-agents/livekit/agents/voice/agent_activity.py:1263), removing task 2's registration. - Only after that does task 1 release
_inline_task_lock. Task 2 wakes up and re-registers much later, indirectly, when its ownsession._update_activity(..., previous_activity="pause", blocked_tasks=...)reaches_pause_scheduling_task(which itself awaitsself._session._keyterm_detector.aclose()first,livekit-agents/livekit/agents/voice/agent_activity.py:1187-1191).
In that window task 2's tool task is neither in _drain_blocked_tasks nor shielded via the shared-speech-handle exclusion in the scheduling loop (livekit-agents/livekit/agents/voice/agent_activity.py:1805-1826). If AgentSession._aclose_impl (or any handoff) drains/pauses the activity there, _pause_scheduling_task waits for the parent speech task, which cannot finish until task 2's tool returns, while task 2 blocks on the activity/session locks the closer holds. _closed is only set after the drain completes, so the ToolError escape hatch at agent_activity.py:1232 never fires.
Prompt for agents
In AgentActivity._inline_task_slot (livekit-agents/livekit/agents/voice/agent_activity.py), the blocked tasks are added to _drain_blocked_tasks once, before awaiting _inline_task_lock. When an earlier queued inline AgentTask hands the activity back, AgentActivity._resume_scheduling_task clears _drain_blocked_tasks, so every task still queued for the slot loses its drain exclusion. It only regains it indirectly and several awaits later, once its own session._update_activity(previous_activity="pause") reaches _pause_scheduling_task. A drain/close beginning in that window waits for the parent speech task (which cannot finish until the queued tool returns) while the queued tool waits for the activity/session lock the closer holds, and _closed is only set after the drain, so the ToolError escape never triggers. Consider re-adding the blocked tasks to the drain set immediately after acquiring _inline_task_lock (before yielding), or making the resume path preserve registrations belonging to tasks still queued for the slot.
Was this helpful? React with 👍 or 👎 to provide feedback.
| async with self._inline_task_lock: | ||
| if self._closed: | ||
| raise ToolError( | ||
| "the activity that awaited the inline task closed while an earlier " | ||
| "one was running" | ||
| ) |
There was a problem hiding this comment.
🟡 A queued task can start a new sub-conversation for a speech that was already cut off
The check that the awaiting speech is still alive happens only before queuing (speech_handle.interrupted at livekit-agents/livekit/agents/voice/agent_activity.py:1221) and is not repeated once the queued task's turn comes, so a task can start a whole agent switch for a speech that was forcibly cut off while it waited.
Impact: After a forced interruption (e.g. session close), a second sub-conversation can still be started and then be aborted mid-switch, leaving the conversation in an inconsistent state.
Why the pre-queue check is insufficient now that tasks queue
The counted interruption hold (SpeechHandle._hold_interruptions) blocks ordinary interruptions, but interrupt(force=True) bypasses it — AgentSession._aclose_impl does exactly that (livekit-agents/livekit/agents/voice/agent_session.py:1227), as does AgentTask.cancel() (livekit-agents/livekit/agents/voice/agent.py:847-848).
If that happens while a task is queued on _inline_task_lock, the task proceeds when it gets the slot: it pauses the parent activity, starts its own activity and on_enter, and waits for user input. Meanwhile SpeechHandle._cancel armed a 5s timeout that cancels the speech's tasks (livekit-agents/livekit/agents/voice/speech_handle.py:265-277), so the queued task's tool task gets cancelled mid-handoff; the cancellation lands inside the finally handoff-back block of AgentTask.__await_impl. Before this PR the pre-await check guaranteed the invariant "the parent speech is not interrupted" still held at handoff time; with queuing it no longer does.
| async with self._inline_task_lock: | |
| if self._closed: | |
| raise ToolError( | |
| "the activity that awaited the inline task closed while an earlier " | |
| "one was running" | |
| ) | |
| async with self._inline_task_lock: | |
| if self._closed: | |
| raise ToolError( | |
| "the activity that awaited the inline task closed while an earlier " | |
| "one was running" | |
| ) | |
| if speech_handle is not None and speech_handle.interrupted: | |
| # a forced interrupt lands regardless of the hold | |
| raise ToolError( | |
| "the speech that awaited the inline task was interrupted while an " | |
| "earlier one was running" | |
| ) |
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Stacked on #6860, which refuses the second inline
AgentTaskof a turn. This runsthem instead.
Pausing an activity is a single slot. When an LLM turn's parallel tool calls each
await an inline
AgentTask, every agent switch but the last is overwritten, andthose tasks are left awaiting a result nothing can produce — function calls that
never return, a speech that never finishes, and a session wedged until close times
out. The inline tasks now queue for the slot and take it in turn.
Where it lives
AgentActivity._inline_task_slotowns the protocol.AgentTask.__await_implkeepsonly the handoff and asks the activity for the slot.
That placement is the point. The returned-
AgentTaskpath already decides handoffsin one place that can see every result of the turn and pick one; the awaited path
let each coroutine call
session._update_activitydirectly, with no coordination.The slot gives the awaited path the same single decision point — and it is the
object that already owns the drain set, the run-state watching, and the activity
lifecycle the protocol has to be ordered against.
The three orderings
Each step is ordered against the queue, and each inverts into a different hang:
tasks share one speech handle. Releasing the hold per task lets the user turns of
one task's sub-conversation interrupt the speech the rest are anchored to, and an
interrupted handle can no longer disallow interruptions — so every task behind it
fails at
speech_handle.py:129.drain set makes session close wait on the slot it is still queued for, which is
the original hang again.
its turn waits for the user input the task ahead of it needs.
The second and third were found by testing, not by reading.
A nested task pauses a different activity, so it never queues behind the task it is
nested in, and a queue that outlives its activity still refuses with a
ToolError.Tests
test_parallel_agent_tasks_run_in_turndrives one turn emitting two tool calls,each awaiting a task, and asserts the second dialog becomes active as a distinct
instance after the first hands back, with both calls carrying non-refusal outputs.
test_nested_agent_task_no_deadlockandtest_agent_task_close_racecover the twoorderings above and both still pass.
Full
--unit: 1957 passed, with the 5 realtime-plugin failures that reproduceidentically on
main.Reviewing the diff
Most of
agent.py's line count is the mechanical re-indent from wrapping the bodyin
async with. Undergit diff -wit is 47 lines, nearly all deletions — thelogic moves out of
__await_impland into the slot.Follow-up
agents-jshas the same single-slot handoff and is not touched here.