Sanad_lite/gemini/client.py

"""Gemini WebSocket client for real-time voice interaction.

Provides:
  - Bidirectional audio streaming (mic → Gemini → speaker)
  - Text-to-speech via typed input
  - Voice-command detection through transcription parsing
  - System instruction injection for persona control
"""

from __future__ import annotations

import asyncio
import base64
import inspect
import json
import urllib.error
import urllib.request
from typing import Any

import websockets

from Project.Sanad.config import (
    GEMINI_API_KEY,
    GEMINI_MODEL,
    GEMINI_VOICE,
    GEMINI_WS_TIMEOUT,
    GEMINI_WS_URI,
)
from Project.Sanad.core.config_loader import section as _cfg_section
from Project.Sanad.core.event_bus import bus
from Project.Sanad.core.logger import get_logger

log = get_logger("gemini_client")

_GC = _cfg_section("gemini", "client")
# Default system prompt — SINGLE SOURCE in core.gemini_defaults
_DEFAULT_SYSTEM_PROMPT = _cfg_section("core", "gemini_defaults").get(
    "default_system_prompt",
    "You are Sanad (Bousandah), a wise and friendly Emirati assistant. "
    "Speak in UAE dialect (Khaleeji). Be helpful and concise."
)

# TTS / typed-replay system prompt. The voice_client speaks TYPED text, so it
# must read the text VERBATIM in its OWN language — NOT answer it and NOT force
# Khaleeji. This is what makes the Live native-audio model return AUDIO instead
# of "thinking" text. Copied from SanadR1/Sanadv3 so the lite dashboard speaks
# with the exact same engine + behavior as the robots.
TTS_SYSTEM_PROMPT = _cfg_section("core", "gemini_defaults").get(
    "tts_system_prompt",
    "You are a pure multilingual text-to-speech voice. The instant the user "
    "sends text, speak it aloud word for word in the SAME language it is "
    "written in, then stop. Output ONLY that spoken audio — no thinking, no "
    "commentary, no acknowledgements, no headers, no explanations, no "
    "greetings, no extra words. Never translate and never change the language: "
    "English stays English, Arabic stays Arabic, Urdu stays Urdu, Indonesian "
    "stays Indonesian. Your speech must be identical to the user's text, "
    "nothing more and nothing less."
)
# Per-voice system-prompt overrides, keyed by prebuilt voice name. Empty by
# default, and INTENTIONALLY so for Charon (Unitree G1) and Puck (Unitree R1):
# those two must keep speaking with the robots' own verbatim TTS prompt, since
# sounding identical to Sanadv3/SanadR1 is the whole point. Only voices with no
# robot to match — Agibot x2 / Kore — get a different instruction here.
# Configured in core_config.json → gemini_defaults.voice_system_prompts.
_VOICE_SYSTEM_PROMPTS = _cfg_section("core", "gemini_defaults").get(
    "voice_system_prompts", {}) or {}

_RECV_TIMEOUT_SEC = _GC.get("recv_timeout_sec", 30)
_RECONNECT_MAX_ATTEMPTS = _GC.get("reconnect_max_attempts", 3)
_RECONNECT_INITIAL_DELAY_SEC = _GC.get("reconnect_initial_delay_sec", 1.0)
_RECONNECT_MAX_DELAY_SEC = _GC.get("reconnect_max_delay_sec", 10.0)

# Dedicated text-to-speech model (stateless REST generateContent). Far more
# reliable than the Live native-audio model for pure TTS: the Live model often
# returns its "thinking" reasoning text instead of speech for short prompts,
# whereas this returns audio-only. Same prebuilt voices (Charon/Puck/Kore/...).
GEMINI_TTS_MODEL = _GC.get("tts_model", "gemini-2.5-flash-preview-tts")
_GEMINI_REST_BASE = _GC.get(
    "rest_base", "https://generativelanguage.googleapis.com/v1beta")
