Hacktoberfest 2026: the issues maintainers tagged for October, open and beginner-friendly. Browse Hacktoberfest issues

StreamPacerWrapper.end_input() sets _input_ended before buffered sentences reach _sentences; last sentence(s) silently never sent to TTS

Open
  • Is something wrong?
#7,810 0 comments 0 reactions 0 assignees View on GitHub

Maintainers usually reply within 1 day

@Hugo-DDT is already working on this.

Since Oct 14, 2026.

  • #7816 by @Hugo-DDT — open

Assessment

Difficulty
3/5
Estimated time
1-2 days
Newbie friendliness
25/100
Issue type
Bug
Clarity
Clearly specified
Activity status
Stale
Tech stack
python
Domain
audio-video-rtc

Research direction

The race is in livekit/agents/tts/stream_pacer.py: end_input() sets _input_ended before the sentence tokenizer flushes, and _send_task can break on _input_ended and not _sentences. Read end_input(), _recv_task's finally block and the _send_task loop, then run the included repro_pacer_tail.py with k from -6 to 12. Done means every k reports FULL. Pull request #7816 is already open against this, so check it before starting.

Written by the indexing model from the issue text.

Description

Summary

StreamPacerWrapper.end_input() sets _input_ended = True before the sentence tokenizer has delivered the buffered tail to _recv_task. If the send loop wakes up in that window, _send_task sees _input_ended and not _sentences, breaks, and the last sentence(s) are never sent to TTS. No error is raised: the turn finishes "successfully" with the end of the reply missing.

We hit this on a deployed voice agent: the reply was cut after the first sentence, the rest of the LLM text never reached TTS.

Versions

livekit-agents 1.6.10 through 1.8.7; livekit-agents/livekit/agents/tts/stream_pacer.py on main is identical. Python 3.14, Windows and Linux.

Setup: StreamAdapter(tts=<non-streaming TTS>, text_pacing=SentenceStreamPacer(min_remaining_audio=3.0, max_text_length=200)).

Cause
def end_input(self) -> None:
    self._sent_stream.end_input()
    self._input_ended = True          # set before the tokenizer flushed its tail
    ...

self._sent_stream.end_input() only schedules the tokenizer flush; the final sentence is appended to self._sentences by _recv_task on a later event-loop iteration. Meanwhile _send_task can be woken by the pending call_later wakeup timer (or _wakeup_event), and then

if self._closing or (self._input_ended and not self._sentences):
    break

exits the loop with the tail still in flight. _event_ch closes, StreamAdapterWrapper._synthesize finishes, and the remaining text is dropped.

Deterministic repro (no network, no wall clock)

The pacer wakeup (what its timer does) is injected k event-loop iterations before/after stream.end_input(). Only k=0 loses the tail, which matches the timing window above.

repro_pacer_tail.py 
import asyncio

from livekit.agents import APIConnectOptions, tts, utils
from livekit.agents.tts import SentenceStreamPacer, StreamAdapter

SAMPLE_RATE = 24000
REPLY = (
    "I'm doing well, thanks! How are you? Today we'll practice minimal "
    "pairs, words that differ by just one sound, like sheep and ship."
)

synthesized: list[str] = []
pacers: list[object] = []

_orig_wrap = SentenceStreamPacer.wrap


def _recording_wrap(self, sent_stream, audio_emitter):
    wrapper = _orig_wrap(self, sent_stream, audio_emitter)
    pacers.append(wrapper)
    return wrapper


SentenceStreamPacer.wrap = _recording_wrap


class FakeChunkedStream(tts.ChunkedStream):
    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        synthesized.append(self._input_text)
        output_emitter.initialize(
            request_id=utils.shortuuid(),
            sample_rate=SAMPLE_RATE,
            num_channels=1,
            mime_type="audio/pcm",
        )
        await asyncio.sleep(0.6)  # network latency
        seconds = len(self._input_text) * 0.08
        sent = 0.0
        while sent < seconds:
            output_emitter.push(b"\x00\x00" * int(SAMPLE_RATE * 0.1))
            sent += 0.1
            await asyncio.sleep(0.01)
        output_emitter.flush()


class FakeTTS(tts.TTS):
    def __init__(self) -> None:
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=SAMPLE_RATE,
            num_channels=1,
        )

    def synthesize(self, text, *, conn_options=APIConnectOptions()):
        return FakeChunkedStream(tts=self, input_text=text, conn_options=conn_options)


async def run(k: int) -> tuple[int, float]:
    synthesized.clear()
    pacers.clear()
    adapter = StreamAdapter(
        tts=FakeTTS(),
        text_pacing=SentenceStreamPacer(min_remaining_audio=3.0, max_text_length=200),
    )
    stream = adapter.stream()

    async def produce() -> None:
        for i in range(0, len(REPLY), 4):
            stream.push_text(REPLY[i : i + 4])
            await asyncio.sleep(0.003)
        # first sentence is in TTS, pacer idles on its wakeup timer with an empty queue
        while not (
            pacers
            and synthesized
            and pacers[0]._wakeup_timer is not None
            and not pacers[0]._sentences
        ):
            await asyncio.sleep(0.003)
        pacer = pacers[0]
        if k >= 0:
            stream.end_input()
            for _ in range(k):
                await asyncio.sleep(0)
            pacer._wakeup_event.set()
        else:
            pacer._wakeup_event.set()
            for _ in range(-k):
                await asyncio.sleep(0)
            stream.end_input()

    async def consume() -> float:
        played = 0.0
        async for ev in stream:
            played += ev.frame.duration
        return played

    _, played = await asyncio.gather(produce(), consume())
    await stream.aclose()
    return len(synthesized), played


async def main() -> None:
    import livekit.agents

    print("livekit-agents", livekit.agents.__version__)
    for k in range(-6, 13):
        requests, played = await run(k)
        verdict = "FULL" if requests == 2 else "TAIL DROPPED"
        print(f"k={k:+3d} tts_requests={requests} {verdict} audio={played:.2f}s", flush=True)


asyncio.run(main())

Output on 1.8.7:

livekit-agents 1.8.7
k= -1 tts_requests=2 FULL audio=10.43s
k= +0 tts_requests=1 TAIL DROPPED audio=1.92s
k= +1 tts_requests=2 FULL audio=10.43s
(all other k in -6..+12: FULL)
Suggested fix

Do not set _input_ended in end_input(). _recv_task already sets it in its finally once the sentence stream is exhausted, i.e. after every sentence is in self._sentences:

def end_input(self) -> None:
    self._sent_stream.end_input()
    if self._audio_emitter._dst_ch.closed:
        self._closing = True
        self._wakeup_event.set()

With this change the deterministic repro gives FULL for all k, and a timer-based sweep (25 offsets around the wakeup deadline) went from 2 dropped tails to 0.

Related but different: #3133, #3140 (adds _wakeup_event.set() but keeps the _input_ended and not _sentences race), #4440.

Dominant language
Python
Stars
14.6k
Forks
3.9k
Avg merge
3d 7h
Merged PRs (30d)
165

Getting set up

This project ships no dev container, Dockerfile or contributing guide, so setting up is up to you: start from its README, and see our first-contribution guide for the general steps.

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from livekit/agents

All issues in livekit/agents

Similar issues

More Python issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.