feat(voice): add duplex streaming v2 pipeline

2026-04-11 17:38:21 +05:00
parent fc976804e4
commit b75b76fae6
10 changed files with 1219 additions and 206 deletions
@@ -1,6 +1,7 @@
 from __future__ import annotations

 import asyncio
+import audioop
 import contextlib
 import hashlib
 import logging
@@ -10,6 +11,7 @@ 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,
@@ -23,7 +25,13 @@ from services.ai_voice_runtime_service.audiosocket import (
    read_packet,
    resample_pcm16le,
 )
-from services.ai_voice_runtime_service.providers.asr import ASRProvider
+from services.ai_voice_runtime_service.providers.asr import (
+    ASRTranscription,
+    ASRProvider,
+    StreamingASRPartial,
+    StreamingASRProvider,
+    StreamingASRUnavailable,
+)
 from services.ai_voice_runtime_service.providers.tts import TTSProvider
 from services.shared.models import VoiceAITurnDecisionOut

@@ -46,6 +54,9 @@ class MediaRegistration:
    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

@@ -71,17 +82,31 @@ class MediaActor:
    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
    partial_intent: str | None = None
+    partial_intent_streak: int = 0
+    stable_partial_intent: str | None = None
    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
    partial_asr_attempted: bool = False
+    asr_stream_id: str | None = None
+    asr_streaming_enabled: bool = False
+    asr_poll_due_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
    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


 class AudioSocketMediaRuntime:
@@ -98,6 +123,7 @@ class AudioSocketMediaRuntime:
        trailing_silence_ms: int,
        max_turn_ms: int,
        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],
@@ -105,8 +131,11 @@ class AudioSocketMediaRuntime:
        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_delivered: Callable[[str, str, bool], 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],
@@ -123,15 +152,20 @@ class AudioSocketMediaRuntime:
        self._trailing_silence_ms = max(trailing_silence_ms, self._frame_ms)
        self._max_turn_ms = max(max_turn_ms, self._frame_ms)
        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
@@ -139,9 +173,12 @@ class AudioSocketMediaRuntime:
        self._loop: asyncio.AbstractEventLoop | None = None
        self._actors: dict[str, MediaActor] = {}
        self._v2_ack_wait_seconds = 0.18
-        self._partial_asr_min_ms = 650
+        self._partial_asr_min_ms = 320
        self._immediate_ack_min_ms = 280
-        self._v2_ack_post_gap_seconds = 0.12
+        self._v2_ack_post_gap_seconds = 0.10
+        self._partial_poll_interval_seconds = 0.20
+        self._stable_partial_hold_seconds = 0.40
+        self._barge_in_trigger_ms = 220

    @staticmethod
    def _normalize_intent_text(text: str) -> str:
@@ -171,7 +208,9 @@ class AudioSocketMediaRuntime:
            return "handoff"
        if intent in {"schedule", "address", "price", "status", "problem"}:
            return "understanding"
-        return "generic"
+        if intent == "unknown":
+            return "unknown"
+        return "clarify"

    @staticmethod
    def _ack_text(language: str | None, ack_kind: str) -> str:
@@ -181,11 +220,15 @@ class AudioSocketMediaRuntime:
                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
@@ -218,6 +261,13 @@ class AudioSocketMediaRuntime:
                "Хорошо, сейчас подскажу.",
                "Понял вас, секунду.",
            )
+        if ack_kind == "clarify":
+            return (
+                "Угу, сейчас уточню.",
+                "Мхм, одну секунду.",
+                "Хорошо, сейчас уточню.",
+                "Ага, сейчас сориентирую.",
+            )
        return (
            "Угу, сейчас подскажу.",
            "Мхм, я в контексте.",
@@ -251,10 +301,11 @@ class AudioSocketMediaRuntime:
            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, f"{ack_kind}:{index}"
+        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")
@@ -271,16 +322,152 @@ class AudioSocketMediaRuntime:
    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.partial_intent = None
+        actor.partial_intent_streak = 0
+        actor.stable_partial_intent = None
        actor.response_plan_id = None
        actor.partial_asr_attempted = 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
        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
+        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
+            actor.registration.voice_v2_duplex = False
+            actor.registration.voice_v2_partial_asr = 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_poll_due_monotonic = 0.0
+
+    async def _close_streaming_asr(self, actor: MediaActor) -> None:
+        stream_id = actor.asr_stream_id
+        actor.asr_stream_id = None
+        actor.asr_streaming_enabled = False
+        if not stream_id:
+            return
+        await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id)
+
+    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
+
+    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
+        actor.partial_transcript = transcript_text
+        intent = self._detect_early_intent(transcript_text)
+        self._update_stable_partial_intent(actor, intent)
+
+    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
+        try:
+            return await asyncio.to_thread(self._streaming_asr_provider.finalize, stream_id)
+        finally:
+            await self._close_streaming_asr(actor)
+
+    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,
@@ -364,6 +551,7 @@ class AudioSocketMediaRuntime:
            ack_kind=ack_kind,
        )
        actor.early_ack_started = True