# Spoken-directive preamble so the TTS model READS the text aloud rather than
# answering it (bare short inputs like "مرحبا" otherwise 400 "Model tried to
# generate text"), and nudges an Emirati/Gulf dialect. The directive is NOT
# spoken (verified: audio length ~unchanged with/without it). Retries cover the
# model's occasional empty response. All config-overridable via gemini_config.json.
_TTS_PREAMBLE = _GC.get("tts_preamble", "بِاللهجة الإماراتية: ")
_TTS_MAX_ATTEMPTS = int(_GC.get("tts_max_attempts", 3))
# BCP-47 language/accent code for the TTS speechConfig. "ar-AE" = UAE Arabic —
# the structured, reliable lever for an Emirati accent. "" omits the field.
_TTS_LANGUAGE_CODE = _GC.get("tts_language_code", "ar-AE")
# Sampling temperature. 0.0 makes the TTS DETERMINISTIC — the same text renders
# the same audio every time (stable tone). The default (~1.0) re-performs with
# random prosody on every call, which is why the tone kept drifting between
# generations. The retry temperature is used only if temp-0 returns an empty
# part for a given input (so we can still get audio for it).
_TTS_TEMPERATURE = float(_GC.get("tts_temperature", 0.0))
_TTS_RETRY_TEMPERATURE = float(_GC.get("tts_retry_temperature", 0.6))
# Per-attempt HTTP timeout for the TTS call. A normal TTS response is ~2s; a
# stuck call would otherwise hang the whole generation and hold the typed-replay
# single-flight lock (blocking the UI), so keep this tight.
_TTS_TIMEOUT = float(_GC.get("tts_timeout_sec", 20))
# Ask the Live session to transcribe its own spoken output. Costs nothing
# extra and gives typed_replay a way to verify the whole text was actually
# read aloud. Set false in gemini_config.json to go back to blind reads.
_OUTPUT_TRANSCRIPTION = bool(_GC.get("output_transcription", True))
# How long to keep reading after `generationComplete` while waiting for
# `turnComplete`. Only the tail of an already-generated turn arrives in this
# window, so it is short.
_POST_GENERATION_GRACE_SEC = float(_GC.get("post_generation_grace_sec", 3.0))


class GeminiQuotaExhausted(RuntimeError):
    """The API key itself is out of credits / over quota.

    Distinct from every other failure because retrying is pointless: the Live
    socket, the reconnect chain and the REST TTS fallback all fail the same
    way. Without this, one Generate & Play burns 4 Live attempts (each with a
    reconnect chain) plus 3 REST retries before answering — long enough to
    blow the Apache proxy timeout, so the user sees a bare 503 and the cron
    health-check restarts a perfectly healthy app.
    """


# Substrings that mean "this key cannot make calls right now". Deliberately
# narrow: a transient 429 rate-limit IS worth retrying, an empty wallet is not.
_QUOTA_MARKERS = (
    "credits are depleted",
    "prepayment credits",
    "billing",
    "exceeded your current quota",
)


def _quota_message(exc: BaseException) -> str:
    """Return a clean reason string if `exc` is a credits/quota failure, else ""."""
    text = f"{exc}".lower()
    if not any(m in text for m in _QUOTA_MARKERS):
        return ""
    if "credits are depleted" in text or "prepayment credits" in text:
        return ("Gemini API credits are depleted — top up billing in AI Studio "
                "for this project's key. No voice can be generated until then.")
    return ("Gemini API quota/billing error — this API key cannot generate "
            "audio right now.")


def _tts_rest_blocking(url: str, body: dict, timeout: float) -> bytes:
    """Blocking POST to the TTS generateContent endpoint → PCM bytes.

    Runs in a worker thread (via asyncio.to_thread) so the event loop is never
    blocked. Returns the first inline audio part decoded from base64; raises
    RuntimeError carrying the API's message on an HTTP/transport error.
    """
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        url, data=data,
        headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        detail = ""
        try:
            detail = json.loads(exc.read().decode("utf-8")).get("error", {}).get("message", "")
        except Exception:
            pass
        raise RuntimeError(f"Gemini TTS HTTP {exc.code}: {detail or exc.reason}")
    except urllib.error.URLError as exc:
        raise RuntimeError(f"Gemini TTS request failed: {exc.reason}")
    payload = json.loads(raw)
    if isinstance(payload, dict) and payload.get("error"):
        msg = payload["error"].get("message", payload["error"])
        raise RuntimeError(f"Gemini TTS error: {msg}")
    for cand in payload.get("candidates", []):
        for part in cand.get("content", {}).get("parts", []):
            inline = part.get("inlineData") or part.get("inline_data")
            if inline and inline.get("data"):
                return base64.b64decode(inline["data"])
    return b""


