Repository: livekit/agents
File: livekit-agents/livekit/agents/voice/transcription/synchronizer.py (Python)

Code before:
            await old_task

        await self._impl.aclose()
        self._impl = _SegmentSynchronizerImpl(
            options=self._opts, next_in_chain=self._text_output._next_in_chain
        )

    def rotate_segment(self) -> None:
        if self._closed:
            return

        if self._rotate_segment_atask and not self._rotate_segment_atask.done():
            logger.warning("rotate_segment called while previous segment is still being rotated")

        self._rotate_segment_atask = asyncio.create_task(
            self._rotate_segment_task(self._rotate_segment_atask)
        )

    async def barrier(self) -> None:
        if self._rotate_segment_atask is None:
            return

        # using a while loop in case rotate_segment is called twice (this should not happen, but
        # just in case, we do log a warning if it does)
        while not self._rotate_segment_atask.done():
            await self._rotate_segment_atask


class _SyncedAudioOutput(io.AudioOutput):
    def __init__(
        self, synchronizer: TranscriptSynchronizer, *, next_in_chain: io.AudioOutput
    ) -> None:
        super().__init__(
            label="TranscriptSynchronizer",
            next_in_chain=next_in_chain,
            sample_rate=next_in_chain.sample_rate,
            capabilities=io.AudioOutputCapabilities(pause=True),
        )
        self._next_in_chain: io.AudioOutput = next_in_chain  # redefined for better typing
        self._synchronizer = synchronizer
        self._pushed_duration: float = 0.0

    async def capture_frame(self, frame: rtc.AudioFrame) -> None:
        # using barrier() on capture should be sufficient, flush() must not be called if
        # capture_frame isn't completed
        await self._synchronizer.barrier()

        await self._next_in_chain.capture_frame(frame)  # passthrough audio
        await super().capture_frame(frame)
        self._pushed_duration += frame.duration


Review comment (suggestion):
when pause the `impl`, it records the time it paused, and when it's resumed it's used to compute the duration the impl paused
```python
# in pause()
        if self._paused_wall_time is None:
            self._paused_wall_time = time.time()

# in resume()
        if self._paused_wall_time is not None:
            self._paused_duration += time.time() - self._paused_wall_time
            self._paused_wall_time = None
```

so if we pause the `impl` here once it's created, the `_paused_wall_time` will be earlier than `_start_wall_time` which is the time the first audio frame pushed to the synchronizer. maybe we should also change the `impl.resume()` to check the `start_wall_time`
```python
        if self._paused_wall_time is not None and self._start_wall_time is not None:
            self._paused_duration += time.time() - max(
                self._start_wall_time, self._paused_wall_time
            )
```

Code after:
            self._audio_attached = audio_attached

        if is_given(text_attached):
            self._text_attached = text_attached

        self.set_enabled(self._audio_attached and self._text_attached)

    async def _rotate_segment_task(self, old_task: asyncio.Task[None] | None) -> None:
        if old_task:
            with contextlib.suppress(Exception):
                await old_task

        try:
            await self._impl.aclose()
        except Exception:
            logger.exception("failed to close segment synchronizer impl during rotation")

        # always create a new impl even if aclose() failed, to avoid leaving
        # self._impl pointing to a closed impl which causes the agent to get stuck
        self._impl = _SegmentSynchronizerImpl(
            options=self._opts, next_in_chain=self._text_output._next_in_chain
        )

        # apply the current pause state to the new impl
        if self._paused:
            self._impl.pause()

    def rotate_segment(self) -> None:
        if self._closed:
            return

        if self._rotate_segment_atask and not self._rotate_segment_atask.done():
            logger.warning("rotate_segment called while previous segment is still being rotated")

        self._rotate_segment_atask = asyncio.create_task(
            self._rotate_segment_task(self._rotate_segment_atask)
        )

    async def barrier(self) -> None:
        if self._rotate_segment_atask is None:
            return

        # using a while loop in case rotate_segment is called twice (this should not happen, but
        # just in case, we do log a warning if it does)
        while not self._rotate_segment_atask.done():
            await self._rotate_segment_atask


class _SyncedAudioOutput(io.AudioOutput):
    def __init__(
        self, synchronizer: TranscriptSynchronizer, *, next_in_chain: io.AudioOutput