google STT connect doesn't gracefully shutdown when agent disconnected

[brightsparc]

When using the google STT, unlike deepgram STT the run loop doesn't gracefully terminate when agent disconnected:

Error

2024-10-07 10:55:10,339 - WARNING livekit.plugins.google - google stt connection failed, retrying in 0s
Traceback (most recent call last):
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/google/api_core/grpc_helpers_async.py", line 106, in _wrapped_aiter
    async for response in self._call:  # pragma: no branch
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/grpc/aio/_call.py", line 365, in _fetch_stream_responses
    await self._raise_for_status()
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/grpc/aio/_call.py", line 272, in _raise_for_status
    raise _create_rpc_error(
grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:
        status = StatusCode.ABORTED
        details = "Stream timed out after receiving no more client requests."
        debug_error_string = "UNKNOWN:Error received from peer ipv6:%5B2404:6800:4006:809::200a%5D:443 {grpc_message:"Stream timed out after receiving no more client requests.", grpc_status:10, created_time:"2024-10-07T10:55:10.333325+11:00"}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/livekit/plugins/google/stt.py", line 259, in _run
    await self._run_stream(stream)
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/livekit/plugins/google/stt.py", line 279, in _run_stream
    async for resp in stream:
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/google/api_core/grpc_helpers_async.py", line 109, in _wrapped_aiter
    raise exceptions.from_grpc_error(rpc_error) from rpc_error
google.api_core.exceptions.Aborted: 409 Stream timed out after receiving no more client requests. [type_url: "type.googleapis.com/util.StatusProto"
value: "\010\n\022\007generic\0329Stream timed out after receiving no more client requests.*D\013\020\206\326\215'\032;\0229Stream timed out after receiving no more client requests.\014"
] {"pid": 61055, "job_id": "AJ_TUiUCQPBher8"}
2024-10-07 10:55:20,880 - WARNING livekit.plugins.google - google stt connection failed, retrying in 0s
Traceback (most recent call last):
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/google/api_core/grpc_helpers_async.py", line 106, in _wrapped_aiter
    async for response in self._call:  # pragma: no branch
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/grpc/aio/_call.py", line 365, in _fetch_stream_responses
    await self._raise_for_status()
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/grpc/aio/_call.py", line 272, in _raise_for_status
    raise _create_rpc_error(
grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:
        status = StatusCode.ABORTED
        details = "Stream timed out after receiving no more client requests."
        debug_error_string = "UNKNOWN:Error received from peer ipv6:%5B2404:6800:4006:809::200a%5D:443 {created_time:"2024-10-07T10:55:20.88013+11:00", grpc_status:10, grpc_message:"Stream timed out after receiving no more client requests."}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/livekit/plugins/google/stt.py", line 259, in _run
    await self._run_stream(stream)
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/livekit/plugins/google/stt.py", line 279, in _run_stream
    async for resp in stream:
  File "/Users/julian/miniforge3/envs/livekit-py311/lib/python3.11/site-packages/google/api_core/grpc_helpers_async.py", line 109, in _wrapped_aiter
    raise exceptions.from_grpc_error(rpc_error) from rpc_error
google.api_core.exceptions.Aborted: 409 Stream timed out after receiving no more client requests. {"pid": 61055, "job_id": "AJ_TUiUCQPBher8"}
2024-10-07 10:55:22,999 - WARNING livekit - livekit::rtc_engine:410:livekit::rtc_engine - received session close: "signal client closed: \"stream closed\"" UnknownReason Resume {"pid": 61055, "job_id": "AJ_TUiUCQPBher8"}
2024-10-07 10:55:23,006 - INFO livekit.agents - job exiting {"reason": "room disconnected", "pid": 61055, "job_id": "AJ_TUiUCQPBher8"}
2024-10-07 10:55:23,005 - DEBUG livekit.agents - shutting down job task {"reason": "room disconnected", "user_initiated": false, "pid": 61055, "job_id": "AJ_TUiUCQPBher8"}

Repo

Steps to repo, update the minimal agent to use google TTS, providing GOOGLE_APPLICATION_CREDENTIALS env var

from livekit.plugins import deepgram, google, openai, silero

async def entrypoint(ctx: JobContext):
    # ....

    agent = VoicePipelineAgent(
        vad=ctx.proc.userdata["vad"],
        # stt=deepgram.STT(model=dg_model),
        stt=google.STT(),
        llm=openai.LLM(),
        tts=openai.TTS(),
        chat_ctx=initial_ctx,
    )