Repository: livekit/agents
File: livekit-plugins/livekit-plugins-inworld/livekit/plugins/inworld/tts.py (Python)

Code before:
                    # Resolve waiter before evicting
                    if ctx.waiter and not ctx.waiter.done():
                        ctx.waiter.set_result(None)
                    self._contexts.pop(ctx.context_id, None)
                    self._last_activity = now
                    self._context_available.set()
                    if self._on_capacity_available:
                        self._on_capacity_available()

    async def _handle_connection_error(self, error: Exception) -> None:
        """Handle connection-level error by failing all active contexts."""
        for ctx in list(self._contexts.values()):
            if ctx.waiter and not ctx.waiter.done():
                ctx.waiter.set_exception(error)
        self._contexts.clear()
        self._closed = True

        # Wake local waiters so they can fail fast
        self._context_available.set()

        # Notify pool so callers blocked on capacity can create replacement connections
        if self._on_capacity_available:
            self._on_capacity_available()

    async def aclose(self) -> None:
        """Close the connection and all contexts."""
        self._closed = True

        # Cancel background tasks
        for task in [self._send_task, self._recv_task, self._cleanup_task]:
            if task:
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

        # Close all contexts and fail their waiters
        for ctx in list(self._contexts.values()):
            if ctx.waiter and not ctx.waiter.done():
                ctx.waiter.cancel()

        self._contexts.clear()

        if self._ws:
            await self._ws.close()
            self._ws = None


DEFAULT_MAX_CONNECTIONS = 20
DEFAULT_IDLE_CONNECTION_TIMEOUT = 300.0  # 5 minutes

Review comment (bug):
would it be possible to only log this if timestamps are enabled? these logs may be noisy:
` 21:09:01.489 DEBUG  livekit.…s.inworld flushCompleted - updating cumulative time {"context_id": "3e9772406a15", "old_cumulative_time": 0.0, "new_cumulative_time": 0.0}`

Code after:
                                        "context_id": context_id,
                                        "words": [str(ts) for ts in timed_strings],
                                        "adjusted_starts": [ts.start_time for ts in timed_strings],
                                        "adjusted_ends": [ts.end_time for ts in timed_strings],
                                        "generation_end_time": ctx.generation_end_time,
                                    },
                                )

                            for ts in timed_strings:
                                ctx.emitter.push_timed_transcript(ts)

                        if audio_content := audio_chunk.get("audioContent"):
                            ctx.emitter.push(base64.b64decode(audio_content))
                    continue

                if "flushCompleted" in result:
                    # Signals the end of a generation, subsequent timestampes from the server
                    # will reset offset to 0. We need to update the cumulative time to the
                    # generation end time to maintain monotonically increasing timestamps
                    # within the agent turn.
                    if ctx.generation_end_time > 0:
                        logger.debug(
                            "flushCompleted - updating cumulative time",
                            extra={
                                "context_id": context_id,
                                "old_cumulative_time": ctx.cumulative_time,
                                "new_cumulative_time": ctx.generation_end_time,
                            },
                        )
                        ctx.cumulative_time = ctx.generation_end_time
                    continue

                if "contextClosed" in result:
                    if ctx.waiter and not ctx.waiter.done():
                        ctx.waiter.set_result(None)
                    self._contexts.pop(context_id, None)
                    self._last_activity = time.time()
                    self._context_available.set()
                    if self._on_capacity_available:
                        self._on_capacity_available()
                    continue

        except Exception as e:
            logger.error("Inworld recv loop error", exc_info=e)
        finally:
            if not self._closed:
                await self._handle_connection_error(APIConnectionError())

    async def _cleanup_stale_contexts(self) -> None:
        """Periodically clean up orphaned contexts stuck in CLOSING state."""
        while not self._closed: