soniox STT: an endpoint after all non-final tokens were retracted never emits END_OF_SPEECH, leaving the user "speaking"

The issue points to livekit-plugins/livekit-plugins-soniox/livekit/plugins/soniox/stt.py, around line 479, and includes a network-free reproduction script that replaces the websocket with a queue. Read _recv_messages_task and send_endpoint_transcript() first, then run the reproduction with the Soniox plugin. Done means an endpoint after retracted non-final tokens emits END_OF_SPEECH, resets speaking state, and the later utterance emits its own START_OF_SPEECH. A linked open pull request (#7729) indicates work is already underway.

Bug Description

SpeechStream._recv_messages_task emits START_OF_SPEECH on the first token of an utterance, final or non-final. It emits END_OF_SPEECH only from send_endpoint_transcript(), and only inside if final.text:.

When Soniox sends one or more non-final tokens and then retracts them, no final text ever builds up. An <end> / <fin> that arrives after that does nothing. is_speaking stays True, no END_OF_SPEECH is sent, and the session keeps the user in speaking until a later utterance finalises and ends.

Effects with turn_detection="stt":

• user_state stays speaking through silence, so user_away_timeout never starts. Anything keyed on the user going quiet (silence re-prompts, filler, presence logic) does not run.
• The next utterance gets no START_OF_SPEECH of its own, because the plugin thinks the previous one is still going.

We saw this on a production phone call. Soniox sent a single non-final token (one number) at 20.47 s. That token was never finalised: the next final text was a different word alone, at 31.38 s. The user was speaking for the 10.8 s between, while the caller's audio track was flat. The caller finally said "Halo?" to get the agent to talk.

Code: https://github.com/livekit/agents/blob/main/livekit-plugins/livekit-plugins-soniox/livekit/plugins/soniox/stt.py#L479

Expected Behavior

Once the plugin has sent START_OF_SPEECH, an endpoint token (<end> / <fin>) always ends that speech segment with END_OF_SPEECH, even when there is no final text to send with it. The user state goes back to listening, and the next utterance starts with its own START_OF_SPEECH.

Reproduction Steps

1. pip install "livekit-agents[soniox]"   # reproduced on 1.7.1 and 1.8.6
2. Save the script below as repro.py. It swaps the websocket for a queue, so it needs no key and no network.
3. python repro.py

Output:
control: non-final '42'
    -> START_OF_SPEECH, INTERIM_TRANSCRIPT ' 42'
control: '42' finalised, then <end>
    -> FINAL_TRANSCRIPT ' 42', END_OF_SPEECH
bug: non-final '42'
    -> START_OF_SPEECH, INTERIM_TRANSCRIPT ' 42'
bug: '42' retracted, <end> with no final tokens
    -> (no event)                      <- expected END_OF_SPEECH
bug: caller says 'Halo.' later
    -> FINAL_TRANSCRIPT ' Halo.', END_OF_SPEECH    <- no START_OF_SPEECH, state was never reset

"""Replays scripted Soniox websocket messages through the Soniox plugin and prints its events.

Shows that an endpoint which arrives after every non-final token was retracted leaves the
plugin in its speaking state: no END_OF_SPEECH is emitted until a later utterance finalises.

Needs no Soniox account and no network: the websocket is replaced by a queue.

    pip install "livekit-agents[soniox]"
    python repro.py
"""

import asyncio
import json
from typing import Any

import aiohttp
from livekit.agents import stt
from livekit.plugins import soniox

WAIT_FOR_EVENTS_S = 0.5

class FakeSonioxSocket:
    """Stands in for the Soniox websocket: yields queued messages, then waits forever."""

    def __init__(self) -> None:
        """Creates an empty message queue."""
        self.messages: asyncio.Queue[aiohttp.WSMessage] = asyncio.Queue()

    def push(self, tokens: list[dict[str, Any]]) -> None:
        """Queues one Soniox response carrying the given tokens.

        Args:
            tokens: The tokens of the response, as Soniox sends them.
        """
        body = json.dumps({"tokens": tokens, "total_audio_proc_ms": 0})
        self.messages.put_nowait(aiohttp.WSMessage(aiohttp.WSMsgType.TEXT, body, None))

    def __aiter__(self) -> "FakeSonioxSocket":
        """Returns itself as the message iterator."""
        return self

    async def __anext__(self) -> aiohttp.WSMessage:
        """Returns the next queued message, waiting for one if none is queued."""
        return await self.messages.get()

    async def send_str(self, data: str) -> None:
        """Drops a text frame the plugin sends.

        Args:
            data: The frame.
        """

    async def send_bytes(self, data: bytes) -> None:
        """Drops an audio frame the plugin sends.

        Args:
            data: The frame.
        """

    async def close(self) -> None:
        """Does nothing; there is no connection to close."""


def token(text: str, *, is_final: bool) -> dict[str, Any]:
    """Builds one Soniox token.

    Args:
        text: The token text, or "<end>" for an endpoint.
        is_final: Whether Soniox has finalised the token.

    Returns:
        The token as Soniox sends it.
    """
    return {"text": text, "is_final": is_final, "language": "pl", "start_ms": 0, "end_ms": 0}


async def record(stream: stt.SpeechStream, events: list[str]) -> None:
    """Appends every event the plugin emits to a list, one line per event.

    Args:
        stream: The plugin's speech stream.
        events: The list to append to: the event type and, where it has one, its text.
    """
    async for event in stream:
        text = f" {event.alternatives[0].text!r}" if event.alternatives else ""
        events.append(f"{event.type.name}{text}")

async def main() -> None:
    """Runs the control case and the failing case and prints the plugin's events for each."""
    socket = FakeSonioxSocket()

    async def connect(_: soniox.stt.SpeechStream) -> FakeSonioxSocket:
        return socket

    soniox.stt.SpeechStream._connect_ws = connect
    stream = soniox.STT(api_key="unused").stream()

    steps = [
        ("control: non-final '42'", [token(" 42", is_final=False)]),
        ("control: '42' finalised, then <end>", [token(" 42", is_final=True), token("<end>", is_final=True)]),
        ("bug: non-final '42'", [token(" 42", is_final=False)]),
        ("bug: '42' retracted, <end> with no final tokens", [token("<end>", is_final=True)]),
        ("bug: caller says 'Halo.' later", [token(" Halo.", is_final=True), token("<end>", is_final=True)]),
    ]
    events: list[str] = []
    recorder = asyncio.create_task(record(stream, events))
    for label, tokens in steps:
        events.clear()
        socket.push(tokens)
        await asyncio.sleep(WAIT_FOR_EVENTS_S)
        print(f"{label}\n    -> {', '.join(events) or '(no event)'}")

    recorder.cancel()
    await stream.aclose()


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

Operating System

LiveKit Cloud agent (Linux container, eu-central). Reproduced locally on Windows 11.

Models Used

Soniox stt-rt-v5 (max_endpoint_delay_ms=720), turn_detection="stt", Silero VAD, LLM and TTS through LiveKit Inference

Package Versions

livekit-agents==1.7.1
livekit-plugins-soniox==1.7.1
# repro also fails on livekit-agents[soniox]==1.8.6

Session/Room/Call IDs

No response

Proposed Solution

def send_endpoint_transcript() -> None:
    nonlocal is_speaking
    if final.text:
        ...  # unchanged: FINAL_TRANSCRIPT, END_OF_SPEECH, reset, is_speaking = False
    else:
        final_original.reset()
        if is_speaking:
            # every token of the segment was retracted: close it anyway
            self._event_ch.send_nowait(stt.SpeechEvent(type=SpeechEventType.END_OF_SPEECH))
            is_speaking = False

Additional Context

Related, not the same: #7699 / #7717 (final after END_OF_SPEECH in STT mode), #5580 / #6641 (VAD driving user_state in STT mode), #5876 (Soniox END_OF_SPEECH has no timestamp).

Screenshots and Recordings

No response