AgentTask handoff deadlocks when a preemptive generation is left parked

[AdemFr]

Bug Description

We saw some Agent tasks hanging, because users would say something at exactly the wrong time, which left some preemptable generation task hanging and the AgentTask waited for it to clear.

This was also indicated by a previous skipping reply to user input, current speech generation cannot be interrupted.log, which pointed towards an existing pending user turn that was being committed during non interruptible agent speech.

We have a workaround that removes all preemptable speech before awaiting any tasks, which seems to have fixed it for us. Couldn't find that this issue would be fixed in a later livekit version.

Some context written up by AI:

Symptom

• Awaiting an AgentTask never returns: the task's on_enter never runs and the call sits in silence.
• It only unblocks when the caller hangs up, at which point the session close logs session is closing, skipping <activity> of <agent id> and the task completes with ToolError("activity doesn't start for ..., likely due to session closing").
• The telltale earlier log is skipping reply to user input, current speech generation cannot be interrupted.

When it happens

• Preemptive generation is enabled (preemptive_generation={"enabled": True}), and
• the caller starts speaking while nothing is playing, so LiveKit starts a speculative reply, and
• the agent then starts an uninterruptible speech (allow_interruptions=False) — for us the transfer hold message, or any AgentTask await, since AgentTask.__await_impl sets allow_interruptions = False itself, and
• the caller's turn ends while that uninterruptible speech is still playing, and
• an AgentTask is awaited afterwards.

Mechanism

• _user_turn_completed_task (agent_activity.py:2433) sees an uninterruptible current_speech, discards the user turn, and returns early — without cancelling the speculative reply it had already started. The reply stays parked in _preemptive_generation, never scheduled.
• Awaiting an AgentTask calls session._update_activity(..., previous_activity="pause", ...) (agent.py:969), which calls AgentActivity.pause().
• pause() waits for all outstanding speech work, and the parked generation is neither scheduled nor cancelled, so the wait never completes.
• drain(), aclose(), and interrupt() all call _cancel_preemptive_generation() before draining. pause() is the only one that doesn't — and it's the path AgentTask takes.

Expected Behavior

Discarded preemptable speech should never block awaiting an agent task.

Reproduction Steps

"""Minimal repro: a parked preemptive generation deadlocks AgentActivity.pause()."""

from __future__ import annotations

import asyncio
import time
from collections.abc import Callable
from typing import Any

from livekit.agents import Agent, AgentSession
from livekit.agents.llm import LLM, ChatChunk, ChatContext, ChoiceDelta, LLMStream
from livekit.agents.types import (
    DEFAULT_API_CONNECT_OPTIONS,
    NOT_GIVEN,
    APIConnectOptions,
    NotGivenOr,
)
from livekit.agents.voice.audio_recognition import (
    _EndOfTurnInfo,
    _EndOfTurnMetrics,
    _PreemptiveGenerationInfo,
)

TIMEOUT = 2.0


