voice: interrupted realtime response leaves never-played messages in the server-side chat context (any_skipped guard can't fire for them)

[Robert54]

Version: livekit-agents 1.6.5 — still present on main @ 5c2acd2.

Summary

When a realtime response produces multiple messages (e.g. gpt-realtime preamble + final answer) and the user interrupts after message A has finished playing but before message B is pulled from message_stream, B is never played and never committed to the local chat context — but it remains in the server-side conversation. On the next turn the model believes it already said B, so it references an answer the user never heard.

AgentActivity._realtime_generation clearly intends to handle this. _process_messages deliberately leaves unconsumed messages out of message_outputs:

async def _process_messages() -> None:
    async for msg in generation_ev.message_stream:
        if speech_handle.interrupted:
            # remaining messages are left out of message_outputs so
            # update_chat_ctx below removes them server-side.
            break
        entry = await _process_one_message(msg)
        message_outputs.append(entry)
        if entry.out.played == "partial":
            break

But the cleanup itself is guarded on any_skipped, which is computed by scanning message_outputs — the very list those messages were left out of:

any_skipped = False
for entry in message_outputs:
    if entry.out.played == "skipped":
        any_skipped = True
        continue
    ...

# sync local chat ctx to the realtime server to remove any items the
# model added but the user never heard (interrupted before we pulled
# them, or message_outputs entries left in "skipped")
if speech_handle.interrupted and any_skipped and self.llm.capabilities.mutable_chat_context:
    await self._rt_session.update_chat_ctx(self._agent._chat_ctx)

So for the exact case the comment describes ("interrupted before we pulled them"), any_skipped is always False and update_chat_ctx() never runs.

The two leak paths

1. Interrupt while waiting on the next message — _process_messages sees speech_handle.interrupted and breaks; the remaining messages never enter message_outputs, so they can't set any_skipped.
2. Interrupt mid-message — message A ends with played == "partial" → break. A itself is handled via truncate(), but a B the server already created is never pulled — same leak.

Conversely, the window where any_skipped can actually become True is very narrow: a message must have been pulled into message_outputs, gone through forward_generation, and played zero frames. In practice the guard is inverted relative to its intent.

Repro

Self-contained, no network — a hermetic RealtimeModel whose session commits conversation items in generate_reply() the way a realtime server does, then an interrupt lands in the window between A's playout and B's pull:

repro_unplayed_message.py

"""Repro: interrupted realtime response leaves a never-played message in the
server-side chat context.

Scenario: a realtime response produces two messages A ("played") and B
("unplayed"). The server commits both conversation items when it creates them.
The agent plays A fully, and the user interrupts while B has not yet been
pulled from the message stream. B is never played and never committed locally,
but the `update_chat_ctx()` cleanup never runs because `any_skipped` only scans
`message_outputs` - the list B was deliberately left out of.

Run: python repro_unplayed_message.py   (livekit-agents 1.6.5)
"""

import asyncio
import time
from collections.abc import AsyncIterator, Callable, Sequence
from typing import Literal

from livekit import rtc
from livekit.agents import Agent, AgentSession, llm
from livekit.agents.types import NOT_GIVEN, NotGivenOr
from livekit.agents.voice import io

MessageStreamFactory = Callable[[], AsyncIterator[llm.MessageGeneration]]


async def _items(items):
    for item in items:
        yield item


def _modalities(*values):
    future = asyncio.get_running_loop().create_future()
    future.set_result(list(values))
    return future


def _audio_frame(duration: float = 0.02) -> rtc.AudioFrame:
    sample_rate = 16000
    samples = int(sample_rate * duration)
    return rtc.AudioFrame(
        data=b"\x00\x00" * samples,
        sample_rate=sample_rate,
        num_channels=1,
        samples_per_channel=samples,
    )


class HermeticRealtimeSession(llm.RealtimeSession):
    def __init__(
        self,
        model: llm.RealtimeModel,
        message_stream_factory: MessageStreamFactory,
        server_created_message_ids: Sequence[str] = (),
    ) -> None:
        super().__init__(model)
        self._chat_context = llm.ChatContext.empty()
        self._tool_context = llm.ToolContext.empty()
        self._message_stream_factory = message_stream_factory
        self._server_created_message_ids = tuple(server_created_message_ids)

    @property
    def chat_ctx(self) -> llm.ChatContext:
        return self._chat_context

    @property
    def tools(self) -> llm.ToolContext:
        return self._tool_context

    async def update_instructions(self, instructions: str) -> None:
        pass

    async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
        self._chat_context = chat_ctx.copy()

    async def update_tools(self, tools) -> None:
        self._tool_context = llm.ToolContext(tools)

    def update_options(self, *, tool_choice=NOT_GIVEN) -> None:
        pass

    def push_audio(self, frame: rtc.AudioFrame) -> None:
        pass

    def push_video(self, frame: rtc.VideoFrame) -> None:
        pass

    def generate_reply(
        self,
        *,
        instructions: NotGivenOr[str] = NOT_GIVEN,
        tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN,
        tools=NOT_GIVEN,
    ) -> asyncio.Future[llm.GenerationCreatedEvent]:
        # The realtime server commits a conversation item as soon as it creates
        # a message for the response, well before the agent consumes it.
        for message_id in self._server_created_message_ids:
            self._chat_context.add_message(
                role="assistant",
                content=f"server content for {message_id}",
                id=message_id,
            )

        event = llm.GenerationCreatedEvent(
            message_stream=self._message_stream_factory(),
            function_stream=_items([]),
            user_initiated=True,
            response_id="response-1",
        )
        future = asyncio.get_running_loop().create_future()
        future.set_result(event)
        return future

    def commit_audio(self) -> None:
        pass

    def clear_audio(self) -> None:
        pass

    def interrupt(self) -> None:
        pass

    def truncate(
        self,
        *,
        message_id: str,
        modalities: list[Literal["text", "audio"]],
        audio_end_ms: int,
        audio_transcript: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        pass

    async def aclose(self) -> None:
        pass


class HermeticRealtimeModel(llm.RealtimeModel):
    def __init__(
        self,
        message_stream_factory: MessageStreamFactory,
        server_created_message_ids: Sequence[str] = (),
    ) -> None:
        super().__init__(
            capabilities=llm.RealtimeCapabilities(
                message_truncation=True,
                turn_detection=False,
                user_transcription=False,
                auto_tool_reply_generation=False,
                audio_output=True,
                manual_function_calls=False,
                mutable_chat_context=True,
                mutable_instructions=True,
                mutable_tools=True,
                per_response_tool_choice=True,
            )
        )
        self._message_stream_factory = message_stream_factory
        self._server_created_message_ids = tuple(server_created_message_ids)
        self.last_session: HermeticRealtimeSession | None = None

    def session(self) -> llm.RealtimeSession:
        self.last_session = HermeticRealtimeSession(
            self,
            self._message_stream_factory,
            server_created_message_ids=self._server_created_message_ids,
        )
        return self.last_session

    async def aclose(self) -> None:
        pass


class AutoFinishAudioOutput(io.AudioOutput):
    def __init__(self) -> None:
        super().__init__(
            label="AutoFinishAudioOutput",
            capabilities=io.AudioOutputCapabilities(pause=True),
        )
        self._playback_duration = 0.0
        self._playback_active = False

    async def capture_frame(self, frame: rtc.AudioFrame) -> None:
        await super().capture_frame(frame)
        self._playback_duration += frame.duration
        if not self._playback_active:
            self._playback_active = True
            self.on_playback_started(created_at=time.time())

    def flush(self) -> None:
        super().flush()
        if not self._playback_active:
            return
        playback_position = self._playback_duration
        self._playback_duration = 0.0
        self._playback_active = False
        self.on_playback_finished(playback_position=playback_position, interrupted=False)

    def clear_buffer(self) -> None:
        if not self._playback_active:
            return
        self._playback_active = False
        self.on_playback_finished(
            playback_position=self._playback_duration, interrupted=True
        )


async def main() -> None:
    played_out = asyncio.Event()
    interrupt_landed = asyncio.Event()

    async def message_stream() -> AsyncIterator[llm.MessageGeneration]:
        yield llm.MessageGeneration(
            message_id="played",
            text_stream=_items(["First answer."]),
            audio_stream=_items([_audio_frame()]),
            modalities=_modalities("text", "audio"),
        )
        # We are only resumed when the agent pulls the next message, which
        # happens after "played" fully finished its playout. That is exactly
        # the window where the interrupt has to land.
        played_out.set()
        await interrupt_landed.wait()
        yield llm.MessageGeneration(
            message_id="unplayed",
            text_stream=_items(["Second answer."]),
            audio_stream=_items([_audio_frame()]),
            modalities=_modalities("text", "audio"),
        )

    model = HermeticRealtimeModel(
        message_stream, server_created_message_ids=("played", "unplayed")
    )
    audio_output = AutoFinishAudioOutput()

    async with AgentSession(llm=model) as session:
        session.output.audio = audio_output
        await session.start(Agent(instructions="Answer briefly."), record=False)

        run_task = asyncio.ensure_future(session.run(user_input="Hello"))
        await asyncio.wait_for(played_out.wait(), timeout=2.0)

        speech = session.current_speech
        assert speech is not None

        # interrupt() only returns once the speech is done, and the speech
        # cannot finish while the message stream is parked on interrupt_landed,
        # so drive the interrupt as a task and release "unplayed" once the
        # interrupted flag has landed.
        interrupt_task = asyncio.ensure_future(session.interrupt(force=True))
        while not speech.interrupted:
            await asyncio.sleep(0)
        interrupt_landed.set()

        await asyncio.wait_for(interrupt_task, timeout=2.0)
        await asyncio.wait_for(run_task, timeout=2.0)

        local_ids = [
            m.id for m in session.history.messages() if m.role == "assistant"
        ]

    assert model.last_session is not None
    server_ids = [
        item.id
        for item in model.last_session.chat_ctx.items
        if getattr(item, "role", None) == "assistant"
    ]

    print(f"assistant messages committed locally : {local_ids}")
    print(f"assistant messages left server-side  : {server_ids}")
    assert server_ids == local_ids, (
        "BUG: the user never heard 'unplayed', but it survived in the "
        "server-side conversation"
    )


asyncio.run(main())

Output on 1.6.5:

assistant messages committed locally : ['played']
assistant messages left server-side  : ['played', 'unplayed']
AssertionError: BUG: the user never heard 'unplayed', but it survived in the server-side conversation

Suggested fix

Track the abandon condition where it happens, instead of inferring it from message_outputs:

messages_abandoned = False

async def _process_messages() -> None:
    nonlocal messages_abandoned
    async for msg in generation_ev.message_stream:
        if speech_handle.interrupted:
            messages_abandoned = True
            break
        entry = await _process_one_message(msg)
        message_outputs.append(entry)
        if entry.out.played == "partial":
            messages_abandoned = True
            break

...

if (
    speech_handle.interrupted
    and (any_skipped or messages_abandoned)
    and self.llm.capabilities.mutable_chat_context
):

Verified locally against 1.6.5: with this change the repro above passes. One caveat: at the break we don't know whether more messages were actually pending, so when the interrupted message happens to be the last one of the response this performs one redundant (idempotent) update_chat_ctx() sync.

Happy to send a PR if this direction looks right.