+        actor.current_reply_phase = "ack"
        await self._plan_reply_segment(
            actor,
            ack_text,
@@ -379,7 +567,10 @@ class AudioSocketMediaRuntime:
                "partial_ack_source": ack_source,
            },
        )
-        await self._speak_text(actor, ack_text, is_greeting=False, style_hints=style_hints)
+        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")
@@ -402,6 +593,112 @@ class AudioSocketMediaRuntime:
            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
+        await self._set_actor_state(actor, "speaking")
+        actor.current_reply_phase = reply_phase
+        interrupted = False
+        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)
+        for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes):
+            if actor.closed or actor.playback_interrupt.is_set():
+                interrupted = True
+                break
+            await self._write_audio_packet(actor, frame)
+            await asyncio.sleep(actor.frame_ms / 1000.0)
+        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.warning(
+            "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}"
@@ -580,17 +877,49 @@ class AudioSocketMediaRuntime:
            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"}:
+        if actor.state not in {"listening", "speaking", "thinking"}:
            return

+        is_speech = audioop.rms(pcm_frame, 2) >= 250
+        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()
+            else:
+                actor.barge_in_speech_ms = 0
+        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)
-        if actor.state == "speaking" and vad_result.speech_started:
-            actor.playback_interrupt.set()
-            actor.barge_in_pending = True
-        self._maybe_schedule_partial_asr(actor)
+            await self._ensure_streaming_asr(actor)
+        if actor.asr_streaming_enabled and actor.asr_stream_id and actor.input_active:
+            try:
+                await asyncio.to_thread(
+                    self._streaming_asr_provider.push_pcm,
+                    actor.asr_stream_id,
+                    pcm_frame,
+                )
+                await self._poll_streaming_partial(actor)
+            except StreamingASRUnavailable as exc:
+                logger.warning(
+                    "audiosocket.streaming_asr_push_failed session_id=%s error=%s",
+                    actor.registration.voice_session_id,
+                    str(exc)[:500],
+                )
+                await self._close_streaming_asr(actor)
+                actor.registration.voice_v2_duplex = False
+                actor.registration.voice_v2_partial_asr = False
+        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()
            await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending))
            actor.barge_in_pending = False

@@ -605,7 +934,7 @@ class AudioSocketMediaRuntime:
                actor.registration.voice_session_id,
                len(greeting_text),
            )
-            await self._speak_text(actor, greeting_text, is_greeting=True)
+            await self._speak_reply(actor, greeting_text, is_greeting=True, reply_phase="greeting")
        if not actor.closed:
            await self._set_actor_state(actor, "listening")

@@ -617,20 +946,12 @@ class AudioSocketMediaRuntime:

    async def _process_utterance(self, actor: MediaActor, pcm_bytes: bytes, barge_in: bool) -> None:
        await self._set_actor_state(actor, "thinking")
-        wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
        actor.playback_generation += 1
+        actor.tts_generation += 1
        actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
        utterance_generation = actor.utterance_generation
        partial_transcript = str(actor.partial_transcript or "").strip()
-        partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
-        full_asr_task = asyncio.create_task(
-            asyncio.to_thread(
-                lambda: self._asr_provider.transcribe(
-                    wav_bytes,
-                    language_hint=actor.registration.language,
-                )
-            )
-        )
+        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,
@@ -638,24 +959,33 @@ class AudioSocketMediaRuntime:
            "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)
