[Bug] Gemini Realtime plugin infinite error loop on network disconnection

[PhiVanDuc]

Description

When the network connection is suddenly lost during a Gemini Realtime session,
the plugin enters an infinite error loop instead of gracefully handling the disconnection.

Environment

• OS: Windows
• Python: 3.13.12
• vision-agents[gemini,getstream,redis] >= 0.5.0

Steps to Reproduce

1. Start the agent server: uv run main.py serve
2. Join a call successfully via POST /api/stream/agent/join
3. Disconnect the network suddenly (e.g., disable Wi-Fi/Ethernet)
4. Observe infinite error spam in logs

Expected Behavior

The plugin should catch ConnectionClosedError, stop the receive loop,
emit RealtimeDisconnectedEvent, and exit gracefully.

Actual Behavior

The _processing_loop in gemini_realtime.py keeps looping and
emitting errors indefinitely (see screenshot above).

Code to Reproduce

import contextvars
from dotenv import load_dotenv
from pydantic import BaseModel

from vision_agents.plugins import getstream, gemini
from vision_agents.core import Agent, AgentLauncher, Runner, User

load_dotenv()

current_call_config = contextvars.ContextVar("current_call_config", default = None)

class JoinRequestData(BaseModel):
    id: str
    name: str
    image: str
    call_id: str
    call_type: str
    instructions: str

async def create_agent(**kwargs) -> Agent:
    config = current_call_config.get()

    if config is None:
        user_image = ""
        user_id = "warmup_id"
        user_name = "Warmup Agent"
        final_instructions = "You are a helpful assistant."
    else:
        user_id = config["id"]
        user_name = config["name"]
        user_image = config["image"]
        
        final_instructions = f"""
            {config["instructions"]}

            LANGUAGE PROTOCOL:
            1. Initially, speak in the same language as these instructions.
            2. Maintain this language unless the user explicitly asks to switch.
            3. If a switch is requested, use the new language for all subsequent interactions.

            RESPONSE RULES:
            1. THIS IS IMPORTANTEST RULE - Never use markdown, bold, italic, bullet points, numbered lists, or any text formatting. Only use formatting that is compatible with text-to-speech engines.
            2. Be concise and direct. Do not provide tangential information.
            3. All information must be verified and from reliable sources.
        """

    llm = gemini.Realtime(model="gemini-3.1-flash-live-preview")

    return Agent(
        llm = llm,
        agent_user = User(
            id = user_id,
            name = user_name,
            image = user_image
        ),
        edge = getstream.Edge(),
        instructions = final_instructions
    )

async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs):
    await agent.authenticate()
    call = await agent.create_call(call_type, call_id)

    async with agent.join(call):
        await agent.finish()

launcher = AgentLauncher(
    join_call = join_call,
    max_sessions_per_call = 1,
    create_agent = create_agent
)

runner = Runner(launcher)

@runner.fast_api.post("/api/stream/agent/join")
async def join(data: JoinRequestData):
    token = current_call_config.set({
        "id": data.id,
        "name": data.name,
        "image": data.image,
        "instructions": data.instructions
    })

    try:
        await launcher.start_session(call_id = data.call_id, call_type = data.call_type)
    finally:
        current_call_config.reset(token)

if __name__ == "__main__":
    runner.cli()