Cancelling a wait_for_idle caller can cancel in-flight user-turn work and poison subsequent turns

[pseokh97]

Bug Description

AgentActivity.wait_for_idle() directly awaits the activity's in-flight end-of-turn and user-turn-completion tasks.

When a task awaiting wait_for_idle() is cancelled, asyncio propagates that cancellation to the SDK-owned task currently being awaited. If _user_turn_completed_atask is active, it is cancelled as a consequence.

The next user turn is still accepted by on_end_of_turn(), but its new _user_turn_completed_task() begins by awaiting the cancelled predecessor through await old_task. This raises CancelledError before the latest transcript can be committed or a reply can be generated.

The cancelled successor is then stored as the predecessor of the following turn, so the same cancellation can repeat across subsequent accepted turns.

This reproduces using only LiveKit Agents. No STT, LLM, TTS, RealtimeModel, turn detector, LiveKit room, or application-specific orchestration is involved.

Expected Behavior

Cancelling a caller that is awaiting wait_for_idle() must not cancel the in-flight SDK-owned end-of-turn or user-turn-completion tasks that it is observing.

When chaining user-turn-completion tasks, cancellation of a predecessor must not automatically cancel the latest accepted turn unless the latest task itself is cancelled or the activity/session is closing.

Reproduction Steps

python - <<'PY'
import asyncio

from livekit.agents import Agent, AgentSession
from livekit.agents.voice.agent_activity import AgentActivity
from livekit.agents.voice.audio_recognition import (
    _EndOfTurnInfo,
    _EndOfTurnMetrics,
)


def end_of_turn(turn_number: int) -> _EndOfTurnInfo:
    return _EndOfTurnInfo(
        skip_reply=False,
        new_transcript=f"turn {turn_number}",
        transcript_confidence=1.0,
        metrics=_EndOfTurnMetrics(
            started_speaking_at=None,
            stopped_speaking_at=None,
            transcription_delay=None,
            end_of_turn_delay=None,
        ),
    )


async def main() -> None:
    session = AgentSession()
    activity = AgentActivity(Agent(instructions="test"), session)

    # Put the activity into its normal turn-accepting state.
    activity._scheduling_paused = False
    activity._new_turns_blocked = False

    active_turn = asyncio.create_task(
        asyncio.Event().wait(),
        name="active-user-turn",
    )
    activity._user_turn_completed_atask = active_turn

    # wait_for_idle() directly awaits active_turn.
    idle_observer = asyncio.create_task(activity.wait_for_idle())
    await asyncio.sleep(0)

    # Cancelling only the observer also cancels the SDK-owned turn task.
    idle_observer.cancel()
    await asyncio.gather(idle_observer, return_exceptions=True)

    print(f"turn_1_cancelled={active_turn.cancelled()}")

    # Every later turn is accepted, but it awaits the cancelled predecessor
    # and becomes cancelled before processing the new transcript.
    for turn_number in (2, 3, 4):
        accepted = activity.on_end_of_turn(end_of_turn(turn_number))
        await asyncio.sleep(0)

        successor = activity._user_turn_completed_atask
        print(
            f"turn_{turn_number}_accepted={accepted},"
            f"cancelled={successor.cancelled() if successor else None}"
        )

        if successor is not None:
            await asyncio.gather(successor, return_exceptions=True)


asyncio.run(main())
PY

# Actual output:
#
# turn_1_cancelled=True
# turn_2_accepted=True,cancelled=True
# turn_3_accepted=True,cancelled=True
# turn_4_accepted=True,cancelled=True

Operating System

macOS Tahoe

Models Used

N/A — no STT, LLM, TTS, RealtimeModel, or turn detector is required.

Package Versions

Python==3.14.6
livekit-agents==1.6.10

Session/Room/Call IDs

No response

Proposed Solution

Additional Context

No response

Screenshots and Recordings

No response