+            if actor.asr_streaming_enabled:
+                with contextlib.suppress(StreamingASRUnavailable):
+                    await self._poll_streaming_partial(actor)
                    partial_transcript = str(actor.partial_transcript or "").strip()
-                    partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
+                    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
+            else:
+                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.partial_transcript or "").strip()
+                        partial_intent = str(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:
                await self._emit_early_ack(
                    actor,
                    language=actor.registration.language,
                    metadata=base_metadata,
-                    ack_source="precomputed_partial_asr",
+                    ack_source="streaming_partial" if actor.asr_streaming_enabled else "precomputed_partial_asr",
                )
            elif len(pcm_bytes) >= self._immediate_ack_min_bytes:
                await self._emit_early_ack(
@@ -665,7 +995,15 @@ class AudioSocketMediaRuntime:
                    ack_source="immediate_turn_close",
                )

-        transcription = await full_asr_task
+        if actor.asr_streaming_enabled:
+            transcription = await self._finalize_streaming_transcription(actor)
+        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_text = str(transcription.text or "").strip() or partial_transcript
        if not transcript_text:
            await self._set_actor_state(actor, "listening")
@@ -674,10 +1012,12 @@ class AudioSocketMediaRuntime:
        actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation)
        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",
        }
        decision_task = asyncio.create_task(
            asyncio.to_thread(
@@ -730,7 +1070,7 @@ class AudioSocketMediaRuntime:
                        "early_ack_started": actor.early_ack_started,
                    },
                )
-            await self._speak_text(actor, decision.reply_text, is_greeting=False)
+            await self._speak_reply(actor, decision.reply_text, is_greeting=False, reply_phase="main")
        if actor.closed:
            return
        if decision.needs_handoff:
@@ -738,6 +1078,7 @@ class AudioSocketMediaRuntime:
                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(
@@ -824,13 +1165,23 @@ class AudioSocketMediaRuntime:
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
+        reply_phase: str | None = "main",
    ) -> None:
        if actor.closed or not text:
            return
        await self._set_actor_state(actor, "speaking")
+        actor.current_reply_phase = reply_phase
        synth_started_at = time.monotonic()
        first_frame_sent = False
        total_audio_bytes = 0
+        interrupted = False
+        await self._record_reply_status(
+            actor,
+            text=text,
+            is_greeting=is_greeting,
+            phase=reply_phase,
+            status="started",
+        )
        async for synthesis in self._stream_tts_chunks(actor, text, style_hints=style_hints):
            if not synthesis.audio_bytes:
                continue
@@ -843,6 +1194,11 @@ class AudioSocketMediaRuntime:
                    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)
            pcm_8k = resample_pcm16le(
                synthesis.audio_bytes,
                input_rate_hz=synthesis.sample_rate_hz,
@@ -850,6 +1206,7 @@ class AudioSocketMediaRuntime:
            )
            for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes):
                if actor.closed or actor.playback_interrupt.is_set():
+                    interrupted = True
                    break
                await self._write_audio_packet(actor, frame)
                if not first_frame_sent:
@@ -862,25 +1219,38 @@ class AudioSocketMediaRuntime:
                    )
                await asyncio.sleep(actor.frame_ms / 1000.0)
            if actor.closed or actor.playback_interrupt.is_set():
+                interrupted = True
                break

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

-        interrupted = actor.playback_interrupt.is_set()
+        interrupted = interrupted or actor.playback_interrupt.is_set()
        actor.playback_interrupt.clear()
-        if not interrupted and not actor.closed:
-            await asyncio.to_thread(
-                self._mark_reply_delivered,
-                actor.registration.voice_session_id,
-                text,
-                is_greeting,
-            )
-            logger.warning(
-                "audiosocket.reply_delivered session_id=%s greeting=%s",
-                actor.registration.voice_session_id,
-                is_greeting,
+        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.warning(
+            "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:
@@ -928,6 +1298,8 @@ class AudioSocketMediaRuntime:
            handoff_reason,
        )
        actor.state = state
+        actor.input_active = state in {"listening", "speaking", "thinking"}
+        actor.playback_active = state == "speaking"
        await asyncio.to_thread(
            self._set_state,
            actor.registration.voice_session_id,
@@ -966,6 +1338,8 @@ class AudioSocketMediaRuntime:
            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)
        if actor.worker_task is not None:
            actor.worker_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):