class GeminiVoiceClient:
    """Manages one WebSocket session to the Gemini Bidi audio API.

    Concurrency model:
      - `_send_lock` serializes ALL websocket writes.
      - `_session_lock` ensures only one consumer (live loop OR typed replay)
        owns the receive stream at a time. Acquired by send_text and
        receive_stream context managers.
      - `_owner` records who currently holds the session lock for diagnostics.
    """

    def __init__(self, system_prompt: str = ""):
        self.system_prompt = system_prompt or _DEFAULT_SYSTEM_PROMPT
        self._ws: Any = None
        self._connected = False
        self._send_lock = asyncio.Lock()
        self._session_lock = asyncio.Lock()
        self._connect_lock = asyncio.Lock()  # serializes reconnect attempts
        self._owner: str | None = None
        self._reconnect_attempts = 0
        # Transcript of the audio returned by the most recent send_text(),
        # when output transcription is enabled. Written under the session
        # lock, so it always belongs to the call that just returned.
        self.last_output_transcript = ""

    @property
    def connected(self) -> bool:
        return self._connected

    @property
    def session_owner(self) -> str | None:
        return self._owner

    def _ws_kwargs(self) -> dict[str, Any]:
        kwargs: dict[str, Any] = {"max_size": None, "open_timeout": 30}
        try:
            sig = inspect.signature(websockets.connect)
            key = "extra_headers" if "extra_headers" in sig.parameters else "additional_headers"
        except Exception:
            key = "extra_headers"
        kwargs[key] = {"Content-Type": "application/json"}
        return kwargs

    async def _safe_close_ws(self) -> None:
        """Best-effort close of the current socket, guarded by a timeout.

        A half-dead socket whose close() hangs must never wedge the event
        loop here, so the close itself is bounded.
        """
        ws, self._ws = self._ws, None
        if ws is None:
            return
        try:
            await asyncio.wait_for(ws.close(), timeout=5)
        except Exception:
            pass

    async def connect(self):
        uri = f"{GEMINI_WS_URI}?key={GEMINI_API_KEY}"
        # The system instruction is baked into this handshake alongside the
        # voice, so per-voice prompts can only be applied here. Voices with no
        # override (Charon/Puck) keep self.system_prompt untouched.
        system_prompt = _VOICE_SYSTEM_PROMPTS.get(GEMINI_VOICE) or self.system_prompt
        try:
            self._ws = await websockets.connect(uri, **self._ws_kwargs())
            setup = {
                "setup": {
                    "model": GEMINI_MODEL,
                    "generationConfig": {
                        "responseModalities": ["AUDIO"],
                        "speechConfig": {
                            "voiceConfig": {
                                "prebuiltVoiceConfig": {"voiceName": GEMINI_VOICE}
                            }
                        },
                    },
                    "systemInstruction": {"parts": [{"text": system_prompt}]},
                }
            }
            if _OUTPUT_TRANSCRIPTION:
                # Ask the server to transcribe the audio it actually speaks.
                # This is the only reliable way to tell a COMPLETE read from a
                # cut-off one: the native-audio model regularly speaks the
                # first few words, stops, and emits reasoning text instead of
                # the rest — and the audio it returns looks perfectly healthy.
                # typed_replay compares this transcript against the requested
                # text and retries when words are missing.
                setup["setup"]["outputAudioTranscription"] = {}
            # Guard the app-level setup handshake with a timeout. websockets'
            # own open_timeout only covers the HTTP upgrade, NOT this send/ACK.
            # A socket that opens but never ACKs would otherwise block this
            # await forever and freeze uvicorn's single event loop — every HTTP
            # request with it. That is exactly what took the site down.
            await asyncio.wait_for(
                self._ws.send(json.dumps(setup)), timeout=GEMINI_WS_TIMEOUT)
            await asyncio.wait_for(self._ws.recv(), timeout=GEMINI_WS_TIMEOUT)  # ACK
            self._connected = True
            self._reconnect_attempts = 0
            log.info("Connected to Gemini (%s, voice=%s%s)", GEMINI_MODEL, GEMINI_VOICE,
                     ", voice-specific prompt" if GEMINI_VOICE in _VOICE_SYSTEM_PROMPTS else "")
            await bus.emit("voice.connected")
        except asyncio.TimeoutError:
            self._connected = False
            await self._safe_close_ws()
            log.warning("Gemini setup handshake timed out after %ss", GEMINI_WS_TIMEOUT)
            raise
        except Exception as exc:
            self._connected = False
            await self._safe_close_ws()
            reason = _quota_message(exc)
            if reason:
                # Out of credits — say so once, plainly, and stop. Retrying
                # only stacks up latency until the proxy times out.
                log.error("Gemini refused the connection: %s", reason)
                raise GeminiQuotaExhausted(reason) from exc
            log.exception("Failed to connect to Gemini")
            raise

    async def disconnect(self):
        await self._safe_close_ws()
        self._connected = False
        self._owner = None
        log.info("Disconnected from Gemini")
        await bus.emit("voice.disconnected")

    async def _ensure_connected(self):
        """Reconnect if dropped, with bounded retries.

        Serialized via _connect_lock so concurrent callers don't trigger
        duplicate handshakes.
        """
        # Fast path — no lock needed
        if self._connected and self._ws is not None:
            return True

        async with self._connect_lock:
            # Re-check inside the lock (another coroutine may have just connected)
            if self._connected and self._ws is not None:
                return True

            max_attempts = _RECONNECT_MAX_ATTEMPTS
            delay = _RECONNECT_INITIAL_DELAY_SEC
            for attempt in range(max_attempts):
                try:
                    log.warning("Reconnecting to Gemini (attempt %d/%d)", attempt + 1, max_attempts)
                    await self.connect()
                    return True
                except GeminiQuotaExhausted:
                    # No amount of reconnecting refills the account — surface it
                    # to the caller immediately instead of sleeping through the
                    # whole backoff chain on every attempt.
                    raise
                except Exception:
                    self._reconnect_attempts += 1
                    await asyncio.sleep(delay)
                    delay = min(delay * 2, _RECONNECT_MAX_DELAY_SEC)
            log.error("Reconnect failed after %d attempts", max_attempts)
            await bus.emit("voice.error", reason="reconnect_failed")
            return False

    async def send_audio_chunk(self, pcm_b64: str) -> bool:
        """Send a base64-encoded PCM audio chunk (mic input).

        Returns False on failure so the caller can react instead of silently
        no-op'ing forever (the original bug).
        """
        if not self._connected or self._ws is None:
            return False
        msg = {
            "realtimeInput": {
                "mediaChunks": [
                    {"mimeType": "audio/pcm;rate=16000", "data": pcm_b64}
                ]
            }
        }
        try:
            async with self._send_lock:
                await asyncio.wait_for(
                    self._ws.send(json.dumps(msg)), timeout=GEMINI_WS_TIMEOUT)
            return True
        except websockets.exceptions.ConnectionClosed:
            log.warning("send_audio_chunk: connection closed")
            self._connected = False
            await bus.emit("voice.error", reason="connection_closed")
            return False
        except asyncio.TimeoutError:
            log.warning("send_audio_chunk: send timed out after %ss", GEMINI_WS_TIMEOUT)
            self._connected = False
            await bus.emit("voice.error", reason="send_timeout")
            return False
        except Exception:
            log.exception("send_audio_chunk failed")
            return False

    async def send_text(self, text: str, owner: str = "send_text") -> tuple[bytes, list[str]]:
        """Send text, receive audio response. Returns (audio_bytes, text_parts).

        Acquires the session lock for the entire request/response cycle so
        no other consumer can steal frames from the receive side.
        If the connection drops mid-request, reconnects once and retries.
        """
        if not await self._ensure_connected():
            raise RuntimeError("Not connected to Gemini and reconnect failed.")

        async with self._session_lock:
            self._owner = owner
            try:
                return await self._send_text_inner(text)
            except (websockets.exceptions.ConnectionClosed, asyncio.TimeoutError) as exc:
                # A 1011 close carrying a billing message is the server telling
                # us the key is empty — the close reason is the only place that
                # information appears, so read it before treating this as a
                # routine drop and reconnecting into the same wall.
                reason = _quota_message(exc)
                if reason:
                    self._connected = False
                    log.error("Gemini closed the session: %s", reason)
                    raise GeminiQuotaExhausted(reason) from exc
                log.warning("send_text: connection died/stalled on send — reconnecting once")
                self._connected = False
                if not await self._ensure_connected():
                    raise RuntimeError("Reconnect after send failure also failed.")
                return await self._send_text_inner(text)
            finally:
                self._owner = None

    async def _drain_socket(self) -> int:
        """Discard frames left over from an earlier turn. Returns how many.

        If a previous turn ended without consuming everything the server sent,
        those frames sit in the socket and the NEXT send_text() reads them as
        its own reply — you get the tail of the last sentence instead of the
        new one, arriving implausibly fast. Clearing them first makes every
        turn start from a known-empty stream.
        """
        dropped = 0
        while dropped < 500:
            try:
                await asyncio.wait_for(self._ws.recv(), timeout=0.01)
            except (asyncio.TimeoutError, asyncio.CancelledError):
                break
            except Exception:
                break
            dropped += 1
        if dropped:
            log.warning("drained %d stale frame(s) from the previous turn", dropped)
        return dropped

    async def _send_text_inner(self, text: str) -> tuple[bytes, list[str]]:
        """Inner send/receive loop — caller must hold _session_lock."""
        request = {
            "client_content": {
                "turns": [{"role": "user", "parts": [{"text": text}]}],
                "turn_complete": True,
            }
        }
        async with self._send_lock:
            await self._drain_socket()
            await asyncio.wait_for(
                self._ws.send(json.dumps(request)), timeout=GEMINI_WS_TIMEOUT)

        audio_chunks: list[bytes] = []
        text_parts: list[str] = []
        transcript_parts: list[str] = []
        self.last_output_transcript = ""

        # `generationComplete` means the model stopped GENERATING — the server
        # still has audio and transcript to deliver, and only `turnComplete`
        # ends the turn. Breaking on the former truncated the tail of every
        # sentence AND left those frames in the socket for the next turn to
        # mis-read. Wait for turnComplete, with a short grace period after
        # generationComplete so a turn that never sends it can't stall us.
        gen_done = False
        while True:
            timeout = _POST_GENERATION_GRACE_SEC if gen_done else GEMINI_WS_TIMEOUT
            try:
                raw = await asyncio.wait_for(self._ws.recv(), timeout=timeout)
            except asyncio.TimeoutError:
                if gen_done:
                    break          # tail delivered, server just never closed the turn
                log.warning("send_text: recv timed out")
                break
            except websockets.exceptions.ConnectionClosed:
                log.warning("send_text: connection closed mid-stream")
                self._connected = False
                break

            try:
                resp = json.loads(raw)
            except json.JSONDecodeError:
                log.warning("send_text: bad JSON from server")
                continue

            if "error" in resp:
                log.error("Gemini error: %s", resp["error"])
                await bus.emit("voice.error", reason=str(resp["error"]))
                break

            sc = resp.get("serverContent", {})
            mt = sc.get("modelTurn", {})
            for part in mt.get("parts", []):
                inline = part.get("inlineData")
                if inline and inline.get("data"):
                    audio_chunks.append(base64.b64decode(inline["data"]))
                tp = part.get("text")
                if isinstance(tp, str) and tp.strip():
                    text_parts.append(tp.strip())

            input_tr = sc.get("inputTranscription", {})
            if input_tr.get("text"):
                await bus.emit("voice.user_said", text=input_tr["text"])

            # Transcript of the audio the model is speaking, streamed in
            # fragments alongside it. Accumulated verbatim; the caller decides
            # whether it covers the requested text.
            out_tr = sc.get("outputTranscription", {})
            if out_tr.get("text"):
                transcript_parts.append(out_tr["text"])

            if sc.get("turnComplete"):
                break
            if sc.get("generationComplete"):
                gen_done = True

        audio_bytes = b"".join(audio_chunks)
        self.last_output_transcript = "".join(transcript_parts).strip()
        if audio_bytes:
            await bus.emit("voice.gemini_spoke", audio_len=len(audio_bytes))
        return audio_bytes, text_parts

    def acquire_session(self, owner: str) -> "_SessionGuard":
        """Return an async context manager for exclusive session ownership.

        Use as `async with client.acquire_session("live_voice"):`.
        While held, no other consumer may call send_text or receive_stream.
        """
        return _SessionGuard(self, owner)

    async def receive_stream(self):
        """Yield server events. Caller MUST hold the session lock."""
        if self._owner is None:
            raise RuntimeError(
                "receive_stream requires session lock — use acquire_session() first"
            )
        if not self._connected or self._ws is None:
            return
        try:
            async for raw in self._ws:
                try:
                    resp = json.loads(raw)
                except json.JSONDecodeError:
                    continue
                yield resp.get("serverContent", {})
        except websockets.exceptions.ConnectionClosed:
            log.warning("receive_stream: connection closed")
            self._connected = False
            await bus.emit("voice.error", reason="connection_closed")

    async def raw_send(self, payload: dict):
        """Low-level send for the live loop. Always use through send lock."""
        if not self._connected or self._ws is None:
            return False
        try:
            async with self._send_lock:
                await asyncio.wait_for(
                    self._ws.send(json.dumps(payload)), timeout=GEMINI_WS_TIMEOUT)
            return True
        except Exception:
            log.exception("raw_send failed")
            return False

    async def synthesize_tts(self, text: str, voice: str | None = None) -> bytes:
        """Text-to-speech via the dedicated Gemini TTS model (stateless REST).

        Returns raw PCM bytes (24 kHz, 16-bit, mono). Reliable audio-only
        output — unlike the Live native-audio model, which frequently returns
        its reasoning text instead of speech for short prompts. Voice defaults
        to the current GEMINI_VOICE (hot-swappable from the dashboard). Runs the
        blocking HTTP call in a worker thread so the event loop stays free.
        """
        text = (text or "").strip()
        if not text:
            return b""
        if not GEMINI_API_KEY:
            raise RuntimeError("No Gemini API key configured.")
        voice = voice or GEMINI_VOICE
        url = (f"{_GEMINI_REST_BASE}/models/{GEMINI_TTS_MODEL}"
               f":generateContent?key={GEMINI_API_KEY}")
        prompt = f"{_TTS_PREAMBLE}{text}" if _TTS_PREAMBLE else text
        speech_config = {
            "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice}}
        }
        if _TTS_LANGUAGE_CODE:
            speech_config["languageCode"] = _TTS_LANGUAGE_CODE
        body = {
            "contents": [{"parts": [{"text": prompt}]}],
            "generationConfig": {
                "responseModalities": ["AUDIO"],
                "speechConfig": speech_config,
            },
        }
        # Deterministic first (temperature 0) so the same text renders the same
        # audio → stable tone across generations. Only bump the temperature on a
        # retry if temp-0 returned an empty part for this input (some very short
        # inputs need it); those retries are the only ones whose tone can vary.
        temps = [_TTS_TEMPERATURE] + [_TTS_RETRY_TEMPERATURE] * max(0, _TTS_MAX_ATTEMPTS - 1)
        last_exc = None
        for attempt, temp in enumerate(temps, 1):
            body["generationConfig"]["temperature"] = temp
            try:
                audio = await asyncio.to_thread(
                    _tts_rest_blocking, url, body, _TTS_TIMEOUT)
            except Exception as exc:
                reason = _quota_message(exc)
                if reason:
                    log.error("Gemini REST TTS refused: %s", reason)
                    raise GeminiQuotaExhausted(reason) from exc
                last_exc = exc
                log.warning("TTS attempt %d/%d (temp=%.1f) failed: %s",
                            attempt, len(temps), temp, exc)
                continue
            if audio:
                if attempt > 1:
                    log.info("TTS succeeded on attempt %d (temp=%.1f)", attempt, temp)
                return audio
            log.warning("TTS attempt %d/%d (temp=%.1f) returned no audio",
                        attempt, len(temps), temp)
        if last_exc is not None:
            raise last_exc
        return b""

    def status(self) -> dict[str, Any]:
        return {
            "connected": self._connected,
            "model": GEMINI_MODEL,
            "voice": GEMINI_VOICE,
            "session_owner": self._owner,
            "reconnect_attempts": self._reconnect_attempts,
        }


class _SessionGuard:
    """Async context manager for exclusive session ownership.

    Always releases owner + lock on exit, even on exceptions.
    """

    def __init__(self, client: GeminiVoiceClient, owner: str):
        self._client = client
        self._owner = owner
        self._held = False

    async def __aenter__(self):
        await self._client._session_lock.acquire()
        self._held = True
        self._client._owner = self._owner
        return self._client

    async def __aexit__(self, exc_type, exc, tb):
        try:
            self._client._owner = None
        finally:
            if self._held:
                self._client._session_lock.release()
                self._held = False
        return False  # don't suppress exceptions