class _GatedStream(LLMStream):
    """Emits one token, then holds the generation open until released."""

    def __init__(self, *args: Any, gate: asyncio.Event, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self._gate = gate

    async def _run(self) -> None:
        self._event_ch.send_nowait(
            ChatChunk(id="chunk", delta=ChoiceDelta(role="assistant", content="one moment"))
        )
        await self._gate.wait()


class _GatedLLM(LLM):
    """Text-only LLM whose replies keep playing until ``release()``."""

    def __init__(self) -> None:
        super().__init__()
        self.gate = asyncio.Event()

    @property
    def model(self) -> str:
        return "gated"

    @property
    def provider(self) -> str:
        return "test"

    def release(self) -> None:
        self.gate.set()

    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[Any] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[Any] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        return _GatedStream(
            self,
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
            gate=self.gate,
        )


async def _wait_until(predicate: Callable[[], bool], *, timeout: float = TIMEOUT) -> None:
    async with asyncio.timeout(timeout):
        while not predicate():
            await asyncio.sleep(0.01)


def _end_of_turn(transcript: str) -> _EndOfTurnInfo:
    return _EndOfTurnInfo(
        skip_reply=False,
        new_transcript=transcript,
        transcript_confidence=1.0,
        metrics=_EndOfTurnMetrics(
            started_speaking_at=None,
            stopped_speaking_at=None,
            transcription_delay=None,
            end_of_turn_delay=None,
        ),
    )


async def pause_completes(*, drop_parked_reply: bool) -> bool:
    """Return True if the AgentTask handoff's pause() finishes within the timeout."""
    llm = _GatedLLM()
    session: AgentSession[None] = AgentSession(llm=llm)
    await session.start(agent=Agent(instructions="qualify the caller"))

    activity = session._activity
    assert activity is not None

    # 1. the caller starts talking, so a speculative reply is started (STT interim transcript).
    #    only accepted while no speech is playing, as between two agent turns.
    assert session.current_speech is None
    activity.on_preemptive_generation(
        _PreemptiveGenerationInfo(
            new_transcript="what is the rate",
            transcript_confidence=1.0,
            started_speaking_at=time.time(),
        )
    )
    assert activity._preemptive_generation is not None

    # 2. the agent starts an uninterruptible message (e.g. a transfer hold message)
    session.generate_reply(instructions="ask them to hold", allow_interruptions=False)
    await _wait_until(lambda: session.current_speech is not None)

    # 3. the caller's turn ends while that message is still playing. _user_turn_completed_task
    #    discards the user turn and returns WITHOUT cancelling the speculative reply.
    activity.on_end_of_turn(_end_of_turn("can you tell me the rate"))
    await _wait_until(
        lambda: (
            activity._user_turn_completed_atask is not None
            and activity._user_turn_completed_atask.done()
        )
    )

    # 4. the uninterruptible message finishes; the parked reply is the only speech work left
    llm.release()
    await _wait_until(lambda: session.current_speech is None)
    assert activity._preemptive_generation is not None

    if drop_parked_reply:
        activity._cancel_preemptive_generation()

    # 5. an AgentTask is awaited, which pauses this activity.
    #    this is the exact call AgentTask.__await_impl makes (agent.py:969).
    pause = asyncio.create_task(
        session._update_activity(
            Agent(instructions="hand off"),
            previous_activity="pause",
            blocked_tasks=[],
            wait_on_enter=False,
        )
    )
    try:
        await asyncio.wait_for(asyncio.shield(pause), timeout=TIMEOUT)
        return True
    except asyncio.TimeoutError:
        return False
    finally:
        pause.cancel()
        await asyncio.gather(pause, return_exceptions=True)
        await asyncio.gather(session.aclose(), return_exceptions=True)


async def main() -> None:
    hung = not await pause_completes(drop_parked_reply=False)
    print(f"without cleanup: pause() {'HUNG (bug)' if hung else 'completed'}")

    completed = await pause_completes(drop_parked_reply=True)
    print(f"with cleanup:    pause() {'completed' if completed else 'HUNG'}")


if __name__ == "__main__":
    asyncio.run(main())

Operating System

macOS, Linux

Models Used

No response

Package Versions

livekit                                  1.1.14
livekit-agents                           1.6.8
livekit-api                              1.2.0
livekit-blingfire                        1.1.0
livekit-local-inference                  0.2.6
livekit-protocol                         1.1.22

Session/Room/Call IDs

No response

Proposed Solution

- Possibly fix at the source: `_user_turn_completed_task` should call `_cancel_preemptive_generation()` on the early-return path where it discards the user turn, since that reply can never be used.
- And / Or have `pause()` drop the parked preemptive generation the way `drain()`/`aclose()`/`interrupt()` already do.

Additional Context

No response

Screenshots and Recordings

No response

[uuzzrm]

Opened a PR for this: #6865

The fix drops the parked preemptive generation in two places: the turn-discard
path in _user_turn_completed_task (the reply can never be scheduled once the
turn is discarded), and pause() itself, so a handoff can never wait on a
generation nobody scheduled - matching what drain()/aclose()/interrupt()
already do. Added a regression test covering both guards.