from __future__ import annotations

import asyncio
import contextlib
import hashlib
import logging
import os
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable

from services.ai_voice_runtime_service.ack_bank import PrebakedAckBank
from services.ai_voice_runtime_service.audiosocket import (
    AUDIO_SOCKET_PACKET_DTMF,
    AUDIO_SOCKET_PACKET_HANGUP,
    AUDIO_SOCKET_PACKET_PCM16,
    AUDIO_SOCKET_PACKET_UUID,
    EnergyVAD,
    chunk_audio,
    encode_audio_packet,
    normalize_media_uuid,
    pcm16le_to_wav_bytes,
    read_packet,
    resample_pcm16le_stateful,
)
from services.ai_voice_runtime_service.providers.asr import (
    ASRTranscription,
    ASRProvider,
    StreamingASRPartial,
    StreamingASRProvider,
    StreamingASRUnavailable,
)
from services.shared.audioop_compat import audioop
from services.ai_voice_runtime_service.providers.tts import TTSProvider
from services.shared.models import VoiceAITurnDecisionOut


logger = logging.getLogger("uvicorn.error")


@dataclass(slots=True)
class MediaRegistration:
    voice_session_id: str
    call_id: str
    interaction_id: str
    ai_session_id: str | None = None
    language: str | None = None
    media_uuid: str | None = None
    queue_code: str | None = None
    queue_id: str | None = None
    agent_profile: str | None = None
    voice_v2_enabled: bool = False
    voice_v2_ack_mode: str = "disabled"
    voice_v2_streaming_tts: bool = False
    voice_v2_partial_asr: bool = False
    voice_v2_duplex: bool = False
    voice_v2_streaming_asr_backend: str | None = None
    voice_v2_prebaked_ack: bool = False
    voice_v2_emotive_ack: bool = False
    voice_v2_emotive_ack_ru_only: bool = True


@dataclass(slots=True)
class MediaActor:
    registration: MediaRegistration
    reader: asyncio.StreamReader
    writer: asyncio.StreamWriter
    vad: EnergyVAD
    frame_ms: int
    frame_bytes: int
    turn_queue: asyncio.Queue[tuple[bytes, bool]] = field(default_factory=asyncio.Queue)
    worker_task: asyncio.Task | None = None
    keepalive_task: asyncio.Task | None = None
    handoff_task: asyncio.Task | None = None
    playback_interrupt: asyncio.Event = field(default_factory=asyncio.Event)
    write_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    closed: bool = False
    state: str = "greeting"
    barge_in_pending: bool = False
    last_media_touch_monotonic: float = 0.0
    last_outbound_audio_monotonic: float = 0.0
    first_pcm_logged: bool = False
    keepalive_loop_logged: bool = False
    input_active: bool = False
    playback_active: bool = False
    early_ack_started: bool = False
    partial_transcript: str | None = None
    stable_partial_transcript: str | None = None
    last_partial_transcript: str | None = None
    partial_transcript_streak: int = 0
    partial_intent: str | None = None
    partial_intent_streak: int = 0
    stable_partial_intent: str | None = None
    early_plan_task: asyncio.Task | None = None
    early_plan_transcript: str | None = None
    early_plan_intent: str | None = None
    early_plan_generation: int = 0
    response_plan_id: str | None = None
    playback_generation: int = 0
    tts_generation: int = 0
    utterance_generation: int = 0
    finalized_utterance_generation: int = 0
    partial_asr_task: asyncio.Task | None = None
    streaming_asr_push_task: asyncio.Task | None = None
    streaming_asr_push_queue: asyncio.Queue[bytes | None] | None = None
    partial_asr_attempted: bool = False
    asr_stream_id: str | None = None
    asr_streaming_enabled: bool = False
    asr_streaming_failed: bool = False
    asr_poll_due_monotonic: float = 0.0
    streaming_asr_backoff_until_monotonic: float = 0.0
    barge_in_speech_ms: int = 0
    barge_in_detected_monotonic: float = 0.0
    speech_started_monotonic: float = 0.0
    speech_ended_monotonic: float = 0.0
    thinking_continuation_pcm: bytearray = field(default_factory=bytearray)
    thinking_continuation_deadline_monotonic: float = 0.0
    last_ack_text: str | None = None
    last_ack_completed_monotonic: float = 0.0
    last_ack_variant: str | None = None
    current_reply_phase: str | None = None
    finalized_caller_turn_count: int = 0
    listening_since_monotonic: float = 0.0
    reprompt_attempts: int = 0
    consecutive_low_signal_turns: int = 0


@dataclass(slots=True)
class _FramePacer:
    """Wall-clock-anchored pacing for one playback: paces frames to the first
    frame's send time instead of naive per-frame relative sleeps, whose drift
    accumulates under event-loop scheduling pressure (audible as stutter)."""

    frame_seconds: float
    start_monotonic: float | None = None
    frames_written: int = 0

    async def wait_for_next_frame(self) -> None:
        if self.start_monotonic is None:
            self.start_monotonic = time.monotonic()
        self.frames_written += 1
        target_monotonic = self.start_monotonic + self.frames_written * self.frame_seconds
        sleep_seconds = target_monotonic - time.monotonic()
        if sleep_seconds > 0:
            await asyncio.sleep(sleep_seconds)


