"""Voice session and duplex WebSocket routes extracted from [mugi_backend._build_app](mugi_backend.py)."""
from __future__ import annotations

import base64
import json
import math
from collections.abc import Awaitable, Callable
from typing import Any

from fastapi import Request as FRequest, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse

from duplex_voice_routes import validate_start_payload
from run_context import sanitize_run_id


def _json_int(value, default: int) -> int:
    try:
        return int(value)
    except Exception:
        return int(default)


def _audio_rms_s16le(payload: bytes) -> float:
    if not payload:
        return 0.0
    sample_count = len(payload) // 2
    if sample_count <= 0:
        return 0.0
    try:
        total = 0.0
        for i in range(0, len(payload) - 1, 2):
            v = int.from_bytes(payload[i : i + 2], "little", signed=True)
            total += float(v * v)
        mean = total / float(sample_count)
        return math.sqrt(mean)
    except Exception:
        return 0.0


def register_voice_duplex_routes(
    fastapi_app: Any,
    *,
    delegate: Any,
    load_app_prefs: Callable[[], dict[str, Any]],
    log: Any,
    duplex_sessions: Any,
    duplex_metrics: Any,
    voice_session_metrics: Any,
    mlx_audio_service: Any,
    auth_websocket: Callable[[WebSocket], Awaitable[bool]],
) -> None:
    """Register /voice/duplex/status, /voice_session/* HTTP, and voice WebSockets."""

    @fastapi_app.get("/voice/duplex/status")
    async def duplex_voice_status():
        return JSONResponse(
            content={
                "ok": True,
                "metrics": dict(duplex_metrics),
                "limits": {
                    "max_frame_bytes": duplex_sessions.max_frame_bytes,
                    "max_session_bytes": duplex_sessions.max_session_bytes,
                    "max_control_events_per_min": duplex_sessions.max_control_events_per_min,
                },
            }
        )

    @fastapi_app.get("/voice_session/status")
    async def voice_session_status():
        prefs = load_app_prefs()
        health = mlx_audio_service.health()
        return JSONResponse(
            content={
                "ok": True,
                "enabled": bool(prefs.get("voice_conversation_mode_enabled", False)),
                "metrics": dict(voice_session_metrics),
                "active_sessions": duplex_sessions.active_count(),
                "limits": {
                    "max_frame_bytes": duplex_sessions.max_frame_bytes,
                    "max_session_bytes": duplex_sessions.max_session_bytes,
                    "max_control_events_per_min": duplex_sessions.max_control_events_per_min,
                    "idle_timeout_s": int(prefs.get("voice_conversation_idle_timeout_s", 900)),
                    "max_session_s": int(prefs.get("voice_conversation_max_session_s", 3600)),
                },
                "service": {
                    "available": bool(health.available),
                    "warmed": bool(health.warmed),
                    "backend": health.backend,
                    "detail": health.detail,
                    "checked_at": health.checked_at,
                },
            }
        )

    @fastapi_app.post("/voice_session/start")
    async def voice_session_start(request: FRequest):
        try:
            body = await request.json()
        except Exception:
            body = {}
        prefs = load_app_prefs()
        if not bool(prefs.get("voice_conversation_mode_enabled", False)):
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "voice_mode_disabled"
            return JSONResponse(content={"ok": False, "error": "voice_mode_disabled"}, status_code=403)
        thread_id = str(body.get("thread_id") or "").strip() or delegate._current_thread_id
        run_id = sanitize_run_id(str(body.get("run_id") or ""))
        sample_rate = _json_int(body.get("sample_rate"), 16000)
        channels = _json_int(body.get("channels"), 1)
        codec = str(body.get("codec") or "pcm_s16le").strip().lower()
        ok_start, err_start = validate_start_payload(sample_rate=sample_rate, channels=channels, codec=codec)
        if not ok_start:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = err_start or "invalid_start"
            return JSONResponse(content={"ok": False, "error": err_start or "invalid_start"}, status_code=400)
        evicted_same_thread = duplex_sessions.end_sessions_for_thread(thread_id)
        if evicted_same_thread > 0:
            voice_session_metrics["evicted_same_thread"] += evicted_same_thread
        health = mlx_audio_service.preload()
        sess = duplex_sessions.start_session(
            thread_id=thread_id,
            sample_rate=sample_rate,
            channels=channels,
            codec=codec,
            run_id=run_id,
        )
        voice_session_metrics["sessions_started"] += 1
        voice_session_metrics["last_transition"] = "started"
        return JSONResponse(
            content={
                "ok": True,
                "session_id": sess.session_id,
                "resume_token": sess.resume_token,
                "thread_id": sess.thread_id,
                "run_id": sess.run_id,
                "service": {
                    "available": bool(health.available),
                    "warmed": bool(health.warmed),
                    "backend": health.backend,
                    "detail": health.detail,
                },
                "evicted_same_thread": evicted_same_thread,
            }
        )

    @fastapi_app.post("/voice_session/interrupt")
    async def voice_session_interrupt(request: FRequest):
        try:
            body = await request.json()
        except Exception:
            body = {}
        thread_id = str(body.get("thread_id") or "").strip() or delegate._current_thread_id
        run_id = sanitize_run_id(str(body.get("run_id") or ""))
        session_id = str(body.get("session_id") or "").strip()
        if not session_id:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "session_id required"
            return JSONResponse(content={"ok": False, "error": "session_id required"}, status_code=400)
        sess = duplex_sessions.get(session_id)
        if sess is None:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "unknown_session"
            return JSONResponse(content={"ok": False, "error": "unknown_session"}, status_code=404)
        if thread_id and sess.thread_id != thread_id:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "session_thread_mismatch"
            return JSONResponse(content={"ok": False, "error": "session_thread_mismatch"}, status_code=409)
        if run_id and sess.run_id and run_id != sess.run_id:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "stale_run_id"
            return JSONResponse(content={"ok": False, "error": "stale_run_id", "active_run_id": sess.run_id}, status_code=409)
        if hasattr(delegate, "onStop_"):
            delegate.onStop_(thread_id=sess.thread_id)
        duplex_sessions.end_session(session_id)
        voice_session_metrics["interruptions"] += 1
        voice_session_metrics["last_transition"] = "interrupted"
        return JSONResponse(content={"ok": True, "session_id": session_id, "thread_id": sess.thread_id, "run_id": sess.run_id})

    @fastapi_app.post("/voice_session/stop")
    async def voice_session_stop(request: FRequest):
        try:
            body = await request.json()
        except Exception:
            body = {}
        session_id = str(body.get("session_id") or "").strip()
        if not session_id:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "session_id required"
            return JSONResponse(content={"ok": False, "error": "session_id required"}, status_code=400)
        sess = duplex_sessions.get(session_id)
        if sess is None:
            voice_session_metrics["errors"] += 1
            voice_session_metrics["last_error"] = "unknown_session"
            return JSONResponse(content={"ok": False, "error": "unknown_session"}, status_code=404)
        duplex_sessions.end_session(session_id)
        voice_session_metrics["sessions_stopped"] += 1
        voice_session_metrics["last_transition"] = "stopped"
        return JSONResponse(content={"ok": True, "session_id": session_id, "thread_id": sess.thread_id, "run_id": sess.run_id})

    @fastapi_app.post("/voice_session/diagnostics")
    async def voice_session_diagnostics(request: FRequest):
        try:
            body = await request.json()
        except Exception:
            body = {}
        session_id = str(body.get("session_id") or "").strip()
        sess = duplex_sessions.get(session_id) if session_id else None
        health = mlx_audio_service.health()
        return JSONResponse(
            content={
                "ok": True,
                "session_id": session_id or None,
                "has_session": bool(sess is not None),
                "state_hint": "active" if sess is not None else "idle",
                "thread_id": (sess.thread_id if sess is not None else ""),
                "run_id": (sess.run_id if sess is not None else ""),
                "service_backend": health.backend,
                "service_available": bool(health.available),
                "service_warmed": bool(health.warmed),
                "session_age_s": (duplex_sessions.session_age_seconds(session_id) if sess is not None and session_id else 0.0),
                "idle_age_s": (duplex_sessions.idle_age_seconds(session_id) if sess is not None and session_id else 0.0),
                "active_sessions": duplex_sessions.active_count(),
                "metrics": dict(voice_session_metrics),
            }
        )

    @fastapi_app.websocket("/voice/stream")
    async def voice_stream_ws(ws: WebSocket):
        log.info("voice_ws_open_attempt path=%s", ws.url.path)
        if not await auth_websocket(ws):
            return
        await ws.accept()
        log.info("voice_ws_accepted path=%s", ws.url.path)
        sess = None
        malformed_strikes = 0
        max_malformed_strikes = 3

        def _dispatch_final_voice_turn(session_obj, text: str) -> None:
            final_text = str(text or "").strip()
            if not final_text:
                return
            try:
                thread_id = str(getattr(session_obj, "thread_id", "") or "")
                if thread_id and thread_id in (getattr(delegate, "_running_thread_ids", set()) or set()):
                    if hasattr(delegate, "onStop_"):
                        delegate.onStop_(thread_id=thread_id)
                if hasattr(delegate, "onSend_text_"):
                    # Voice mode runs against the session's own thread (e.g. __voice_conversation__),
                    # not whichever chat tab the user happens to be on. onSend_text_ reads
                    # delegate._current_thread_id, so we pin it here.
                    prev_thread_id = getattr(delegate, "_current_thread_id", None)
                    if thread_id:
                        delegate._current_thread_id = thread_id
                    try:
                        delegate.onSend_text_(None, final_text)
                    finally:
                        # Restore the chat-tab thread so chat UI keeps working when user exits voice mode.
                        try:
                            delegate._current_thread_id = prev_thread_id
                        except Exception:
                            pass
            except Exception:
                log.debug("voice final turn dispatch failed", exc_info=True)

        try:
            while True:
                raw = await ws.receive_text()
                try:
                    msg = json.loads(raw)
                except Exception:
                    duplex_metrics["errors"] += 1
                    await ws.send_json({"type": "error", "code": "invalid_json", "message": "Invalid JSON payload"})
                    malformed_strikes += 1
                    if malformed_strikes >= max_malformed_strikes:
                        await ws.close(code=1003)
                        break
                    continue
                if not isinstance(msg, dict):
                    duplex_metrics["errors"] += 1
                    await ws.send_json({"type": "error", "code": "invalid_payload", "message": "Object payload required"})
                    malformed_strikes += 1
                    if malformed_strikes >= max_malformed_strikes:
                        await ws.close(code=1003)
                        break
                    continue
                malformed_strikes = 0
                mtype = str(msg.get("type") or "").strip()
                seq = msg.get("seq")

                if mtype == "start":
                    thread_id = str(msg.get("thread_id") or "").strip() or delegate._current_thread_id
                    run_id = sanitize_run_id(str(msg.get("run_id") or ""))
                    sample_rate = _json_int(msg.get("sample_rate"), 16000)
                    channels = _json_int(msg.get("channels"), 1)
                    codec = str(msg.get("codec") or "pcm_s16le").strip().lower()
                    ok_start, err_start = validate_start_payload(sample_rate=sample_rate, channels=channels, codec=codec)
                    if not ok_start:
                        await ws.send_json({"type": "error", "code": err_start or "invalid_start", "message": "invalid start payload"})
                        continue
                    prefs = load_app_prefs()
                    max_session_s = max(300, _json_int(prefs.get("voice_conversation_max_session_s"), 3600))
                    vad_silence_ms = _json_int(msg.get("vad_silence_ms"), _json_int(prefs.get("voice_duplex_vad_silence_ms"), 700))
                    max_utterance_ms = _json_int(msg.get("max_utterance_ms"), _json_int(prefs.get("voice_duplex_max_utterance_ms"), 12000))
                    forced_finalize_ms = _json_int(msg.get("forced_finalize_ms"), _json_int(prefs.get("voice_duplex_forced_finalize_ms"), 1500))
                    punct_ms = _json_int(
                        msg.get("punctuation_stabilization_ms"),
                        _json_int(prefs.get("voice_duplex_punctuation_stabilization_ms"), 350),
                    )

                    resume_session_id = str(msg.get("resume_session_id") or "").strip()
                    resume_token = str(msg.get("resume_token") or "").strip()
                    if resume_session_id and resume_token:
                        resumed = duplex_sessions.resume_session(resume_session_id, resume_token)
                        if resumed is not None:
                            if thread_id and resumed.thread_id != thread_id:
                                await ws.send_json(
                                    {
                                        "type": "error",
                                        "code": "thread_mismatch",
                                        "message": "resume session does not match thread",
                                    }
                                )
                                continue
                            if run_id and resumed.run_id and run_id != resumed.run_id:
                                await ws.send_json(
                                    {
                                        "type": "error",
                                        "code": "run_mismatch",
                                        "message": "resume session does not match run",
                                        "session_id": resumed.session_id,
                                    }
                                )
                                continue
                            sess = resumed
                            duplex_metrics["sessions_resumed"] += 1
                            await ws.send_json(
                                {
                                    "type": "session_state",
                                    "state": "resumed",
                                    "session_id": sess.session_id,
                                    "resume_token": sess.resume_token,
                                    "thread_id": sess.thread_id,
                                    "run_id": sess.run_id,
                                    "seq_ack": seq,
                                }
                            )
                            await ws.send_json({"type": "vad_state", "state": "silence", "session_id": sess.session_id, "seq_ack": seq})
                            continue
                    sess = duplex_sessions.start_session(
                        thread_id=thread_id,
                        sample_rate=sample_rate,
                        channels=channels,
                        codec=codec,
                        run_id=run_id,
                        vad_silence_ms=vad_silence_ms,
                        max_utterance_ms=max_utterance_ms,
                        forced_finalize_ms=forced_finalize_ms,
                        punctuation_stabilization_ms=punct_ms,
                    )
                    duplex_metrics["sessions_started"] += 1
                    await ws.send_json(
                        {
                            "type": "session_state",
                            "state": "started",
                            "session_id": sess.session_id,
                            "resume_token": sess.resume_token,
                            "thread_id": sess.thread_id,
                            "run_id": sess.run_id,
                            "seq_ack": seq,
                        }
                    )
                    try:
                        delegate._structured_bus.publish(
                            sess.thread_id,
                            json.dumps(
                                {
                                    "type": "duplex_agent_state",
                                    "state": "listening",
                                    "thread_id": sess.thread_id,
                                    "run_id": sess.run_id,
                                    "session_id": sess.session_id,
                                }
                            ),
                        )
                    except Exception:
                        pass
                    await ws.send_json({"type": "session_state", "state": "listening", "session_id": sess.session_id, "seq_ack": seq})
                    continue

                if sess is None:
                    await ws.send_json({"type": "error", "code": "session_required", "message": "Send start before other messages"})
                    continue
                if duplex_sessions.session_age_seconds(sess.session_id) > float(max_session_s):
                    voice_session_metrics["errors"] += 1
                    voice_session_metrics["last_error"] = "max_session_exceeded"
                    await ws.send_json(
                        {
                            "type": "error",
                            "code": "max_session_exceeded",
                            "message": "session exceeded configured max duration",
                            "session_id": sess.session_id,
                            "seq_ack": seq,
                        }
                    )
                    await ws.close(code=1008)
                    break
                if not duplex_sessions.validate_and_advance_seq(sess, seq if isinstance(seq, int) else None):
                    duplex_metrics["errors"] += 1
                    await ws.send_json(
                        {
                            "type": "error",
                            "code": "stale_sequence",
                            "message": "sequence must be strictly increasing",
                            "session_id": sess.session_id,
                            "seq_ack": getattr(sess, "last_seq", None),
                        }
                    )
                    continue
                if mtype != "audio_chunk" and not duplex_sessions.count_control_event(sess):
                    duplex_metrics["errors"] += 1
                    await ws.send_json(
                        {
                            "type": "error",
                            "code": "control_rate_limited",
                            "message": "too many control messages",
                            "session_id": sess.session_id,
                            "seq_ack": seq,
                        }
                    )
                    continue

                if mtype == "audio_chunk":
                    if str(msg.get("session_id") or "").strip() != sess.session_id:
                        await ws.send_json({"type": "error", "code": "session_mismatch", "message": "session_id does not match active session", "session_id": sess.session_id})
                        continue
                    msg_thread = str(msg.get("thread_id") or "").strip()
                    if msg_thread and msg_thread != sess.thread_id:
                        await ws.send_json({"type": "error", "code": "thread_mismatch", "message": "thread_id does not match session", "session_id": sess.session_id})
                        continue
                    msg_run = sanitize_run_id(str(msg.get("run_id") or ""))
                    if msg_run and msg_run != (sess.run_id or ""):
                        await ws.send_json({"type": "error", "code": "run_mismatch", "message": "run_id does not match session", "session_id": sess.session_id})
                        continue
                    b64 = str(msg.get("audio_b64") or "")
                    if not b64:
                        await ws.send_json({"type": "error", "code": "missing_audio", "message": "audio_b64 required", "session_id": sess.session_id})
                        continue
                    try:
                        payload = base64.b64decode(b64, validate=True)
                    except Exception:
                        duplex_metrics["errors"] += 1
                        await ws.send_json({"type": "error", "code": "invalid_audio_b64", "message": "audio_b64 must be valid base64", "session_id": sess.session_id})
                        continue
                    ok, err_code = duplex_sessions.add_audio_bytes(sess, len(payload))
                    if not ok:
                        duplex_metrics["errors"] += 1
                        await ws.send_json({"type": "error", "code": err_code or "audio_rejected", "message": "audio frame rejected", "session_id": sess.session_id})
                        continue
                    duplex_metrics["audio_frames"] += 1
                    duplex_metrics["audio_bytes"] += len(payload)
                    speech_start, speech_end = duplex_sessions.mark_audio_activity(
                        sess,
                        rms=_audio_rms_s16le(payload) if sess.codec == "pcm_s16le" else 0.0,
                    )
                    if speech_start:
                        await ws.send_json({"type": "vad_state", "state": "speech_start", "session_id": sess.session_id, "seq_ack": seq})
                    if speech_end:
                        await ws.send_json({"type": "vad_state", "state": "speech_end", "session_id": sess.session_id, "seq_ack": seq})
                    if duplex_sessions.should_finalize_by_timeout(sess):
                        final_text = duplex_sessions.finalize_text(sess)
                        await ws.send_json({"type": "stt_final", "session_id": sess.session_id, "text": final_text, "seq_ack": seq})
                        try:
                            delegate._structured_bus.publish(
                                sess.thread_id,
                                json.dumps(
                                    {
                                        "type": "duplex_user_final",
                                        "text": final_text,
                                        "thread_id": sess.thread_id,
                                        "run_id": sess.run_id,
                                        "session_id": sess.session_id,
                                    }
                                ),
                            )
                        except Exception:
                            pass
                        _dispatch_final_voice_turn(sess, final_text)
                    # Client may provide local transcript chunk text; this route forwards it as partial.
                    dbg = str(msg.get("text_chunk") or "").strip()
                    if dbg:
                        duplex_sessions.append_partial_text(sess, dbg + " ")
                        await ws.send_json({"type": "stt_partial", "session_id": sess.session_id, "text": dbg, "seq_ack": seq})
                        try:
                            delegate._structured_bus.publish(
                                sess.thread_id,
                                json.dumps(
                                    {
                                        "type": "duplex_user_partial",
                                        "text": dbg,
                                        "thread_id": sess.thread_id,
                                        "run_id": sess.run_id,
                                        "session_id": sess.session_id,
                                    }
                                ),
                            )
                        except Exception:
                            pass
                elif mtype == "text_chunk":
                    if str(msg.get("session_id") or "").strip() != sess.session_id:
                        await ws.send_json({"type": "error", "code": "session_mismatch", "message": "session_id does not match active session", "session_id": sess.session_id})
                        continue
                    chunk_text = str(msg.get("text") or msg.get("text_chunk") or "").strip()
                    if not chunk_text:
                        await ws.send_json({"type": "error", "code": "missing_text", "message": "text required", "session_id": sess.session_id})
                        continue
                    duplex_sessions.append_partial_text(sess, chunk_text + " ")
                    await ws.send_json({"type": "stt_partial", "session_id": sess.session_id, "text": chunk_text, "seq_ack": seq})
                    try:
                        delegate._structured_bus.publish(
                            sess.thread_id,
                            json.dumps(
                                {
                                    "type": "duplex_user_partial",
                                    "text": chunk_text,
                                    "thread_id": sess.thread_id,
                                    "run_id": sess.run_id,
                                    "session_id": sess.session_id,
                                }
                            ),
                        )
                    except Exception:
                        pass
                elif mtype == "end_utterance":
                    if str(msg.get("session_id") or "").strip() != sess.session_id:
                        await ws.send_json({"type": "error", "code": "session_mismatch", "message": "session_id does not match active session", "session_id": sess.session_id})
                        continue
                    final_text = duplex_sessions.finalize_text(sess)
                    await ws.send_json({"type": "vad_state", "state": "speech_end", "session_id": sess.session_id, "seq_ack": seq})
                    await ws.send_json({"type": "stt_final", "session_id": sess.session_id, "text": final_text, "seq_ack": seq})
                    await ws.send_json({"type": "session_state", "state": "ended", "session_id": sess.session_id, "seq_ack": seq})
                    try:
                        delegate._structured_bus.publish(
                            sess.thread_id,
                            json.dumps(
                                {
                                    "type": "duplex_user_final",
                                    "text": final_text,
                                    "thread_id": sess.thread_id,
                                    "run_id": sess.run_id,
                                    "session_id": sess.session_id,
                                }
                            ),
                        )
                    except Exception:
                        pass
                    _dispatch_final_voice_turn(sess, final_text)
                elif mtype == "cancel":
                    req_thread = str(msg.get("thread_id") or "").strip() or sess.thread_id
                    req_run = sanitize_run_id(str(msg.get("run_id") or "")) or (sess.run_id or "")
                    if req_thread != sess.thread_id:
                        await ws.send_json({"type": "error", "code": "thread_mismatch", "message": "thread_id does not match session", "session_id": sess.session_id})
                        continue
                    if req_run and sess.run_id and req_run != sess.run_id:
                        await ws.send_json({"type": "error", "code": "run_mismatch", "message": "run_id does not match session", "session_id": sess.session_id})
                        continue
                    await ws.send_json({"type": "session_state", "state": "cancelled", "session_id": sess.session_id, "seq_ack": seq})
                    try:
                        worker = (getattr(delegate, "_workers", {}) or {}).get(sess.thread_id)
                        active_run = sanitize_run_id(str(getattr(worker, "run_id", "") or ""))
                        if req_run and active_run and req_run != active_run:
                            await ws.send_json(
                                {
                                    "type": "error",
                                    "code": "stale_run_id",
                                    "message": "active run differs from requested run",
                                    "active_run_id": active_run,
                                    "session_id": sess.session_id,
                                }
                            )
                        else:
                            if hasattr(delegate, "onStop_"):
                                delegate.onStop_(thread_id=sess.thread_id)
                            try:
                                delegate._structured_bus.publish(
                                    sess.thread_id,
                                    json.dumps(
                                        {
                                            "type": "duplex_interrupted",
                                            "thread_id": sess.thread_id,
                                            "run_id": req_run,
                                            "session_id": sess.session_id,
                                        }
                                    ),
                                )
                            except Exception:
                                pass
                    except Exception:
                        log.debug("duplex cancel dispatch failed", exc_info=True)
                    # Keep the websocket voice session alive after barge-in cancel so the
                    # next utterance can continue in the dedicated Voice pane.
                    await ws.send_json({"type": "session_state", "state": "listening", "session_id": sess.session_id, "seq_ack": seq})
                else:
                    await ws.send_json({"type": "error", "code": "unknown_type", "message": f"unknown type: {mtype}", "session_id": sess.session_id, "seq_ack": seq})
        except WebSocketDisconnect:
            pass
        except Exception:
            duplex_metrics["errors"] += 1
            log.exception("duplex voice websocket failed")
        finally:
            if sess is not None:
                duplex_sessions.end_session(sess.session_id)

    @fastapi_app.websocket("/voice_session/stream")
    async def voice_session_stream_ws(ws: WebSocket):
        log.info("voice_session_ws_open_attempt path=%s", ws.url.path)
        # `/voice_session/start` already enforces the feature flag before session
        # creation. Avoid an extra pre-accept close here, which can surface as
        # an opaque `close_code_0` on the macOS client during handshake.
        await voice_stream_ws(ws)