class AudioSocketMediaRuntime:
    def __init__(
        self,
        *,
        enabled: bool,
        host: str,
        port: int,
        frame_ms: int,
        idle_timeout_seconds: float,
        registration_wait_timeout_seconds: float,
        min_speech_ms: int,
        trailing_silence_ms: int,
        max_turn_ms: int,
        vad_rms_threshold: int = 250,
        asr_provider: ASRProvider,
        streaming_asr_provider: StreamingASRProvider | None = None,
        tts_provider: TTSProvider,
        load_registration_by_media_uuid: Callable[[str], MediaRegistration | None],
        mark_media_connected: Callable[[str, str], None],
        mark_media_ended: Callable[[str, str], None],
        touch_media_frame: Callable[[str], None],
        set_state: Callable[[str, str, str | None, dict[str, Any] | None], None],
        get_pending_greeting: Callable[[str], str | None],
        mark_reply_started: Callable[[str, str, bool, str | None], None] | None = None,
        mark_reply_delivered: Callable[[str, str, bool, str | None], None],
        mark_reply_discarded: Callable[[str, str, str | None, str], None] | None = None,
        plan_reply: Callable[[str, str, dict[str, Any] | None, str], None],
        record_latency: Callable[[str, str, int], None] | None = None,
        process_turn: Callable[[str, str, str | None, bool, dict[str, Any] | None], VoiceAITurnDecisionOut],
        request_handoff: Callable[[str, str, VoiceAITurnDecisionOut], None],
        handle_media_error: Callable[[str, str, dict[str, Any] | None], None],
    ) -> None:
        self._enabled = enabled
        self._host = host
        self._port = port
        self._frame_ms = max(frame_ms, 20)
        self._frame_bytes = int((8000 * self._frame_ms / 1000.0) * 2)
        self._idle_timeout_seconds = max(idle_timeout_seconds, 5.0)
        self._registration_wait_timeout_seconds = max(registration_wait_timeout_seconds, 0.0)
        self._outbound_keepalive_interval_seconds = 0.75
        self._min_speech_ms = max(min_speech_ms, self._frame_ms)
        self._trailing_silence_ms = max(trailing_silence_ms, self._frame_ms)
        self._max_turn_ms = max(max_turn_ms, self._frame_ms)
        self._vad_rms_threshold = max(int(vad_rms_threshold), 1)
        self._asr_provider = asr_provider
        self._streaming_asr_provider = streaming_asr_provider or StreamingASRProvider()
        self._tts_provider = tts_provider
        self._ack_bank = PrebakedAckBank(tts_provider=tts_provider)
        self._load_registration_by_media_uuid = load_registration_by_media_uuid
        self._mark_media_connected = mark_media_connected
        self._mark_media_ended = mark_media_ended
        self._touch_media_frame = touch_media_frame
        self._set_state = set_state
        self._get_pending_greeting = get_pending_greeting
        self._mark_reply_started = mark_reply_started or (lambda session_id, text, is_greeting, phase: None)
        self._mark_reply_delivered = mark_reply_delivered
        self._mark_reply_discarded = mark_reply_discarded or (lambda session_id, text, phase, status: None)
        self._plan_reply = plan_reply
        self._record_latency = record_latency or (lambda session_id, metric, latency_ms: None)
        self._process_turn = process_turn
        self._request_handoff = request_handoff
        self._handle_media_error = handle_media_error
        self._server: asyncio.base_events.Server | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._actors: dict[str, MediaActor] = {}
        self._v2_ack_wait_seconds = 0.18
        self._partial_asr_min_ms = 320
        self._immediate_ack_min_ms = 700
        self._v2_ack_post_gap_seconds = 0.10
        self._v1_ack_wait_seconds = 0.6
        self._partial_poll_interval_seconds = 0.20
        # Small startup cushion for streamed TTS playback: absorb ElevenLabs
        # network delivery jitter before we start pacing frames out to the
        # caller, so a brief mid-download stall doesn't read as a dead-air gap.
        self._tts_stream_prebuffer_ms = 200
        self._streaming_asr_partial_poll_enabled = (
            str(os.getenv("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "0")).strip().lower()
            in {"1", "true", "yes", "on"}
        )
        self._stable_partial_hold_seconds = 0.40
        self._barge_in_trigger_ms = 220
        self._streaming_asr_reopen_backoff_seconds = 2.0
        self._streaming_asr_push_queue_max_frames = 250
        self._streaming_asr_push_batch_max_bytes = self._frame_bytes * 8
        self._streaming_asr_push_drain_timeout_seconds = 1.5
        self._thinking_continuation_grace_seconds = 0.45
        self._thinking_continuation_max_bytes = int(1800 * 16)
        self._partial_first_final_enabled = (
            str(os.getenv("AI_VOICE_V2_PARTIAL_FIRST_FINAL", "1")).strip().lower()
            in {"1", "true", "yes", "on"}
        )
        self._early_plan_enabled = (
            str(os.getenv("AI_VOICE_V2_EARLY_PLAN_ENABLED", "1")).strip().lower()
            in {"1", "true", "yes", "on"}
        )
        self._no_speech_reprompt_enabled = (
            str(os.getenv("AI_VOICE_NO_SPEECH_REPROMPT_ENABLED", "1")).strip().lower()
            in {"1", "true", "yes", "on"}
        )
        self._no_speech_reprompt_seconds = 7.0
        self._no_speech_reprompt_low_signal_turns = 2
        self._no_speech_reprompt_max_attempts = 2
        raw_early_plan_intents = str(
            os.getenv(
                "AI_VOICE_V2_EARLY_PLAN_INTENTS",
                "schedule,address,price,status,problem,operator_request",
            )
        )
        self._early_plan_intents = {
            item.strip().lower()
            for item in raw_early_plan_intents.split(",")
            if item.strip()
        }

    @staticmethod
    def _normalize_intent_text(text: str) -> str:
        compact = str(text or "").strip().lower()
        for char in ",.!?;:…\"'()[]{}":
            compact = compact.replace(char, " ")
        return " ".join(compact.split())

    _FINAL_LOW_SIGNAL_PHRASES = {
        "алло",
        "ой",
        "слышу",
        "слышно",
        "твой",
        "поргай что это",
    }

    @classmethod
    def _is_low_signal_partial_transcript(cls, text: str | None) -> bool:
        """Broad filler check used only for mid-utterance ack/early-plan gating."""
        normalized = cls._normalize_intent_text(text)
        if not normalized:
            return True
        return normalized in cls._FINAL_LOW_SIGNAL_PHRASES | {
            "ага",
            "да",
            "добрый день",
            "здравствуйте",
            "ладно",
            "неа",
            "нет",
            "ок",
            "понял",
            "поняла",
            "привет",
            "угу",
            "хорошо",
            "ясно",
            "давай",
        }

    @classmethod
    def _is_low_signal_final_transcript(cls, text: str | None) -> bool:
        """Narrow filler check for a *finalized* caller turn.

        Unlike the partial-transcript check, this must not swallow real answers
        such as "да"/"нет"/"хорошо" — those are legitimate replies and a caller
        who says only that deserves a response, not silence.
        """
        normalized = cls._normalize_intent_text(text)
        if not normalized:
            return True
        return normalized in cls._FINAL_LOW_SIGNAL_PHRASES

    def _detect_early_intent(self, text: str) -> str:
        normalized = self._normalize_intent_text(text)
        if not normalized:
            return "unknown"
        if any(token in normalized for token in ("оператор", "оператором", "человеком", "менеджер", "сотрудник")):
            return "operator_request"
        if any(token in normalized for token in ("график", "распис", "время работы", "work schedule", "жұмыс")):
            return "schedule"
        if any(token in normalized for token in ("адрес", "филиал", "офис", "где вы", "мекен", "қайда")):
            return "address"
        if any(token in normalized for token in ("тариф", "цена", "стоимость", "сколько стоит", "баға")):
            return "price"
        if any(token in normalized for token in ("статус", "заявк", "заказ", "обращени", "өтінім")):
            return "status"
        if any(token in normalized for token in ("не работает", "ошибка", "проблем", "сломал", "істемей")):
            return "problem"
        return "unknown"

    @staticmethod
    def _asr_failure_category(exc: Exception) -> str:
        text = str(exc or "").lower()
        if "unauthorized" in text or "unknown api key" in text or "invalid api key" in text:
            return "auth"
        if "timeout" in text:
            return "timeout"
        if "not found" in text:
            return "provider_not_found"
        if "service unavailable" in text or "temporarily unavailable" in text:
            return "provider_unavailable"
        return "provider_error"

    @staticmethod
    def _ack_kind_for_intent(intent: str) -> str:
        if intent == "operator_request":
            return "handoff"
        if intent in {"schedule", "address", "price", "status", "problem"}:
            return "understanding"
        if intent == "unknown":
            return "unknown"
        return "clarify"

    @staticmethod
    def _ack_text(language: str | None, ack_kind: str) -> str:
        normalized = str(language or "").strip().lower()
        if normalized == "kz":
            if ack_kind == "handoff":
                return "Бір сәт."
            if ack_kind == "understanding":
                return "Қазір айтып шығамын."
            if ack_kind == "clarify":
                return "Қазір нақтылайын."
            return "Қазір айтайын."
        if ack_kind == "handoff":
            return "Секунду."
        if ack_kind == "understanding":
            return "Сейчас сориентирую."
        if ack_kind == "clarify":
            return "Сейчас уточню."
        return "Секунду."

    @staticmethod
    def _no_speech_reprompt_text(language: str | None) -> str:
        normalized = str(language or "").strip().lower()
        if normalized == "kz":
            return "Алло, сізді тыңдап тұрмын. Немен көмектесе аламын?"
        return "Алло, я вас слушаю. Подскажите, чем могу помочь?"

    @staticmethod
    def _technical_issue_text(language: str | None) -> str:
        normalized = str(language or "").strip().lower()
        if normalized == "kz":
            return "Техникалық ақау шықты. Оператормен қосамын."
        return "Возникла техническая проблема со связью. Соединяю с оператором."

    @staticmethod
    def _should_use_emotive_ack(registration: MediaRegistration, language: str | None) -> bool:
        if not registration.voice_v2_emotive_ack:
            return False
        if not registration.voice_v2_emotive_ack_ru_only:
            return True
        normalized = str(language or registration.language or "").strip().lower()
        return normalized.startswith("ru")

    def _should_emit_partial_ack(self, actor: MediaActor, partial_transcript: str, partial_intent: str) -> bool:
        transcript_text = str(partial_transcript or "").strip()
        if not transcript_text or self._is_low_signal_partial_transcript(transcript_text):
            return False
        normalized_intent = str(partial_intent or "").strip() or "unknown"
        if actor.finalized_caller_turn_count > 0:
            return True
        return normalized_intent != "unknown"

    def _should_emit_blind_ack(self, actor: MediaActor, pcm_bytes: bytes) -> bool:
        del actor
        return len(pcm_bytes) >= self._immediate_ack_min_bytes

    @staticmethod
    def _base_ack_variants(language: str | None, ack_kind: str) -> tuple[str, ...]:
        normalized = str(language or "").strip().lower()
        if normalized == "kz":
            if ack_kind == "handoff":
                return (
                    "Жарайды, қосамын.",
                    "Түсінікті, бір сәт.",
                    "Қазір ауыстырамын.",
                    "Жақсы, операторға қосамын.",
                )
            if ack_kind == "understanding":
                return (
                    "Түсінікті, қазір айтамын.",
                    "Жарайды, қазір көмектесемін.",
                    "Ия, қазір қараймын.",
                    "Жақсы, қазір айтып беремін.",
                )
            if ack_kind == "clarify":
                return (
                    "Жарайды, қазір нақтылаймын.",
                    "Түсінікті, бір сәт.",
                    "Ия, қазір анықтаймын.",
                )
            return (
                "Жарайды.",
                "Түсінікті.",
                "Бір сәт.",
                "Қазір.",
                "Ия, тыңдап тұрмын.",
            )
        if ack_kind == "handoff":
            return (
                "Угу, секунду.",
                "Понял вас, соединяю.",
                "Мхм, соединяю.",
                "Хорошо, сейчас соединю.",
                "Да, сейчас соединю.",
                "Одну минуту, переключаю.",
            )
        if ack_kind == "understanding":
            return (
                "Угу, сейчас подскажу.",
                "Мхм, сориентирую.",
                "Ага, понял вас.",
                "Хм, сейчас уточню.",
                "Хорошо, сейчас подскажу.",
                "Ясно, сейчас разберусь.",
                "Так, минутку.",
                "Сейчас разберусь.",
            )
        if ack_kind == "clarify":
            return (
                "Угу, сейчас уточню.",
                "Хм, дайте посмотрю.",
                "Хорошо, сейчас уточню.",
                "Ага, сейчас сориентирую.",
                "Так, уточню детали.",
                "Дайте уточню.",
            )
        return (
            "Ага.",
            "Ясно, минутку.",
            "Угу.",
            "Хорошо, сейчас.",
            "Так, слушаю.",
            "Момент.",
            "Понял вас.",
            "Хм, сейчас гляну.",
        )

    def _select_ack_payload(
        self,
        actor: MediaActor,
        *,
        language: str | None,
        ack_kind: str,
    ) -> tuple[str, dict[str, object] | None, str]:
        effective_language = language or actor.registration.language
        emotive_ack = self._should_use_emotive_ack(actor.registration, effective_language)
        if emotive_ack:
            variants = self._base_ack_variants(effective_language, ack_kind)
        else:
            variants = (self._ack_text(effective_language, ack_kind),)
        seed = (
            f"{actor.registration.voice_session_id}:"
            f"{actor.response_plan_id or actor.utterance_generation}:"
            f"{effective_language or ''}:"
            f"{ack_kind}"
        ).encode("utf-8")
        index = int.from_bytes(hashlib.sha256(seed).digest()[:4], "big") % len(variants)
        ack_text = variants[index]
        if actor.last_ack_text == ack_text and len(variants) > 1:
            index = (index + 1) % len(variants)
            ack_text = variants[index]
        actor.last_ack_text = ack_text
        actor.last_ack_variant = f"{ack_kind}:{index}"
        style_hints: dict[str, object] | None = None
        if emotive_ack:
            style_hints = {"role": "good"}
        return ack_text, style_hints, actor.last_ack_variant

    def _should_use_voice_v2(self, registration: MediaRegistration) -> bool:
        return bool(registration.voice_v2_enabled and str(registration.voice_v2_ack_mode or "").strip() == "immediate_short")

    @property
    def _partial_asr_min_bytes(self) -> int:
        return self._partial_asr_min_ms * 16

    @property
    def _immediate_ack_min_bytes(self) -> int:
        return self._immediate_ack_min_ms * 16

    @property
    def _tts_stream_prebuffer_bytes(self) -> int:
        return self._tts_stream_prebuffer_ms * 16

    @staticmethod
    def _reset_live_turn_state(actor: MediaActor) -> None:
        actor.utterance_generation += 1
        actor.finalized_utterance_generation = 0
        actor.input_active = True
        actor.early_ack_started = False
        actor.partial_transcript = None
        actor.stable_partial_transcript = None
        actor.last_partial_transcript = None
        actor.partial_transcript_streak = 0
        actor.partial_intent = None
        actor.partial_intent_streak = 0
        actor.stable_partial_intent = None
        actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
        actor.early_plan_transcript = None
        actor.early_plan_intent = None
        actor.early_plan_generation = 0
        early_plan_task = actor.early_plan_task
        actor.early_plan_task = None
        if early_plan_task is not None and not early_plan_task.done():
            early_plan_task.cancel()
        actor.partial_asr_attempted = False
        actor.asr_streaming_failed = False
        actor.asr_poll_due_monotonic = 0.0
        actor.speech_started_monotonic = time.monotonic()
        actor.speech_ended_monotonic = 0.0
        actor.barge_in_speech_ms = 0
        actor.barge_in_detected_monotonic = 0.0
        actor.current_reply_phase = None
        actor.thinking_continuation_pcm.clear()
        actor.thinking_continuation_deadline_monotonic = 0.0
        partial_task = actor.partial_asr_task
        actor.partial_asr_task = None
        if partial_task is not None and not partial_task.done():
            partial_task.cancel()

    @staticmethod
    def _is_streaming_v2_session(registration: MediaRegistration) -> bool:
        return bool(
            registration.voice_v2_enabled
            and registration.voice_v2_duplex
            and registration.voice_v2_partial_asr
            and registration.voice_v2_streaming_asr_backend
        )

    async def _ensure_streaming_asr(self, actor: MediaActor) -> None:
        if actor.closed or actor.asr_streaming_enabled:
            return
        if not self._is_streaming_v2_session(actor.registration):
            return
        if time.monotonic() < actor.streaming_asr_backoff_until_monotonic:
            return
        try:
            stream_id = await asyncio.to_thread(
                self._streaming_asr_provider.open_stream,
                actor.registration.voice_session_id,
                language_hint=actor.registration.language,
            )
        except StreamingASRUnavailable as exc:
            actor.asr_streaming_enabled = False
            logger.warning(
                "audiosocket.streaming_asr_unavailable session_id=%s backend=%s error=%s",
                actor.registration.voice_session_id,
                actor.registration.voice_v2_streaming_asr_backend,
                str(exc)[:500],
            )
            return
        actor.asr_stream_id = stream_id
        actor.asr_streaming_enabled = True
        actor.asr_streaming_failed = False
        actor.asr_poll_due_monotonic = 0.0
        actor.streaming_asr_push_queue = asyncio.Queue(maxsize=self._streaming_asr_push_queue_max_frames)
        actor.streaming_asr_push_task = asyncio.create_task(
            self._streaming_asr_push_loop(actor, stream_id),
            name=f"streaming-asr-push-{actor.registration.voice_session_id}",
        )

    def _mark_streaming_asr_backoff(self, actor: MediaActor) -> None:
        actor.streaming_asr_backoff_until_monotonic = (
            time.monotonic() + self._streaming_asr_reopen_backoff_seconds
        )

    async def _close_streaming_asr(self, actor: MediaActor, *, drain: bool = True) -> None:
        stream_id = actor.asr_stream_id
        actor.asr_stream_id = None
        actor.asr_streaming_enabled = False
        await self._stop_streaming_asr_push_loop(actor, drain=drain)
        if not stream_id:
            return
        await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id)

    async def _streaming_asr_push_loop(self, actor: MediaActor, stream_id: str) -> None:
        queue = actor.streaming_asr_push_queue
        if queue is None:
            return
        while True:
            item = await queue.get()
            if item is None:
                queue.task_done()
                return

            batch = bytearray(item)
            task_done_count = 1
            stop_after_batch = False
            while len(batch) < self._streaming_asr_push_batch_max_bytes:
                try:
                    extra = queue.get_nowait()
                except asyncio.QueueEmpty:
                    break
                if extra is None:
                    stop_after_batch = True
                    queue.task_done()
                    break
                batch.extend(extra)
                task_done_count += 1

            try:
                await asyncio.to_thread(
                    self._streaming_asr_provider.push_pcm,
                    stream_id,
                    bytes(batch),
                )
            except StreamingASRUnavailable as exc:
                actor.asr_streaming_failed = True
                actor.asr_streaming_enabled = False
                self._mark_streaming_asr_backoff(actor)
                logger.warning(
                    "audiosocket.streaming_asr_push_failed session_id=%s error=%s",
                    actor.registration.voice_session_id,
                    str(exc)[:500],
                )
                return
            finally:
                for _ in range(task_done_count):
                    queue.task_done()
            if stop_after_batch:
                return

    async def _stop_streaming_asr_push_loop(self, actor: MediaActor, *, drain: bool) -> None:
        task = actor.streaming_asr_push_task
        queue = actor.streaming_asr_push_queue
        actor.streaming_asr_push_task = None
        actor.streaming_asr_push_queue = None
        if task is None:
            return
        if task.done():
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await task
            return
        if queue is not None:
            if drain:
                try:
                    await asyncio.wait_for(
                        queue.join(),
                        timeout=self._streaming_asr_push_drain_timeout_seconds,
                    )
                except asyncio.TimeoutError:
                    actor.asr_streaming_failed = True
                    actor.asr_streaming_enabled = False
                    self._mark_streaming_asr_backoff(actor)
                    logger.warning(
                        "audiosocket.streaming_asr_push_drain_timeout session_id=%s queued_frames=%s",
                        actor.registration.voice_session_id,
                        queue.qsize(),
                    )
            sentinel_enqueued = False
            with contextlib.suppress(asyncio.QueueFull):
                queue.put_nowait(None)
                sentinel_enqueued = True
            if not sentinel_enqueued:
                task.cancel()
        try:
            await asyncio.wait_for(task, timeout=0.5)
        except asyncio.TimeoutError:
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await task

    def _queue_streaming_asr_pcm(self, actor: MediaActor, pcm_frame: bytes) -> None:
        queue = actor.streaming_asr_push_queue
        if queue is None or actor.asr_streaming_failed:
            return
        try:
            queue.put_nowait(pcm_frame)
        except asyncio.QueueFull:
            actor.asr_streaming_failed = True
            actor.asr_streaming_enabled = False
            self._mark_streaming_asr_backoff(actor)
            logger.warning(
                "audiosocket.streaming_asr_push_queue_full session_id=%s max_frames=%s",
                actor.registration.voice_session_id,
                self._streaming_asr_push_queue_max_frames,
            )
            asyncio.create_task(self._close_streaming_asr(actor, drain=False))

    def _buffer_thinking_continuation(self, actor: MediaActor, pcm_frame: bytes, *, is_speech: bool, now: float) -> None:
        if not is_speech and not actor.thinking_continuation_pcm:
            return
        if (
            not is_speech
            and actor.thinking_continuation_deadline_monotonic > 0
            and now > actor.thinking_continuation_deadline_monotonic
        ):
            return
        remaining = self._thinking_continuation_max_bytes - len(actor.thinking_continuation_pcm)
        if remaining <= 0:
            if is_speech:
                actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds
            return
        actor.thinking_continuation_pcm.extend(pcm_frame[:remaining])
        if is_speech:
            actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds

    async def _extend_with_thinking_continuation(self, actor: MediaActor, pcm_bytes: bytes) -> bytes:
        deadline = time.monotonic() + self._thinking_continuation_grace_seconds
        observed_size = len(actor.thinking_continuation_pcm)
        while time.monotonic() < deadline:
            await asyncio.sleep(0.05)
            if actor.closed:
                break
            if actor.thinking_continuation_pcm:
                observed_size = len(actor.thinking_continuation_pcm)
                deadline = max(deadline, actor.thinking_continuation_deadline_monotonic)
                continue
            if observed_size > 0:
                break
        if not actor.thinking_continuation_pcm:
            actor.thinking_continuation_deadline_monotonic = 0.0
            return pcm_bytes
        merged = pcm_bytes + bytes(actor.thinking_continuation_pcm)
        actor.thinking_continuation_pcm.clear()
        actor.thinking_continuation_deadline_monotonic = 0.0
        return merged

    def _update_stable_partial_intent(self, actor: MediaActor, intent: str) -> None:
        normalized = str(intent or "").strip() or "unknown"
        if actor.partial_intent == normalized:
            actor.partial_intent_streak += 1
        else:
            actor.partial_intent = normalized
            actor.partial_intent_streak = 1
        if actor.partial_intent_streak >= 2:
            actor.stable_partial_intent = normalized

    @staticmethod
    def _update_stable_partial_transcript(actor: MediaActor, transcript_text: str, *, provider_stable: bool) -> None:
        normalized = " ".join(str(transcript_text or "").strip().lower().split())
        if not normalized:
            return
        if actor.last_partial_transcript == normalized:
            actor.partial_transcript_streak += 1
        else:
            actor.last_partial_transcript = normalized
            actor.partial_transcript_streak = 1
        if provider_stable or actor.partial_transcript_streak >= 2:
            actor.stable_partial_transcript = transcript_text

    def _should_schedule_early_plan(self, actor: MediaActor, transcript_text: str, intent: str) -> bool:
        if not self._early_plan_enabled or not self._should_use_voice_v2(actor.registration):
            return False
        if actor.closed or self._is_low_signal_partial_transcript(transcript_text):
            return False
        normalized_intent = str(intent or "").strip().lower() or "unknown"
        if normalized_intent not in self._early_plan_intents:
            return False
        task = actor.early_plan_task
        if task is not None and task.done() and actor.early_plan_transcript == transcript_text and actor.early_plan_intent == normalized_intent:
            return False
        return task is None or task.done()

    def _schedule_early_plan(
        self,
        actor: MediaActor,
        *,
        transcript_text: str,
        language: str | None,
        intent: str,
        source: str,
    ) -> None:
        if not self._should_schedule_early_plan(actor, transcript_text, intent):
            return
        if not actor.response_plan_id:
            actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
        actor.early_plan_generation = actor.utterance_generation
        actor.early_plan_transcript = transcript_text
        actor.early_plan_intent = str(intent or "").strip() or "unknown"
        metadata = {
            "voice_v2_enabled": actor.registration.voice_v2_enabled,
            "reply_phase": "early_plan",
            "response_plan_id": actor.response_plan_id,
            "playback_generation": actor.playback_generation + 1,
            "tts_generation": actor.tts_generation + 1,
            "partial_transcript": transcript_text,
            "early_intent": actor.early_plan_intent,
            "transcript_source": source,
            "suppress_name_prefix": True,
        }
        actor.early_plan_task = asyncio.create_task(
            asyncio.to_thread(
                self._process_turn,
                actor.registration.voice_session_id,
                transcript_text,
                language or actor.registration.language,
                False,
                metadata,
            ),
            name=f"voice-early-plan-{actor.registration.voice_session_id}",
        )
        logger.info(
            "audiosocket.early_plan_started session_id=%s generation=%s intent=%s source=%s text_len=%s",
            actor.registration.voice_session_id,
            actor.early_plan_generation,
            actor.early_plan_intent,
            source,
            len(transcript_text),
        )

    async def _take_early_plan_decision(
        self,
        actor: MediaActor,
        *,
        timeout_seconds: float,
    ) -> VoiceAITurnDecisionOut | None:
        task = actor.early_plan_task
        if task is None:
            return None
        if actor.early_plan_generation != actor.utterance_generation:
            return None
        if not task.done() and timeout_seconds > 0:
            with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
                await asyncio.wait_for(asyncio.shield(task), timeout=timeout_seconds)
        if not task.done():
            return None
        try:
            decision = task.result()
        except Exception as exc:
            logger.warning(
                "audiosocket.early_plan_failed session_id=%s generation=%s error=%s",
                actor.registration.voice_session_id,
                actor.early_plan_generation,
                str(exc)[:500],
            )
            return None
        if not getattr(decision, "reply_text", None) and not getattr(decision, "needs_handoff", False):
            return None
        return decision

    async def _reconcile_final_decision_after_early_plan(
        self,
        actor: MediaActor,
        *,
        decision_task: asyncio.Task,
        early_decision: VoiceAITurnDecisionOut,
        transcript_text: str,
    ) -> None:
        try:
            final_decision = await decision_task
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.warning(
                "audiosocket.final_plan_reconcile_failed session_id=%s error=%s",
                actor.registration.voice_session_id,
                str(exc)[:500],
            )
            return
        if actor.closed:
            return
        if final_decision.needs_handoff and not early_decision.needs_handoff:
            await self._set_actor_state(actor, "handoff_requested", final_decision.handoff_reason)
            self._start_handoff_request(actor, transcript_text, final_decision)

    async def _poll_streaming_partial(self, actor: MediaActor) -> None:
        if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id:
            return
        now = time.monotonic()
        if now < actor.asr_poll_due_monotonic:
            return
        actor.asr_poll_due_monotonic = now + self._partial_poll_interval_seconds
        partial = await asyncio.to_thread(self._streaming_asr_provider.poll_partial, actor.asr_stream_id)
        if partial is None:
            return
        transcript_text = str(partial.text or "").strip()
        if not transcript_text:
            return
        if transcript_text != actor.partial_transcript:
            logger.info(
                "audiosocket.streaming_asr_partial session_id=%s stable=%s final=%s text=%s",
                actor.registration.voice_session_id,
                bool(partial.is_stable),
                bool(partial.is_final),
                transcript_text[:160],
            )
        actor.partial_transcript = transcript_text
        self._update_stable_partial_transcript(actor, transcript_text, provider_stable=bool(partial.is_stable or partial.is_final))
        intent = self._detect_early_intent(transcript_text)
        self._update_stable_partial_intent(actor, intent)
        stable_text = str(actor.stable_partial_transcript or "").strip()
        stable_intent = str(actor.stable_partial_intent or "").strip()
        if stable_text and stable_intent:
            self._schedule_early_plan(
                actor,
                transcript_text=stable_text,
                language=partial.language or actor.registration.language,
                intent=stable_intent,
                source="streaming_stable_partial",
            )

    async def _run_streaming_partial_poll(self, actor: MediaActor, utterance_generation: int) -> None:
        try:
            if utterance_generation != actor.utterance_generation:
                return
            await self._poll_streaming_partial(actor)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.warning(
                "audiosocket.streaming_asr_partial_failed session_id=%s generation=%s error=%s",
                actor.registration.voice_session_id,
                utterance_generation,
                str(exc)[:500],
            )

    def _maybe_schedule_streaming_partial_poll(self, actor: MediaActor) -> None:
        if not self._streaming_asr_partial_poll_enabled:
            return
        if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id:
            return
        if time.monotonic() < actor.asr_poll_due_monotonic:
            return
        task = actor.partial_asr_task
        if task is not None and not task.done():
            return
        actor.partial_asr_task = asyncio.create_task(
            self._run_streaming_partial_poll(actor, actor.utterance_generation),
            name=f"streaming-asr-partial-{actor.registration.voice_session_id}",
        )

    async def _finalize_streaming_transcription(self, actor: MediaActor) -> ASRTranscription:
        if not actor.asr_streaming_enabled or not actor.asr_stream_id:
            raise StreamingASRUnavailable("Streaming ASR stream is not active")
        stream_id = actor.asr_stream_id
        start_monotonic = time.monotonic()
        await self._stop_streaming_asr_push_loop(actor, drain=True)
        if actor.asr_streaming_failed:
            raise StreamingASRUnavailable("Streaming ASR push failed before finalize")
        try:
            transcription = await asyncio.to_thread(self._streaming_asr_provider.finalize, stream_id)
        finally:
            await self._close_streaming_asr(actor)
        logger.info(
            "audiosocket.streaming_asr_finalized session_id=%s elapsed_ms=%s text=%s",
            actor.registration.voice_session_id,
            int((time.monotonic() - start_monotonic) * 1000),
            str(transcription.text or "")[:160],
        )
        return transcription

    async def _record_reply_status(
        self,
        actor: MediaActor,
        *,
        text: str,
        is_greeting: bool,
        phase: str | None,
        status: str,
    ) -> None:
        if status == "started":
            await asyncio.to_thread(self._invoke_reply_status_callback, self._mark_reply_started, actor.registration.voice_session_id, text, is_greeting, phase)
            return
        if status == "delivered":
            await asyncio.to_thread(self._invoke_reply_status_callback, self._mark_reply_delivered, actor.registration.voice_session_id, text, is_greeting, phase)
            return
        if status in {"interrupted", "discarded"}:
            await asyncio.to_thread(
                self._invoke_discard_callback,
                actor.registration.voice_session_id,
                text,
                phase,
                status,
            )

    async def _record_latency_metric(self, actor: MediaActor, metric: str, start_monotonic: float) -> None:
        if start_monotonic <= 0:
            return
        latency_ms = int(max((time.monotonic() - start_monotonic) * 1000.0, 0.0))
        await asyncio.to_thread(
            self._record_latency,
            actor.registration.voice_session_id,
            metric,
            latency_ms,
        )

    @staticmethod
    def _invoke_reply_status_callback(callback, session_id: str, text: str, is_greeting: bool, phase: str | None) -> None:
        try:
            callback(session_id, text, is_greeting, phase)
        except TypeError:
            callback(session_id, text, is_greeting)

    def _invoke_discard_callback(self, session_id: str, text: str, phase: str | None, status: str) -> None:
        try:
            self._mark_reply_discarded(session_id, text, phase, status)
        except TypeError:
            return

    async def _run_partial_asr_probe(
        self,
        actor: MediaActor,
        *,
        utterance_generation: int,
        pcm_bytes: bytes,
    ) -> None:
        try:
            wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
            transcription = await asyncio.to_thread(
                lambda: self._asr_provider.transcribe_partial(
                    wav_bytes,
                    language_hint=actor.registration.language,
                )
            )
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.warning(
                "audiosocket.partial_asr_failed session_id=%s generation=%s error=%s",
                actor.registration.voice_session_id,
                utterance_generation,
                str(exc)[:500],
            )
            return

        if actor.closed:
            return
        if utterance_generation != actor.utterance_generation:
            return
        if actor.finalized_utterance_generation >= utterance_generation:
            return

        transcript_text = str(transcription.text or "").strip()
        if not transcript_text:
            return
        actor.partial_transcript = transcript_text
        self._update_stable_partial_transcript(actor, transcript_text, provider_stable=True)
        actor.partial_intent = self._detect_early_intent(transcript_text)
        actor.stable_partial_intent = actor.partial_intent
        self._schedule_early_plan(
            actor,
            transcript_text=transcript_text,
            language=transcription.language or actor.registration.language,
            intent=actor.partial_intent,
            source="partial_asr_probe",
        )
        logger.info(
            "audiosocket.partial_asr_ready session_id=%s generation=%s text_len=%s intent=%s",
            actor.registration.voice_session_id,
            utterance_generation,
            len(transcript_text),
            actor.partial_intent,
        )

    def _maybe_schedule_partial_asr(self, actor: MediaActor) -> None:
        if actor.closed or not actor.registration.voice_v2_partial_asr:
            return
        if actor.partial_asr_attempted:
            return
        if not actor.vad.is_active:
            return
        snapshot = actor.vad.snapshot_utterance_pcm()
        if len(snapshot) < self._partial_asr_min_bytes:
            return
        actor.partial_asr_attempted = True
        utterance_generation = actor.utterance_generation
        actor.partial_asr_task = asyncio.create_task(
            self._run_partial_asr_probe(
                actor,
                utterance_generation=utterance_generation,
                pcm_bytes=snapshot,
            )
        )

    async def _emit_early_ack(
        self,
        actor: MediaActor,
        *,
        language: str | None,
        metadata: dict[str, Any],
        ack_source: str,
        ack_kind: str | None = None,
    ) -> None:
        if actor.closed or actor.early_ack_started:
            return
        ack_kind = ack_kind or self._ack_kind_for_intent(actor.partial_intent or "unknown")
        ack_text, style_hints, ack_variant = self._select_ack_payload(
            actor,
            language=language,
            ack_kind=ack_kind,
        )
        actor.early_ack_started = True
        actor.current_reply_phase = "ack"
        await self._plan_reply_segment(
            actor,
            ack_text,
            kind="ack",
            metadata={
                **metadata,
                "partial_transcript": actor.partial_transcript,
                "early_intent": metadata.get("early_intent") or actor.partial_intent,
                "ack_kind": ack_kind,
                "ack_variant": ack_variant,
                "voice_style": "emotive_ack" if style_hints else "neutral_ack",
                "phase": "ack",
                "partial_ack_source": ack_source,
            },
        )
        if actor.registration.voice_v2_prebaked_ack and str(language or actor.registration.language or "").strip().lower().startswith("ru"):
            await self._play_prebaked_ack(actor, ack_text, style_hints=style_hints)
        else:
            await self._speak_reply(actor, ack_text, is_greeting=False, style_hints=style_hints, reply_phase="ack")
        actor.last_ack_completed_monotonic = time.monotonic()
        if not actor.closed:
            await self._set_actor_state(actor, "thinking")

    async def _plan_reply_segment(
        self,
        actor: MediaActor,
        text: str,
        *,
        kind: str,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if actor.closed or not str(text or "").strip():
            return
        await asyncio.to_thread(
            self._plan_reply,
            actor.registration.voice_session_id,
            text,
            metadata or {},
            kind,
        )

    async def _play_pcm_payload(
        self,
        actor: MediaActor,
        *,
        text: str,
        pcm_8k: bytes,
        is_greeting: bool,
        reply_phase: str | None,
    ) -> None:
        if actor.closed or not text or not pcm_8k:
            return
        actor.playback_interrupt.clear()
        actor.barge_in_detected_monotonic = 0.0
        actor.barge_in_speech_ms = 0
        await self._set_actor_state(actor, "speaking")
        actor.current_reply_phase = reply_phase
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="started",
        )
        if reply_phase == "ack":
            await self._record_latency_metric(actor, "speech_end_to_ack_start", actor.speech_ended_monotonic)
            await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic)
        elif reply_phase == "main":
            await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic)
        pacer = _FramePacer(frame_seconds=actor.frame_ms / 1000.0)
        interrupted = await self._write_paced_pcm_frames(actor, pcm_8k, pacer)
        if actor.playback_interrupt.is_set():
            interrupted = True
        actor.playback_interrupt.clear()
        actor.current_reply_phase = None
        if interrupted or actor.closed:
            await self._record_latency_metric(actor, "barge_in_detected_to_playback_stopped", actor.barge_in_detected_monotonic)
            await self._record_reply_status(
                actor,
                text=text,
                is_greeting=is_greeting,
                phase=reply_phase,
                status="interrupted" if interrupted else "discarded",
            )
            return
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="delivered",
        )
        logger.info(
            "audiosocket.reply_delivered session_id=%s greeting=%s phase=%s",
            actor.registration.voice_session_id,
            is_greeting,
            reply_phase,
        )

    async def _play_prebaked_ack(
        self,
        actor: MediaActor,
        text: str,
        *,
        style_hints: dict[str, object] | None = None,
    ) -> None:
        clip = await asyncio.to_thread(
            self._ack_bank.get_clip,
            text=text,
            language=actor.registration.language,
            style_hints=style_hints,
        )
        await self._play_pcm_payload(
            actor,
            text=text,
            pcm_8k=clip.pcm_8k_bytes,
            is_greeting=False,
            reply_phase="ack",
        )

    async def _speak_reply(
        self,
        actor: MediaActor,
        text: str,
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
        reply_phase: str | None = "main",
    ) -> None:
        try:
            await self._speak_text(
                actor,
                text,
                is_greeting=is_greeting,
                style_hints=style_hints,
                reply_phase=reply_phase,
            )
        except TypeError:
            await self._speak_text(
                actor,
                text,
                is_greeting=is_greeting,
                style_hints=style_hints,
            )

    @property
    def address(self) -> str:
        return f"{self._host}:{self._port}"

    async def start(self) -> None:
        if not self._enabled or self._server is not None:
            return
        self._loop = asyncio.get_running_loop()
        self._server = await asyncio.start_server(self._handle_connection, self._host, self._port)

    async def _await_registration(
        self,
        media_uuid: str,
        *,
        writer: asyncio.StreamWriter | None = None,
        frame_bytes: int | None = None,
    ) -> MediaRegistration | None:
        deadline = time.monotonic() + self._registration_wait_timeout_seconds
        last_keepalive_at = 0.0
        keepalive_payload = b"\x00" * max(frame_bytes or 0, 0)
        keepalive_logged = False
        while True:
            registration = await asyncio.to_thread(self._load_registration_by_media_uuid, media_uuid)
            if registration is not None:
                return registration
            if time.monotonic() >= deadline:
                logger.warning("audiosocket.registration_timeout media_uuid=%s", media_uuid)
                return None
            if writer is not None and keepalive_payload:
                now = time.monotonic()
                if now - last_keepalive_at >= self._outbound_keepalive_interval_seconds:
                    writer.write(encode_audio_packet(keepalive_payload))
                    await writer.drain()
                    if not keepalive_logged:
                        keepalive_logged = True
                        logger.info(
                            "audiosocket.prereg_keepalive media_uuid=%s frame_bytes=%s",
                            media_uuid,
                            len(keepalive_payload),
                        )
                    last_keepalive_at = now
            await asyncio.sleep(0.05)

    async def stop(self) -> None:
        actor_ids = list(self._actors.keys())
        for session_id in actor_ids:
            await self._close_session(session_id, reason="runtime_shutdown")
        if self._server is not None:
            self._server.close()
            await self._server.wait_closed()
            self._server = None

    def close_session_sync(self, session_id: str, *, reason: str) -> None:
        if not self._loop:
            return
        future = asyncio.run_coroutine_threadsafe(
            self._close_session(session_id, reason=reason),
            self._loop,
        )
        with contextlib.suppress(Exception):
            future.result(timeout=5.0)

    async def _handle_connection(
        self,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
    ) -> None:
        actor: MediaActor | None = None
        close_reason = "connection_closed"
        close_as_error = False
        peer = writer.get_extra_info("peername")
        try:
            logger.info("audiosocket.accept peer=%s", peer)
            packet_type, payload = await read_packet(reader, timeout=self._idle_timeout_seconds)
            logger.info(
                "audiosocket.initial_packet peer=%s packet_type=%s payload_len=%s",
                peer,
                packet_type,
                len(payload),
            )
            if packet_type != AUDIO_SOCKET_PACKET_UUID:
                raise RuntimeError("AudioSocket UUID handshake is required")
            media_uuid = normalize_media_uuid(payload)
            logger.info("audiosocket.handshake peer=%s media_uuid=%s", peer, media_uuid)
            registration = await self._await_registration(
                media_uuid,
                writer=writer,
                frame_bytes=self._frame_bytes,
            )
            if registration is None:
                raise RuntimeError(f"Unknown AudioSocket media_uuid: {media_uuid}")
            logger.info(
                "audiosocket.registered peer=%s session_id=%s media_uuid=%s",
                peer,
                registration.voice_session_id,
                media_uuid,
            )

            actor = MediaActor(
                registration=registration,
                reader=reader,
                writer=writer,
                vad=EnergyVAD(
                    frame_ms=self._frame_ms,
                    min_speech_ms=self._min_speech_ms,
                    trailing_silence_ms=self._trailing_silence_ms,
                    max_turn_ms=self._max_turn_ms,
                    rms_threshold=self._vad_rms_threshold,
                ),
                frame_ms=self._frame_ms,
                frame_bytes=self._frame_bytes,
            )
            previous_actor = self._actors.get(registration.voice_session_id)
            if previous_actor is not None:
                await self._cleanup_actor(previous_actor, reason="media_replaced", error=False)
            self._actors[registration.voice_session_id] = actor
            actor.last_outbound_audio_monotonic = time.monotonic()
            await asyncio.to_thread(self._mark_media_connected, registration.voice_session_id, media_uuid)
            actor.keepalive_task = asyncio.create_task(self._keepalive_loop(actor))
            actor.worker_task = asyncio.create_task(self._worker(actor))

            while not actor.closed:
                packet_type, payload = await read_packet(reader, timeout=self._idle_timeout_seconds)
                if packet_type == AUDIO_SOCKET_PACKET_PCM16:
                    await self._handle_pcm(actor, payload)
                    continue
                if packet_type == AUDIO_SOCKET_PACKET_DTMF:
                    logger.info(
                        "audiosocket.dtmf session_id=%s payload_len=%s",
                        actor.registration.voice_session_id,
                        len(payload),
                    )
                    continue
                if packet_type == AUDIO_SOCKET_PACKET_HANGUP:
                    close_reason = "audiosocket_hangup"
                    break
                if packet_type == AUDIO_SOCKET_PACKET_UUID:
                    logger.info(
                        "audiosocket.extra_uuid session_id=%s payload_len=%s",
                        actor.registration.voice_session_id,
                        len(payload),
                    )
                    continue
                raise RuntimeError(f"Unsupported AudioSocket packet type: {packet_type}")
        except asyncio.TimeoutError:
            close_reason = "media_idle_timeout"
            close_as_error = True
            logger.warning("audiosocket.timeout peer=%s", peer)
        except asyncio.IncompleteReadError:
            close_reason = "connection_closed"
            logger.info("audiosocket.peer_closed peer=%s", peer)
        except Exception as exc:
            close_reason = str(exc)[:1000] or "media_runtime_error"
            close_as_error = True
            logger.warning("audiosocket.error peer=%s reason=%s", peer, close_reason)
        finally:
            if actor is not None:
                await self._cleanup_actor(actor, reason=close_reason, error=close_as_error)
            else:
                writer.close()
                with contextlib.suppress(Exception):
                    await writer.wait_closed()

    async def _handle_pcm(self, actor: MediaActor, pcm_frame: bytes) -> None:
        if actor.closed or not pcm_frame:
            return
        if not actor.first_pcm_logged:
            actor.first_pcm_logged = True
            logger.info(
                "audiosocket.first_pcm session_id=%s state=%s frame_bytes=%s",
                actor.registration.voice_session_id,
                actor.state,
                len(pcm_frame),
            )
        now = time.monotonic()
        if now - actor.last_media_touch_monotonic >= 1.0:
            actor.last_media_touch_monotonic = now
            await asyncio.to_thread(self._touch_media_frame, actor.registration.voice_session_id)

        if actor.state not in {"listening", "speaking", "thinking"}:
            return

        is_speech = audioop.rms(pcm_frame, 2) >= self._vad_rms_threshold
        if actor.state == "thinking":
            self._buffer_thinking_continuation(actor, pcm_frame, is_speech=is_speech, now=now)
            return
        if actor.state == "speaking":
            if is_speech:
                actor.barge_in_speech_ms += actor.frame_ms
                if actor.barge_in_speech_ms >= self._barge_in_trigger_ms and not actor.playback_interrupt.is_set():
                    actor.barge_in_detected_monotonic = time.monotonic()
                    actor.playback_interrupt.set()
                    actor.barge_in_pending = True
                    actor.speech_started_monotonic = actor.speech_started_monotonic or time.monotonic()
                    return
            else:
                actor.barge_in_speech_ms = 0
            if not actor.barge_in_pending:
                return
        else:
            actor.barge_in_speech_ms = 0

        vad_result = actor.vad.feed(pcm_frame)
        if vad_result.speech_started:
            self._reset_live_turn_state(actor)
            logger.info(
                "audiosocket.speech_started session_id=%s",
                actor.registration.voice_session_id,
            )
            await self._ensure_streaming_asr(actor)
            if actor.asr_streaming_enabled and actor.asr_stream_id:
                # The local VAD only flags speech_started after min_speech_ms of
                # buffered pre-roll audio (see EnergyVAD.feed), and that pre-roll
                # is folded into snapshot_utterance_pcm() together with the current
                # frame. Without forwarding it here, every utterance's opening
                # ~min_speech_ms would never reach the realtime ASR stream, clipping
                # the first word(s) of each turn.
                leadin_pcm = actor.vad.snapshot_utterance_pcm()
                if len(leadin_pcm) > len(pcm_frame):
                    self._queue_streaming_asr_pcm(actor, leadin_pcm[: -len(pcm_frame)])
        stream_input_active = actor.input_active or actor.barge_in_pending
        if actor.asr_streaming_enabled and actor.asr_stream_id and stream_input_active:
            self._queue_streaming_asr_pcm(actor, pcm_frame)
            self._maybe_schedule_streaming_partial_poll(actor)
        elif actor.registration.voice_v2_partial_asr:
            self._maybe_schedule_partial_asr(actor)
        if vad_result.utterance_pcm:
            actor.input_active = False
            actor.speech_ended_monotonic = time.monotonic()
            logger.info(
                "audiosocket.utterance_finalized session_id=%s duration_ms=%s barge_in=%s",
                actor.registration.voice_session_id,
                int(len(vad_result.utterance_pcm) / 16),
                actor.barge_in_pending,
            )
            await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending))
            actor.barge_in_pending = False

    async def _worker(self, actor: MediaActor) -> None:
        try:
            greeting_text = await asyncio.to_thread(
                self._get_pending_greeting,
                actor.registration.voice_session_id,
            )
            if greeting_text:
                logger.info(
                    "audiosocket.greeting session_id=%s text_len=%s",
                    actor.registration.voice_session_id,
                    len(greeting_text),
                )
                await self._speak_reply(actor, greeting_text, is_greeting=True, reply_phase="greeting")
            if not actor.closed:
                await self._set_actor_state(actor, "listening")

            while not actor.closed:
                pcm_bytes, barge_in = await actor.turn_queue.get()
                if actor.closed:
                    break
                await self._process_utterance(actor, pcm_bytes, barge_in)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            await self._handle_turn_exception(actor, exc)

    async def _handle_turn_exception(self, actor: MediaActor, exc: Exception) -> None:
        error_text = str(exc)[:500] or "voice_turn_failed"
        metadata = {
            "media_uuid": actor.registration.media_uuid,
            "queue_code": actor.registration.queue_code,
            "reply_phase": actor.current_reply_phase or "error_handoff",
            "error_class": exc.__class__.__name__,
            "error_category": self._asr_failure_category(exc),
        }
        logger.warning(
            "audiosocket.turn_failed session_id=%s category=%s error=%s",
            actor.registration.voice_session_id,
            metadata["error_category"],
            error_text,
        )
        await self._set_actor_state(actor, "handoff_requested", error_text)
        fallback_text = self._technical_issue_text(actor.registration.language)
        with contextlib.suppress(Exception):
            await self._plan_reply_segment(
                actor,
                fallback_text,
                kind="reply",
                metadata={
                    **metadata,
                    "phase": "error_handoff",
                    "technical_fallback": True,
                },
            )
            await self._speak_reply(
                actor,
                fallback_text,
                is_greeting=False,
                reply_phase="error_handoff",
            )
        await asyncio.to_thread(
            self._handle_media_error,
            actor.registration.voice_session_id,
            error_text,
            metadata,
        )

    async def _process_utterance(self, actor: MediaActor, pcm_bytes: bytes, barge_in: bool) -> None:
        await self._set_actor_state(actor, "thinking")
        pcm_bytes = await self._extend_with_thinking_continuation(actor, pcm_bytes)
        actor.playback_generation += 1
        actor.tts_generation += 1
        actor.response_plan_id = actor.response_plan_id or f"rsp_{uuid.uuid4().hex[:10]}"
        utterance_generation = actor.utterance_generation
        partial_transcript = str(actor.stable_partial_transcript or actor.partial_transcript or "").strip()
        partial_intent = str(actor.stable_partial_intent or actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
        base_metadata = {
            "turn_duration_ms": int(len(pcm_bytes) / 16),
            "media_uuid": actor.registration.media_uuid,
            "queue_code": actor.registration.queue_code,
            "voice_v2_enabled": actor.registration.voice_v2_enabled,
            "response_plan_id": actor.response_plan_id,
            "playback_generation": actor.playback_generation,
            "tts_generation": actor.tts_generation,
            "partial_transcript": partial_transcript,
            "early_intent": partial_intent,
        }
        if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started:
            partial_task = actor.partial_asr_task
            if partial_task is not None and not partial_task.done():
                with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
                    await asyncio.wait_for(asyncio.shield(partial_task), timeout=0.08)
            partial_transcript = str(actor.stable_partial_transcript or actor.partial_transcript or "").strip()
            partial_intent = str(actor.stable_partial_intent or actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
            base_metadata["partial_transcript"] = partial_transcript
            base_metadata["early_intent"] = partial_intent
            if partial_transcript:
                self._schedule_early_plan(
                    actor,
                    transcript_text=partial_transcript,
                    language=actor.registration.language,
                    intent=partial_intent,
                    source="turn_close_partial",
                )
            if self._should_emit_partial_ack(actor, partial_transcript, partial_intent):
                await self._emit_early_ack(
                    actor,
                    language=actor.registration.language,
                    metadata=base_metadata,
                    ack_source="streaming_partial" if actor.asr_streaming_enabled else "precomputed_partial_asr",
                )
            elif self._should_emit_blind_ack(actor, pcm_bytes):
                await self._emit_early_ack(
                    actor,
                    language=actor.registration.language,
                    metadata=base_metadata,
                    ack_source="immediate_turn_close",
                    ack_kind="unknown",
                )

        transcript_source = "batch"
        asr_start_monotonic = time.monotonic()
        if actor.asr_streaming_enabled and not actor.asr_streaming_failed:
            try:
                transcription = await self._finalize_streaming_transcription(actor)
                transcript_source = "streaming_final"
            except StreamingASRUnavailable as exc:
                logger.warning(
                    "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s",
                    actor.registration.voice_session_id,
                    str(exc)[:500],
                )
                self._mark_streaming_asr_backoff(actor)
                await self._close_streaming_asr(actor, drain=False)
                partial_first_text = str(actor.stable_partial_transcript or actor.partial_transcript or "").strip()
                if self._partial_first_final_enabled and partial_first_text and not self._is_low_signal_final_transcript(partial_first_text):
                    transcription = ASRTranscription(
                        text=partial_first_text,
                        language=actor.registration.language,
                        confidence=None,
                    )
                    transcript_source = "streaming_partial_after_finalize_failure"
                else:
                    wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
                    transcription = await asyncio.to_thread(
                        self._asr_provider.transcribe,
                        wav_bytes,
                        language_hint=actor.registration.language,
                    )
                    transcript_source = "batch_fallback_after_streaming_failure"
        else:
            wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
            transcription = await asyncio.to_thread(
                self._asr_provider.transcribe,
                wav_bytes,
                language_hint=actor.registration.language,
            )
            transcript_source = "batch"
        transcript_text = str(transcription.text or "").strip() or partial_transcript
        logger.info(
            "audiosocket.asr_turn_ready session_id=%s provider=%s source=%s utterance_ms=%s elapsed_ms=%s "
            "text_len=%s empty=%s text=%s",
            actor.registration.voice_session_id,
            getattr(self._asr_provider, "name", "unknown"),
            transcript_source,
            int(len(pcm_bytes) / 16),
            int((time.monotonic() - asr_start_monotonic) * 1000),
            len(str(transcript_text or "").strip()),
            not bool(str(transcript_text or "").strip()),
            transcript_text[:160],
        )
        if not transcript_text:
            actor.consecutive_low_signal_turns += 1
            await self._set_actor_state(actor, "listening")
            if actor.consecutive_low_signal_turns >= self._no_speech_reprompt_low_signal_turns:
                await self._maybe_play_no_speech_reprompt(actor, trigger="empty_transcript")
            return
        if self._is_low_signal_final_transcript(transcript_text):
            actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation)
            actor.consecutive_low_signal_turns += 1
            logger.info(
                "audiosocket.low_signal_ignored session_id=%s transcript=%s",
                actor.registration.voice_session_id,
                transcript_text[:120],
            )
            await self._set_actor_state(actor, "listening")
            if actor.consecutive_low_signal_turns >= self._no_speech_reprompt_low_signal_turns:
                await self._maybe_play_no_speech_reprompt(actor, trigger="low_signal_transcript")
            return

        actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation)
        actor.finalized_caller_turn_count += 1
        actor.consecutive_low_signal_turns = 0
        actor.reprompt_attempts = 0
        actor.partial_transcript = transcript_text if actor.registration.voice_v2_partial_asr else None
        actor.partial_intent = self._detect_early_intent(transcript_text)
        actor.stable_partial_intent = actor.partial_intent
        metadata = {
            **base_metadata,
            "partial_transcript": actor.partial_transcript,
            "early_intent": actor.partial_intent,
            "reply_phase": "final",
            "transcript_source": transcript_source,
        }
        decision_task = asyncio.create_task(
            asyncio.to_thread(
                self._process_turn,
                actor.registration.voice_session_id,
                transcript_text,
                transcription.language or actor.registration.language,
                barge_in,
                {
                    **metadata,
                    "runtime_defer_reply_planned": self._should_use_voice_v2(actor.registration),
                },
            )
        )
        early_plan_decision: VoiceAITurnDecisionOut | None = None
        if self._should_use_voice_v2(actor.registration):
            early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.04)
            if early_plan_decision is not None:
                decision = early_plan_decision
                logger.info(
                    "audiosocket.early_plan_used session_id=%s generation=%s intent=%s",
                    actor.registration.voice_session_id,
                    actor.early_plan_generation,
                    actor.early_plan_intent,
                )
            else:
                try:
                    decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds)
                except asyncio.TimeoutError:
                    if not actor.early_ack_started:
                        await self._emit_early_ack(
                            actor,
                            language=transcription.language or actor.registration.language,
                            metadata=metadata,
                            ack_source="decision_timeout",
                        )
                    early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.12)
                    decision = early_plan_decision if early_plan_decision is not None else await decision_task
        else:
            try:
                decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v1_ack_wait_seconds)
            except asyncio.TimeoutError:
                if not actor.early_ack_started:
                    await self._emit_early_ack(
                        actor,
                        language=transcription.language or actor.registration.language,
                        metadata=metadata,
                        ack_source="v1_decision_timeout",
                        ack_kind="unknown",
                    )
                decision = await decision_task
        handoff_task: asyncio.Task | None = None
        if decision.needs_handoff:
            await self._set_actor_state(actor, "handoff_requested", decision.handoff_reason)
            handoff_task = self._start_handoff_request(actor, transcript_text, decision)
        if decision.reply_text:
            if self._should_use_voice_v2(actor.registration) and actor.early_ack_started:
                remaining_gap = self._v2_ack_post_gap_seconds - max(
                    0.0,
                    time.monotonic() - actor.last_ack_completed_monotonic,
                )
                if remaining_gap > 0:
                    await asyncio.sleep(remaining_gap)
            if self._should_use_voice_v2(actor.registration):
                await self._plan_reply_segment(
                    actor,
                    decision.reply_text,
                    kind="reply",
                    metadata={
                        **metadata,
                        **(decision.metadata if isinstance(decision.metadata, dict) else {}),
                        "phase": "main",
                        "early_ack_started": actor.early_ack_started,
                    },
                )
            await self._speak_reply(actor, decision.reply_text, is_greeting=False, reply_phase="main")
        if early_plan_decision is not None:
            if decision_task.done():
                await self._reconcile_final_decision_after_early_plan(
                    actor,
                    decision_task=decision_task,
                    early_decision=early_plan_decision,
                    transcript_text=transcript_text,
                )
            else:
                asyncio.create_task(
                    self._reconcile_final_decision_after_early_plan(
                        actor,
                        decision_task=decision_task,
                        early_decision=early_plan_decision,
                        transcript_text=transcript_text,
                    ),
                    name=f"voice-final-plan-reconcile-{actor.registration.voice_session_id}",
                )
        if actor.closed:
            return
        if decision.needs_handoff:
            if handoff_task is not None and handoff_task.done():
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await handoff_task
            return
        actor.input_active = False
        await self._set_actor_state(actor, "listening")

    def _start_handoff_request(
        self,
        actor: MediaActor,
        customer_request_text: str,
        decision: VoiceAITurnDecisionOut,
    ) -> asyncio.Task:
        existing = actor.handoff_task
        if existing is not None and not existing.done():
            return existing

        async def _run() -> None:
            try:
                await asyncio.to_thread(
                    self._request_handoff,
                    actor.registration.voice_session_id,
                    customer_request_text,
                    decision,
                )
            except Exception as exc:
                logger.warning(
                    "audiosocket.handoff_request_failed session_id=%s error=%s",
                    actor.registration.voice_session_id,
                    str(exc)[:500],
                )

        actor.handoff_task = asyncio.create_task(_run())
        return actor.handoff_task

    async def _stream_tts_chunks(
        self,
        actor: MediaActor,
        text: str,
        *,
        style_hints: dict[str, object] | None = None,
    ):
        if not actor.registration.voice_v2_streaming_tts:
            synthesis = await asyncio.to_thread(
                lambda: self._tts_provider.synthesize(
                    text,
                    language=actor.registration.language,
                    style_hints=style_hints,
                )
            )
            if synthesis.audio_bytes:
                yield synthesis
            return

        loop = asyncio.get_running_loop()
        queue: asyncio.Queue[Any] = asyncio.Queue()
        done = object()

        def _producer() -> None:
            try:
                for synthesis in self._tts_provider.synthesize_chunks(
                    text,
                    language=actor.registration.language,
                    style_hints=style_hints,
                ):
                    asyncio.run_coroutine_threadsafe(queue.put(synthesis), loop).result()
            except Exception as exc:  # pragma: no cover - defensive bridge from thread to loop
                asyncio.run_coroutine_threadsafe(queue.put(exc), loop).result()
            finally:
                asyncio.run_coroutine_threadsafe(queue.put(done), loop).result()

        threading.Thread(
            target=_producer,
            name=f"tts-stream-{actor.registration.voice_session_id}",
            daemon=True,
        ).start()
        while True:
            item = await queue.get()
            if item is done:
                break
            if isinstance(item, Exception):
                raise item
            yield item

    async def _speak_text(
        self,
        actor: MediaActor,
        text: str,
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
        reply_phase: str | None = "main",
    ) -> None:
        if actor.closed or not text:
            return
        actor.playback_interrupt.clear()
        actor.barge_in_detected_monotonic = 0.0
        actor.barge_in_speech_ms = 0
        await self._set_actor_state(actor, "speaking")
        actor.current_reply_phase = reply_phase
        synth_started_at = time.monotonic()
        first_chunk_logged = False
        total_audio_bytes = 0
        interrupted = False
        pending_pcm = bytearray()
        resample_state: object | None = None
        prebuffered = False
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="started",
        )

        pacer = _FramePacer(frame_seconds=actor.frame_ms / 1000.0)

        def _log_first_frame(frame: bytes) -> None:
            logger.info(
                "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s",
                actor.registration.voice_session_id,
                is_greeting,
                len(frame),
            )

        async def _write_pcm_frames(pcm_bytes: bytes) -> bool:
            return await self._write_paced_pcm_frames(actor, pcm_bytes, pacer, on_first_frame=_log_first_frame)

        async for synthesis in self._stream_tts_chunks(actor, text, style_hints=style_hints):
            if not synthesis.audio_bytes:
                continue
            total_audio_bytes += len(synthesis.audio_bytes)
            if not first_chunk_logged:
                first_chunk_logged = True
                logger.info(
                    "audiosocket.tts_ready session_id=%s greeting=%s synth_ms=%s audio_bytes=%s",
                    actor.registration.voice_session_id,
                    is_greeting,
                    int((time.monotonic() - synth_started_at) * 1000),
                    total_audio_bytes,
                )
                if reply_phase == "ack":
                    await self._record_latency_metric(actor, "speech_end_to_ack_start", actor.speech_ended_monotonic)
                    await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic)
                elif reply_phase == "main":
                    await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic)
            # Carry ratecv state across chunks: resampling each network chunk
            # independently introduces an audible click/discontinuity at every
            # chunk boundary, which gets much more noticeable with a fast model
            # that streams many small chunks (e.g. eleven_flash_v2_5).
            pcm_8k, resample_state = resample_pcm16le_stateful(
                synthesis.audio_bytes,
                input_rate_hz=synthesis.sample_rate_hz,
                output_rate_hz=8000,
                state=resample_state,
            )
            if prebuffered:
                # Past the initial cushion: feed each resampled chunk straight
                # through instead of round-tripping it through pending_pcm.
                interrupted = await _write_pcm_frames(pcm_8k)
                if interrupted:
                    break
                continue
            pending_pcm.extend(pcm_8k)
            if len(pending_pcm) < self._tts_stream_prebuffer_bytes:
                continue
            prebuffered = True
            interrupted = await _write_pcm_frames(bytes(pending_pcm))
            pending_pcm.clear()
            if interrupted:
                break

        if not interrupted and pending_pcm:
            interrupted = await _write_pcm_frames(bytes(pending_pcm))
            pending_pcm.clear()

        if total_audio_bytes <= 0:
            raise RuntimeError("TTS provider returned empty audio")

        interrupted = interrupted or actor.playback_interrupt.is_set()
        actor.playback_interrupt.clear()
        actor.current_reply_phase = None
        if interrupted or actor.closed:
            await self._record_latency_metric(actor, "barge_in_detected_to_playback_stopped", actor.barge_in_detected_monotonic)
            await self._record_reply_status(
                actor,
                text=text,
                is_greeting=is_greeting,
                phase=reply_phase,
                status="interrupted" if interrupted else "discarded",
            )
            return
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="delivered",
        )
        logger.info(
            "audiosocket.reply_delivered session_id=%s greeting=%s phase=%s",
            actor.registration.voice_session_id,
            is_greeting,
            reply_phase,
        )

    async def _write_audio_packet(self, actor: MediaActor, pcm_frame: bytes) -> None:
        if actor.closed:
            return
        async with actor.write_lock:
            if actor.closed:
                return
            actor.writer.write(encode_audio_packet(pcm_frame))
            await actor.writer.drain()
            actor.last_outbound_audio_monotonic = time.monotonic()

    async def _write_paced_pcm_frames(
        self,
        actor: MediaActor,
        pcm_bytes: bytes,
        pacer: _FramePacer,
        *,
        on_first_frame: Callable[[bytes], None] | None = None,
    ) -> bool:
        """Write pcm_bytes out as frame_bytes-sized packets, paced by `pacer`.

        Returns True if playback was interrupted (barge-in or actor closed).
        """
        for frame in chunk_audio(pcm_bytes, frame_bytes=actor.frame_bytes):
            if actor.closed or actor.playback_interrupt.is_set():
                return True
            await self._write_audio_packet(actor, frame)
            if pacer.frames_written == 0 and on_first_frame is not None:
                on_first_frame(frame)
            await pacer.wait_for_next_frame()
        return False

    async def _keepalive_loop(self, actor: MediaActor) -> None:
        silence_frame = b"\x00" * actor.frame_bytes
        while not actor.closed:
            await asyncio.sleep(0.25)
            if actor.closed:
                break
            await self._maybe_reprompt_on_silence(actor)
            if (time.monotonic() - actor.last_outbound_audio_monotonic) < self._outbound_keepalive_interval_seconds:
                continue
            if not actor.keepalive_loop_logged:
                actor.keepalive_loop_logged = True
                logger.info(
                    "audiosocket.keepalive_loop session_id=%s state=%s frame_bytes=%s",
                    actor.registration.voice_session_id,
                    actor.state,
                    actor.frame_bytes,
                )
            await self._write_audio_packet(actor, silence_frame)

    async def _maybe_play_no_speech_reprompt(self, actor: MediaActor, *, trigger: str) -> None:
        if actor.closed or actor.state != "listening":
            return
        if actor.reprompt_attempts >= self._no_speech_reprompt_max_attempts:
            return
        actor.reprompt_attempts += 1
        actor.consecutive_low_signal_turns = 0
        logger.info(
            "audiosocket.no_speech_reprompt session_id=%s attempt=%s trigger=%s",
            actor.registration.voice_session_id,
            actor.reprompt_attempts,
            trigger,
        )
        reprompt_text = self._no_speech_reprompt_text(actor.registration.language)
        await self._speak_reply(actor, reprompt_text, is_greeting=False, reply_phase="no_speech_reprompt")
        if not actor.closed:
            await self._set_actor_state(actor, "listening")

    async def _maybe_reprompt_on_silence(self, actor: MediaActor) -> None:
        if not self._no_speech_reprompt_enabled:
            return
        if actor.state != "listening" or actor.listening_since_monotonic <= 0:
            return
        if time.monotonic() - actor.listening_since_monotonic < self._no_speech_reprompt_seconds:
            return
        await self._maybe_play_no_speech_reprompt(actor, trigger="silence_timeout")

    async def _set_actor_state(
        self,
        actor: MediaActor,
        state: str,
        handoff_reason: str | None = None,
    ) -> None:
        if actor.closed and state not in {"error", "closed"}:
            return
        if actor.state == state and (handoff_reason or None) is None:
            return
        logger.info(
            "audiosocket.state session_id=%s from_state=%s to_state=%s handoff_reason=%s",
            actor.registration.voice_session_id,
            actor.state,
            state,
            handoff_reason,
        )
        actor.state = state
        actor.input_active = state == "listening"
        actor.playback_active = state == "speaking"
        if state == "listening":
            actor.listening_since_monotonic = time.monotonic()
        await asyncio.to_thread(
            self._set_state,
            actor.registration.voice_session_id,
            state,
            handoff_reason,
            {"media_uuid": actor.registration.media_uuid},
        )

    async def _close_session(self, session_id: str, *, reason: str) -> None:
        actor = self._actors.get(session_id)
        if actor is None:
            return
        await self._cleanup_actor(actor, reason=reason, error=False)

    async def _cleanup_actor(self, actor: MediaActor, *, reason: str, error: bool) -> None:
        if actor.closed:
            return
        actor.closed = True
        logger.info(
            "audiosocket.cleanup session_id=%s reason=%s error=%s first_pcm=%s",
            actor.registration.voice_session_id,
            reason,
            error,
            actor.first_pcm_logged,
        )
        self._actors.pop(actor.registration.voice_session_id, None)
        if actor.keepalive_task is not None:
            actor.keepalive_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.keepalive_task
        if actor.handoff_task is not None:
            actor.handoff_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.handoff_task
        if actor.partial_asr_task is not None:
            actor.partial_asr_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.partial_asr_task
        with contextlib.suppress(Exception):
            await self._close_streaming_asr(actor, drain=False)
        if actor.worker_task is not None:
            actor.worker_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.worker_task
        actor.writer.close()
        with contextlib.suppress(Exception):
            await actor.writer.wait_closed()

        if error:
            await asyncio.to_thread(
                self._handle_media_error,
                actor.registration.voice_session_id,
                reason,
                {"media_uuid": actor.registration.media_uuid},
            )
            return
        await asyncio.to_thread(self._mark_media_ended, actor.registration.voice_session_id, reason)
