feat(voice): add phase 1.1 partial asr fast ack

.env.example

@@ -50,10 +50,15 @@ AI_WHATSAPP_MAX_KB_RESULTS=3
 AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65
 AI_VOICE_RUNTIME_SERVICE_URL=http://localhost:8018
 AI_VOICE_ENABLED=0
-AI_VOICE_POLICY_MODE=llm_guarded
+AI_VOICE_POLICY_MODE=v2_fast_conversational
 AI_VOICE_QUEUE_CONFIG_JSON={}
 AI_VOICE_ASR_PROVIDER=openai
-AI_VOICE_TTS_PROVIDER=openai
+AI_VOICE_TTS_PROVIDER=yandex
+AI_VOICE_V2_ENABLED=1
+AI_VOICE_V2_QUEUE_CODES=voice_lab_ai
+AI_VOICE_V2_ACK_MODE=immediate_short
+AI_VOICE_V2_STREAMING_TTS=1
+AI_VOICE_V2_PARTIAL_ASR=1
 AI_VOICE_AUDIOSOCKET_ENABLED=0
 AI_VOICE_AUDIOSOCKET_HOST=0.0.0.0
 AI_VOICE_AUDIOSOCKET_PORT=9019
@@ -73,7 +78,7 @@ AI_VOICE_TTS_YANDEX_SAMPLE_RATE_HZ=8000
 AI_VOICE_TTS_CACHE_ENABLED=1
 AI_VOICE_TTS_CACHE_DIR=.data/ai_voice_tts_cache
 AI_VOICE_VAD_MIN_SPEECH_MS=300
-AI_VOICE_VAD_TRAILING_SILENCE_MS=700
+AI_VOICE_VAD_TRAILING_SILENCE_MS=400
 AI_VOICE_TURN_MAX_MS=10000
 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15
 AI_VOICE_MAX_CONTEXT_SEGMENTS=8

.env.production.template

@@ -45,10 +45,15 @@ AI_WHATSAPP_MAX_CONTEXT_MESSAGES=20
 AI_WHATSAPP_MAX_KB_RESULTS=3
 AI_WHATSAPP_CONFIDENCE_HANDOFF_THRESHOLD=0.65
 AI_VOICE_ENABLED=0
-AI_VOICE_POLICY_MODE=llm_guarded
+AI_VOICE_POLICY_MODE=v2_fast_conversational
 AI_VOICE_QUEUE_CONFIG_JSON={"voice_lab_ai":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}}
 AI_VOICE_ASR_PROVIDER=openai
-AI_VOICE_TTS_PROVIDER=openai
+AI_VOICE_TTS_PROVIDER=yandex
+AI_VOICE_V2_ENABLED=1
+AI_VOICE_V2_QUEUE_CODES=voice_lab_ai
+AI_VOICE_V2_ACK_MODE=immediate_short
+AI_VOICE_V2_STREAMING_TTS=1
+AI_VOICE_V2_PARTIAL_ASR=1
 AI_VOICE_AUDIOSOCKET_ENABLED=1
 AI_VOICE_AUDIOSOCKET_HOST=0.0.0.0
 AI_VOICE_AUDIOSOCKET_PORT=9019
@@ -69,7 +74,7 @@ AI_VOICE_TTS_YANDEX_SAMPLE_RATE_HZ=8000
 AI_VOICE_TTS_CACHE_ENABLED=1
 AI_VOICE_TTS_CACHE_DIR=/app/.data_local/ai_voice_tts_cache
 AI_VOICE_VAD_MIN_SPEECH_MS=300
-AI_VOICE_VAD_TRAILING_SILENCE_MS=700
+AI_VOICE_VAD_TRAILING_SILENCE_MS=400
 AI_VOICE_TURN_MAX_MS=10000
 AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS=15
 AI_VOICE_MAX_CONTEXT_SEGMENTS=8

deployment/docker-compose.yml

@@ -18,8 +18,9 @@ x-app-env: &app_env
   AI_VOICE_RUNTIME_SERVICE_URL: http://ai-voice-runtime-service:8000
   AI_VOICE_ENABLED: "0"
   AI_VOICE_QUEUE_CONFIG_JSON: "{}"
+  AI_VOICE_POLICY_MODE: v2_fast_conversational
   AI_VOICE_ASR_PROVIDER: openai
-  AI_VOICE_TTS_PROVIDER: openai
+  AI_VOICE_TTS_PROVIDER: yandex
   AI_VOICE_AUDIOSOCKET_ENABLED: "1"
   AI_VOICE_AUDIOSOCKET_HOST: 0.0.0.0
   AI_VOICE_AUDIOSOCKET_PORT: "9019"
@@ -29,7 +30,12 @@ x-app-env: &app_env
   AI_VOICE_TTS_CACHE_ENABLED: "1"
   AI_VOICE_TTS_CACHE_DIR: /app/.data/ai_voice_tts_cache
   AI_VOICE_VAD_MIN_SPEECH_MS: "300"
-  AI_VOICE_VAD_TRAILING_SILENCE_MS: "700"
+  AI_VOICE_VAD_TRAILING_SILENCE_MS: "400"
+  AI_VOICE_V2_ENABLED: "1"
+  AI_VOICE_V2_QUEUE_CODES: voice_lab_ai
+  AI_VOICE_V2_ACK_MODE: immediate_short
+  AI_VOICE_V2_STREAMING_TTS: "1"
+  AI_VOICE_V2_PARTIAL_ASR: "1"
   AI_VOICE_TURN_MAX_MS: "10000"
   AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS: "15"
   AI_VOICE_MAX_CONTEXT_SEGMENTS: "8"

scripts/local_stack.py

@@ -309,9 +309,16 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_
     env["AI_VOICE_ENABLED"] = env.get("AI_VOICE_ENABLED", "0")
     env["AI_VOICE_QUEUE_CONFIG_JSON"] = env.get("AI_VOICE_QUEUE_CONFIG_JSON", "{}")
     env["AI_VOICE_ASR_PROVIDER"] = env.get("AI_VOICE_ASR_PROVIDER", "openai")
-    env["AI_VOICE_TTS_PROVIDER"] = env.get("AI_VOICE_TTS_PROVIDER", "openai")
+    env["AI_VOICE_TTS_PROVIDER"] = env.get("AI_VOICE_TTS_PROVIDER", "yandex")
+    env["AI_VOICE_POLICY_MODE"] = env.get("AI_VOICE_POLICY_MODE", "v2_fast_conversational")
+    env["AI_VOICE_V2_ENABLED"] = env.get("AI_VOICE_V2_ENABLED", "1")
+    env["AI_VOICE_V2_QUEUE_CODES"] = env.get("AI_VOICE_V2_QUEUE_CODES", "voice_lab_ai")
+    env["AI_VOICE_V2_ACK_MODE"] = env.get("AI_VOICE_V2_ACK_MODE", "immediate_short")
+    env["AI_VOICE_V2_STREAMING_TTS"] = env.get("AI_VOICE_V2_STREAMING_TTS", "1")
+    env["AI_VOICE_V2_PARTIAL_ASR"] = env.get("AI_VOICE_V2_PARTIAL_ASR", "1")
     env["AI_VOICE_MAX_CONTEXT_SEGMENTS"] = env.get("AI_VOICE_MAX_CONTEXT_SEGMENTS", "8")
     env["AI_VOICE_HANDOFF_TIMEOUT_SECONDS"] = env.get("AI_VOICE_HANDOFF_TIMEOUT_SECONDS", "8")
+    env["AI_VOICE_VAD_TRAILING_SILENCE_MS"] = env.get("AI_VOICE_VAD_TRAILING_SILENCE_MS", "400")
     env["AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS"] = env.get(
         "AI_VOICE_RUNTIME_TRUSTED_SERVICE_SUBJECTS",
         "svc:ai-voice-runtime",

services/ai_orchestrator_service/voice.py

@@ -1060,6 +1060,78 @@ def _voice_policy_mode() -> str:
     return persona.voice_policy_mode()
 
 
+def _voice_v2_enabled(metadata: dict[str, Any] | None = None) -> bool:
+    payload = metadata if isinstance(metadata, dict) else {}
+    if bool(payload.get("voice_v2_enabled")):
+        return True
+    return _voice_policy_mode() == "v2_fast_conversational"
+
+
+def _voice_early_intent_bucket(text: str) -> str:
+    normalized = " ".join(str(text or "").strip().lower().split())
+    if not normalized:
+        return "unknown"
+    if any(token in normalized for token in ("оператор", "оператором", "человек", "менеджер", "сотрудник")):
+        return "operator_request"
+    if any(token in normalized for token in ("график", "распис", "жұмыс")):
+        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"
+
+
+def _voice_ack_kind_for_intent(intent: str) -> str:
+    if intent == "operator_request":
+        return "handoff"
+    if intent in {"schedule", "address", "price", "status", "problem"}:
+        return "understanding"
+    return "generic"
+
+
+def _voice_compact_reply_text(text: str, *, language: str) -> str:
+    normalized = " ".join(str(text or "").strip().split())
+    if not normalized:
+        return normalized
+    if len(normalized) <= 180:
+        return normalized
+    parts = [segment.strip() for segment in re.split(r"(?<=[.!?])\s+", normalized) if segment.strip()]
+    if parts:
+        compact = " ".join(parts[:2]).strip()
+        if len(compact) <= 180:
+            return compact
+    shortened = normalized[:177].rsplit(" ", 1)[0].strip()
+    if not shortened:
+        shortened = normalized[:177].strip()
+    ending = "…" if language == "kz" else "..."
+    return f"{shortened}{ending}"
+
+
+def _voice_v2_metadata(
+    transcript_text: str,
+    request_metadata: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    if not _voice_v2_enabled(request_metadata):
+        return {}
+    payload = request_metadata if isinstance(request_metadata, dict) else {}
+    early_intent = _voice_early_intent_bucket(transcript_text)
+    metadata: dict[str, Any] = {
+        "voice_v2_enabled": True,
+        "early_intent": early_intent,
+        "ack_kind": _voice_ack_kind_for_intent(early_intent),
+    }
+    for key in ("response_plan_id", "playback_generation", "partial_transcript"):
+        value = payload.get(key)
+        if value not in {None, ""}:
+            metadata[key] = value
+    return metadata
+
+
 def _voice_llm_prompt_messages(
     *,
     language: str,
@@ -1155,7 +1227,7 @@ def _voice_llm_decision(
     name_status: str | None,
 ) -> dict[str, Any] | None:
     app = _app()
-    if _voice_policy_mode() != "llm_guarded":
+    if _voice_policy_mode() not in {"llm_guarded", "v2_fast_conversational"}:
         return None
     if app._ai_provider() != "openai_compatible":
         return None
@@ -1385,12 +1457,14 @@ def _voice_decision(
     disclosure_required: bool,
     customer_name_value: str | None = None,
     customer_name_status: str | None = None,
+    request_metadata: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     app = _app()
     normalized = str(transcript_text or "").strip()
     lower_text = normalized.lower()
     caller_texts = _voice_recent_caller_texts(transcript_window)
     model = app._ai_model()
+    v2_metadata = _voice_v2_metadata(transcript_text, request_metadata)
 
     if app._looks_like_human_request(lower_text) or app._is_sensitive_request(lower_text):
         return {
@@ -1441,12 +1515,21 @@ def _voice_decision(
         name_status=customer_name_status,
     )
     if llm_decision is not None:
+        if v2_metadata:
+            llm_decision["reply_text"] = _voice_compact_reply_text(
+                str(llm_decision.get("reply_text") or ""),
+                language=language,
+            )
+            llm_decision["metadata"] = {
+                **(llm_decision.get("metadata") or {}),
+                **v2_metadata,
+            }
         return llm_decision
 
     if kb_results:
         article = kb_results[0]
         snippet = app._article_snippet(article, limit=220)
-        return {
+        decision = {
             "language": language,
             "intent": "kb_answer",
             "reply_text": (
@@ -1463,11 +1546,15 @@ def _voice_decision(
             "model": "voice_policy_fallback",
             "latency_ms": 1,
         }
+        if v2_metadata:
+            decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language)
+            decision["metadata"] = v2_metadata
+        return decision
 
     reply_text = _voice_confusion_prompt(language, caller_texts) if caller_confused else (
         _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language)
     )
-    return {
+    decision = {
         "language": language,
         "intent": "clarification",
         "reply_text": reply_text,
@@ -1480,6 +1567,10 @@ def _voice_decision(
         "model": "voice_policy_fallback",
         "latency_ms": 1,
     }
+    if v2_metadata:
+        decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language)
+        decision["metadata"] = v2_metadata
+    return decision
 
 
 def start_voice_session(session_id: str, payload: VoiceAIStartIn) -> VoiceAIStartOut:
@@ -1971,6 +2062,7 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe
             disclosure_required=disclosure_required,
             customer_name_value=name_update["value"],
             customer_name_status=name_update["status"],
+            request_metadata=payload.metadata if isinstance(payload.metadata, dict) else {},
         )
 
         if decision.get("extracted_name"):
@@ -2008,6 +2100,7 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe
             source=name_update["source"],
             resolved_at=name_update["resolved_at"],
         )
+        decision_metadata.update(decision.get("metadata") or {})
         if name_update["status"] == "name_obtained" and name_update["value"]:
             decision["reply_text"] = _voice_reply_with_name(
                 decision["language"],
@@ -2021,6 +2114,11 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe
                 if str(decision["reply_text"] or "").strip()
                 else inline_followup
             )
+        if decision_metadata.get("voice_v2_enabled"):
+            decision["reply_text"] = _voice_compact_reply_text(
+                decision["reply_text"],
+                language=decision["language"],
+            )
         decision["metadata"] = decision_metadata
         _record_voice_ai_turn(
             session,

services/ai_voice_runtime_service/app.py

@@ -33,7 +33,7 @@ from services.shared.models import (
 )
 from services.shared.security import issue_app_token, require_roles
 from services.shared.sql_init import init_sql_schema
-from services.shared.sql_models import VoiceAISessionRow, VoiceTranscriptSegmentRow
+from services.shared.sql_models import AsteriskCallLinkRow, VoiceAISessionRow, VoiceTranscriptSegmentRow
 from services.shared.voice_transcripts import add_transcript_segment, next_transcript_sequence
 
 init_sql_schema()
@@ -86,7 +86,7 @@ def _asr_provider_name() -> str:
 
 
 def _tts_provider_name() -> str:
-    return os.getenv("AI_VOICE_TTS_PROVIDER", "openai").strip() or "openai"
+    return os.getenv("AI_VOICE_TTS_PROVIDER", "yandex").strip() or "yandex"
 
 
 def _handoff_timeout_seconds() -> float:
@@ -114,7 +114,7 @@ def _vad_min_speech_ms() -> int:
 
 
 def _vad_trailing_silence_ms() -> int:
-    return max(_int_env("AI_VOICE_VAD_TRAILING_SILENCE_MS", 500), _vad_frame_ms())
+    return max(_int_env("AI_VOICE_VAD_TRAILING_SILENCE_MS", 400), _vad_frame_ms())
 
 
 def _turn_max_ms() -> int:
@@ -132,6 +132,27 @@ def _media_registration_wait_timeout_seconds() -> float:
     return max(_float_env("AI_VOICE_MEDIA_REGISTRATION_WAIT_TIMEOUT_SECONDS", 12.0), 0.0)
 
 
+def _voice_v2_enabled() -> bool:
+    return _bool_env("AI_VOICE_V2_ENABLED", False)
+
+
+def _voice_v2_queue_codes() -> set[str]:
+    raw = str(os.getenv("AI_VOICE_V2_QUEUE_CODES", "voice_lab_ai") or "voice_lab_ai").strip()
+    return {item.strip() for item in raw.split(",") if item.strip()}
+
+
+def _voice_v2_ack_mode() -> str:
+    return str(os.getenv("AI_VOICE_V2_ACK_MODE", "immediate_short") or "immediate_short").strip().lower()
+
+
+def _voice_v2_streaming_tts_enabled() -> bool:
+    return _bool_env("AI_VOICE_V2_STREAMING_TTS", True)
+
+
+def _voice_v2_partial_asr_enabled() -> bool:
+    return _bool_env("AI_VOICE_V2_PARTIAL_ASR", True)
+
+
 def _service_headers() -> dict[str, str]:
     token = issue_app_token(
         subject="svc:ai-voice-runtime",
@@ -457,7 +478,11 @@ def _push_bridge_call_state(
     )
 
 
-def _media_registration_from_row(row: VoiceAISessionRow) -> MediaRegistration:
+def _media_registration_from_row(row: VoiceAISessionRow, *, queue_code: str | None = None) -> MediaRegistration:
+    normalized_queue_code = str(queue_code or "").strip() or None
+    voice_v2_for_session = bool(
+        _voice_v2_enabled() and normalized_queue_code and normalized_queue_code in _voice_v2_queue_codes()
+    )
     return MediaRegistration(
         voice_session_id=row.session_id,
         call_id=row.call_id,
@@ -465,6 +490,13 @@ def _media_registration_from_row(row: VoiceAISessionRow) -> MediaRegistration:
         ai_session_id=str(row.ai_session_id or "").strip() or None,
         language=str(row.language or "").strip() or None,
         media_uuid=str(row.media_uuid or "").strip() or None,
+        queue_code=normalized_queue_code,
+        queue_id=str(row.queue_id or "").strip() or None,
+        agent_profile=str(row.agent_profile or "").strip() or None,
+        voice_v2_enabled=voice_v2_for_session,
+        voice_v2_ack_mode=_voice_v2_ack_mode() if voice_v2_for_session else "disabled",
+        voice_v2_streaming_tts=bool(voice_v2_for_session and _voice_v2_streaming_tts_enabled()),
+        voice_v2_partial_asr=bool(voice_v2_for_session and _voice_v2_partial_asr_enabled()),
     )
 
 
@@ -480,7 +512,14 @@ def _load_media_registration_by_uuid(media_uuid: str) -> MediaRegistration | Non
         ).scalar_one_or_none()
         if row is None:
             return None
-        return _media_registration_from_row(row)
+        call_link = session.execute(
+            select(AsteriskCallLinkRow)
+            .where(AsteriskCallLinkRow.call_id == row.call_id)
+            .order_by(AsteriskCallLinkRow.id.desc())
+            .limit(1)
+        ).scalar_one_or_none()
+        queue_code = str(call_link.queue_code or "").strip() or None if call_link is not None else None
+        return _media_registration_from_row(row, queue_code=queue_code)
     finally:
         session.close()
 
@@ -635,6 +674,42 @@ def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool) -> None
     _retry_db_write(_write)
 
 
+def _record_runtime_reply_planned(
+    session_id: str,
+    text: str,
+    metadata: dict[str, Any] | None = None,
+    kind: str = "reply",
+) -> None:
+    normalized_text = str(text or "").strip()
+    if not normalized_text:
+        return
+
+    def _write() -> None:
+        session = get_session()
+        try:
+            voice_session = _load_voice_session(session, session_id)
+            _record_segment(
+                session,
+                voice_session=voice_session,
+                speaker="assistant",
+                source_type="tts",
+                text=normalized_text,
+                sequence_no=_next_sequence(session, voice_session.session_id),
+                payload={
+                    "delivery_status": "planned",
+                    "kind": kind,
+                    "metadata": metadata or {},
+                },
+                is_final=False,
+            )
+            voice_session.updated_at = utc_now_iso()
+            session.commit()
+        finally:
+            session.close()
+
+    _retry_db_write(_write)
+
+
 def _set_voice_session_state(
     session_id: str,
     ai_state: str,
@@ -866,6 +941,7 @@ def _process_voice_ai_turn_sync(
 ) -> VoiceAITurnDecisionOut:
     session = get_session()
     voice_session = None
+    payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {}
     try:
         voice_session = _load_voice_session(session, session_id)
         if voice_session.status in {"human_owned", "completed", "error"}:
@@ -883,7 +959,7 @@ def _process_voice_ai_turn_sync(
             source_type="asr",
             text=payload.transcript_text,
             sequence_no=max(payload.sequence_no, _next_sequence(session, voice_session.session_id)),
-            payload={"metadata": payload.metadata, "barge_in": payload.barge_in},
+            payload={"metadata": payload_metadata, "barge_in": payload.barge_in},
         )
         session.commit()
 
@@ -899,7 +975,7 @@ def _process_voice_ai_turn_sync(
         _apply_voice_start_metadata(voice_session, decision.metadata)
         voice_session.status = decision.status or ("handoff_requested" if decision.needs_handoff else "active")
         voice_session.updated_at = utc_now_iso()
-        if decision.reply_text:
+        if decision.reply_text and not bool(payload_metadata.get("runtime_defer_reply_planned")):
             _record_segment(
                 session,
                 voice_session=voice_session,
@@ -1057,6 +1133,7 @@ _MEDIA_RUNTIME = AudioSocketMediaRuntime(
     set_state=_set_voice_session_state,
     get_pending_greeting=_load_pending_greeting_text,
     mark_reply_delivered=_mark_reply_delivered,
+    plan_reply=_record_runtime_reply_planned,
     process_turn=_media_process_turn,
     request_handoff=_request_runtime_handoff,
     handle_media_error=_handle_media_error,

services/ai_voice_runtime_service/audiosocket.py

@@ -158,3 +158,12 @@ class EnergyVAD:
         samples_per_ms = 8
         bytes_per_sample = 2
         return self._max_turn_ms * samples_per_ms * bytes_per_sample
+
+    @property
+    def is_active(self) -> bool:
+        return self._active
+
+    def snapshot_utterance_pcm(self) -> bytes:
+        if not self._active or not self._utterance_pcm:
+            return b""
+        return bytes(self._utterance_pcm)

services/ai_voice_runtime_service/media_runtime.py

@@ -3,7 +3,9 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import logging
+import threading
 import time
+import uuid
 from dataclasses import dataclass, field
 from typing import Any, Callable
 
@@ -36,6 +38,13 @@ class MediaRegistration:
     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
 
 
 @dataclass(slots=True)
@@ -59,6 +68,15 @@ class MediaActor:
     last_outbound_audio_monotonic: float = 0.0
     first_pcm_logged: bool = False
     keepalive_loop_logged: bool = False
+    early_ack_started: bool = False
+    partial_transcript: str | None = None
+    partial_intent: str | None = None
+    response_plan_id: str | None = None
+    playback_generation: int = 0
+    utterance_generation: int = 0
+    finalized_utterance_generation: int = 0
+    partial_asr_task: asyncio.Task | None = None
+    partial_asr_attempted: bool = False
 
 
 class AudioSocketMediaRuntime:
@@ -83,6 +101,7 @@ class AudioSocketMediaRuntime:
         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],
+        plan_reply: Callable[[str, str, dict[str, Any] | None, str], 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],
@@ -107,12 +126,165 @@ class AudioSocketMediaRuntime:
         self._set_state = set_state
         self._get_pending_greeting = get_pending_greeting
         self._mark_reply_delivered = mark_reply_delivered
+        self._plan_reply = plan_reply
         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 = 650
+
+    @staticmethod
+    def _normalize_intent_text(text: str) -> str:
+        return " ".join(str(text or "").strip().lower().split())
+
+    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 _ack_kind_for_intent(intent: str) -> str:
+        if intent == "operator_request":
+            return "handoff"
+        if intent in {"schedule", "address", "price", "status", "problem"}:
+            return "understanding"
+        return "generic"
+
+    @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 "Қазір айтып шығамын."
+            return "Қазір айтайын."
+        if ack_kind == "handoff":
+            return "Секунду."
+        if ack_kind == "understanding":
+            return "Сейчас сориентирую."
+        return "Сейчас подскажу."
+
+    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
+
+    @staticmethod
+    def _reset_live_turn_state(actor: MediaActor) -> None:
+        actor.utterance_generation += 1
+        actor.finalized_utterance_generation = 0
+        actor.early_ack_started = False
+        actor.partial_transcript = None
+        actor.partial_intent = None
+        actor.response_plan_id = None
+        actor.partial_asr_attempted = False
+        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()
+
+    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
+        actor.partial_intent = self._detect_early_intent(transcript_text)
+        logger.warning(
+            "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 _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,
+        )
 
     @property
     def address(self) -> str:
@@ -296,9 +468,12 @@ class AudioSocketMediaRuntime:
             return
 
         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)
         if vad_result.utterance_pcm:
             await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending))
             actor.barge_in_pending = False
@@ -327,34 +502,123 @@ 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)
-        transcription = await asyncio.to_thread(
-            lambda: self._asr_provider.transcribe(
-                wav_bytes,
-                language_hint=actor.registration.language,
+        actor.playback_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,
+                )
             )
         )
-        transcript_text = str(transcription.text or "").strip()
+        if self._should_use_voice_v2(actor.registration) and partial_transcript and not actor.early_ack_started:
+            ack_kind = self._ack_kind_for_intent(partial_intent or "unknown")
+            ack_text = self._ack_text(actor.registration.language, ack_kind)
+            actor.early_ack_started = True
+            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,
+                "partial_transcript": partial_transcript,
+                "early_intent": partial_intent,
+            }
+            await self._plan_reply_segment(
+                actor,
+                ack_text,
+                kind="ack",
+                metadata={
+                    **base_metadata,
+                    "ack_kind": ack_kind,
+                    "phase": "ack",
+                    "partial_ack_source": "precomputed_partial_asr",
+                },
+            )
+            await self._speak_text(actor, ack_text, is_greeting=False)
+            if not actor.closed:
+                await self._set_actor_state(actor, "thinking")
+
+        transcription = await full_asr_task
+        transcript_text = str(transcription.text or "").strip() or partial_transcript
         if not transcript_text:
             await self._set_actor_state(actor, "listening")
             return
 
+        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)
         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,
+            "partial_transcript": actor.partial_transcript,
+            "early_intent": actor.partial_intent,
         }
-        decision = await asyncio.to_thread(
-            self._process_turn,
-            actor.registration.voice_session_id,
-            transcript_text,
-            transcription.language or actor.registration.language,
-            barge_in,
-            metadata,
+        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),
+                },
+            )
         )
+        if self._should_use_voice_v2(actor.registration):
+            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:
+                    ack_kind = self._ack_kind_for_intent(actor.partial_intent or "unknown")
+                    ack_text = self._ack_text(transcription.language or actor.registration.language, ack_kind)
+                    actor.early_ack_started = True
+                    await self._plan_reply_segment(
+                        actor,
+                        ack_text,
+                        kind="ack",
+                        metadata={
+                            **metadata,
+                            "partial_transcript": actor.partial_transcript,
+                            "early_intent": actor.partial_intent,
+                            "ack_kind": ack_kind,
+                            "phase": "ack",
+                        },
+                    )
+                    await self._speak_text(actor, ack_text, is_greeting=False)
+                    if not actor.closed:
+                        await self._set_actor_state(actor, "thinking")
+                decision = await decision_task
+        else:
+            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):
+                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_text(actor, decision.reply_text, is_greeting=False)
         if actor.closed:
             return
@@ -393,46 +657,86 @@ class AudioSocketMediaRuntime:
         actor.handoff_task = asyncio.create_task(_run())
         return actor.handoff_task
 
+    async def _stream_tts_chunks(self, actor: MediaActor, text: str):
+        if not actor.registration.voice_v2_streaming_tts:
+            synthesis = await asyncio.to_thread(
+                lambda: self._tts_provider.synthesize(
+                    text,
+                    language=actor.registration.language,
+                )
+            )
+            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):
+                    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) -> None:
         if actor.closed or not text:
             return
         await self._set_actor_state(actor, "speaking")
         synth_started_at = time.monotonic()
-        synthesis = await asyncio.to_thread(
-            lambda: self._tts_provider.synthesize(
-                text,
-                language=actor.registration.language,
-            )
-        )
-        if not synthesis.audio_bytes:
-            raise RuntimeError("TTS provider returned empty audio")
-        logger.warning(
-            "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),
-            len(synthesis.audio_bytes),
-        )
-
-        pcm_8k = resample_pcm16le(
-            synthesis.audio_bytes,
-            input_rate_hz=synthesis.sample_rate_hz,
-            output_rate_hz=8000,
-        )
         first_frame_sent = False
-        for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes):
-            if actor.closed or actor.playback_interrupt.is_set():
-                break
-            await self._write_audio_packet(actor, frame)
+        total_audio_bytes = 0
+        async for synthesis in self._stream_tts_chunks(actor, text):
+            if not synthesis.audio_bytes:
+                continue
+            total_audio_bytes += len(synthesis.audio_bytes)
             if not first_frame_sent:
-                first_frame_sent = True
                 logger.warning(
-                    "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s",
+                    "audiosocket.tts_ready session_id=%s greeting=%s synth_ms=%s audio_bytes=%s",
                     actor.registration.voice_session_id,
                     is_greeting,
-                    len(frame),
+                    int((time.monotonic() - synth_started_at) * 1000),
+                    total_audio_bytes,
                 )
-            await asyncio.sleep(actor.frame_ms / 1000.0)
+            pcm_8k = resample_pcm16le(
+                synthesis.audio_bytes,
+                input_rate_hz=synthesis.sample_rate_hz,
+                output_rate_hz=8000,
+            )
+            for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes):
+                if actor.closed or actor.playback_interrupt.is_set():
+                    break
+                await self._write_audio_packet(actor, frame)
+                if not first_frame_sent:
+                    first_frame_sent = True
+                    logger.warning(
+                        "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s",
+                        actor.registration.voice_session_id,
+                        is_greeting,
+                        len(frame),
+                    )
+                await asyncio.sleep(actor.frame_ms / 1000.0)
+            if actor.closed or actor.playback_interrupt.is_set():
+                break
+
+        if total_audio_bytes <= 0:
+            raise RuntimeError("TTS provider returned empty audio")
 
         interrupted = actor.playback_interrupt.is_set()
         actor.playback_interrupt.clear()
@@ -529,6 +833,10 @@ class AudioSocketMediaRuntime:
             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
         if actor.worker_task is not None:
             actor.worker_task.cancel()
             with contextlib.suppress(asyncio.CancelledError, Exception):

services/ai_voice_runtime_service/providers/asr.py

@@ -41,6 +41,9 @@ class ASRProvider:
         del audio_bytes
         return ASRTranscription(text="", language=language_hint, confidence=None)
 
+    def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+        return self.transcribe(audio_bytes, language_hint=language_hint)
+
 
 class OpenAIASRProvider(ASRProvider):
     name = "openai"
@@ -77,6 +80,9 @@ class OpenAIASRProvider(ASRProvider):
             confidence=None,
         )
 
+    def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+        return self.transcribe(audio_bytes, language_hint=language_hint)
+
 
 def build_asr_provider(name: str) -> ASRProvider:
     normalized = str(name or "stub").strip().lower()

services/ai_voice_runtime_service/providers/tts.py

@@ -154,6 +154,11 @@ class TTSProvider:
         del language
         return TTSSynthesis(text=text, audio_bytes=b"", sample_rate_hz=8000)
 
+    def synthesize_chunks(self, text: str, *, language: str | None = None):
+        synthesis = self.synthesize(text, language=language)
+        if synthesis.audio_bytes:
+            yield synthesis
+
 
 class OpenAITTSProvider(TTSProvider):
     name = "openai"

tests/test_ai_orchestrator_service.py

@@ -1300,6 +1300,62 @@ def test_voice_llm_guarded_decision_uses_operator_style_without_ai_or_kb(monkeyp
     assert "Do not say or imply that you are an AI" in system_prompt
 
 
+def test_voice_v2_fast_conversational_adds_ack_metadata_and_compacts_reply(monkeypatch):
+    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_fast_conversational")
+    monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible")
+
+    def _fake_structured(messages):
+        del messages
+        return {
+            "language": "ru",
+            "intent": "kb_answer",
+            "reply_text": (
+                "Сейчас сориентирую по графику работы филиала в Алматы. "
+                "Он работает с понедельника по пятницу с 9:00 до 18:00. "
+                "Если нужно, подскажу и по субботе."
+            ),
+            "confidence": 0.91,
+            "needs_handoff": False,
+            "handoff_reason": None,
+            "case_action": "keep_open",
+            "kb_refs": ["kba_voice_v2"],
+            "_model": "gpt-test",
+            "_latency_ms": 35,
+            "_finish_reason": "stop",
+        }
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _fake_structured)
+
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=SimpleNamespace(customer_id="cus_voice_v2", display_name="Ернор"),
+        interaction=SimpleNamespace(interaction_id="int_voice_v2", status="new", queue_id="que_voice", subject="hours"),
+        transcript_text="Хочу узнать график работы филиала в Алматы",
+        transcript_window=[
+            SimpleNamespace(
+                speaker="caller",
+                text="Хочу узнать график работы филиала в Алматы",
+                sequence_no=1,
+                source_type="voice_asr",
+                barge_in_interrupted=False,
+                created_at=utc_now_iso(),
+            )
+        ],
+        kb_results=[SimpleNamespace(article_id="kba_voice_v2", title="График", body="Будни 9:00-18:00")],
+        disclosure_required=False,
+        customer_name_value="Ернор",
+        customer_name_status="name_obtained",
+        request_metadata={"voice_v2_enabled": True, "response_plan_id": "rsp_test"},
+    )
+
+    assert decision["intent"] == "kb_answer"
+    assert decision["metadata"]["voice_v2_enabled"] is True
+    assert decision["metadata"]["early_intent"] == "schedule"
+    assert decision["metadata"]["ack_kind"] == "understanding"
+    assert decision["metadata"]["response_plan_id"] == "rsp_test"
+    assert len(decision["reply_text"]) <= 180
+
+
 def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch):
     monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
     monkeypatch.setenv("AI_PROVIDER", "stub")

tests/test_ai_voice_media_runtime.py

@@ -107,6 +107,7 @@ def test_media_runtime_streams_greeting_and_turn():
         set_state=lambda session_id, state, handoff_reason, metadata: states.append((session_id, state)),
         get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime" else None,
         mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
+        plan_reply=lambda session_id, text, metadata, kind: None,
         process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
             turns.append((session_id, transcript_text, barge_in))
             or VoiceAITurnDecisionOut(
@@ -201,6 +202,7 @@ def test_media_runtime_waits_for_late_registration():
         set_state=lambda session_id, state, handoff_reason, metadata: None,
         get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_late" else None,
         mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
+        plan_reply=lambda session_id, text, metadata, kind: None,
         process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
             language=language or "ru",
             intent="answer",
@@ -303,6 +305,7 @@ def test_media_runtime_sends_keepalive_while_tts_is_slow():
         set_state=lambda session_id, state, handoff_reason, metadata: None,
         get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_keepalive" else None,
         mark_reply_delivered=lambda session_id, text, is_greeting: None,
+        plan_reply=lambda session_id, text, metadata, kind: None,
         process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
             language=language or "ru",
             intent="answer",
@@ -369,6 +372,7 @@ def test_media_runtime_sends_keepalive_before_registration_is_ready():
         set_state=lambda session_id, state, handoff_reason, metadata: None,
         get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_prereg" else None,
         mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
+        plan_reply=lambda session_id, text, metadata, kind: None,
         process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
             language=language or "ru",
             intent="answer",
@@ -453,6 +457,7 @@ def test_media_runtime_starts_handoff_before_handoff_tts_finishes():
         set_state=lambda session_id, state, handoff_reason, metadata: None,
         get_pending_greeting=lambda session_id: None,
         mark_reply_delivered=lambda session_id, text, is_greeting: None,
+        plan_reply=lambda session_id, text, metadata, kind: None,
         process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
             language=language or "ru",
             intent="handoff_request",
@@ -512,3 +517,208 @@ def test_media_runtime_starts_handoff_before_handoff_tts_finishes():
 
     assert handoff_started.is_set()
     assert events.index("handoff") < events.index("speak_end")
+
+
+def test_media_runtime_voice_v2_plays_short_ack_before_main_reply():
+    planned: list[tuple[str, str, str, dict | None]] = []
+    delivered: list[tuple[str, str, bool]] = []
+
+    class _ScheduleASRProvider(ASRProvider):
+        name = "schedule-asr"
+
+        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+            del audio_bytes
+            return ASRTranscription(text="Хочу узнать график работы", language=language_hint or "ru", confidence=0.9)
+
+    runtime = AudioSocketMediaRuntime(
+        enabled=True,
+        host="127.0.0.1",
+        port=0,
+        frame_ms=20,
+        idle_timeout_seconds=2.0,
+        registration_wait_timeout_seconds=0.5,
+        min_speech_ms=40,
+        trailing_silence_ms=40,
+        max_turn_ms=400,
+        asr_provider=_ScheduleASRProvider(),
+        tts_provider=_StubTTSProvider(),
+        load_registration_by_media_uuid=lambda value: None,
+        mark_media_connected=lambda session_id, value: None,
+        mark_media_ended=lambda session_id, reason: None,
+        touch_media_frame=lambda session_id: None,
+        set_state=lambda session_id, state, handoff_reason, metadata: None,
+        get_pending_greeting=lambda session_id: None,
+        mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
+        plan_reply=lambda session_id, text, metadata, kind: planned.append((session_id, text, kind, metadata)),
+        process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
+            time.sleep(0.25)
+            or VoiceAITurnDecisionOut(
+                language=language or "ru",
+                intent="clarification",
+                reply_text="Подскажите, пожалуйста, какой город вас интересует?",
+                confidence=0.9,
+                needs_handoff=False,
+                handoff_reason=None,
+                case_action="keep_open",
+                kb_refs=[],
+                summary_text="reply ready",
+                model="stub-voice",
+                latency_ms=1,
+                status="active",
+                metadata={"early_intent": "schedule", "ack_kind": "understanding"},
+            )
+        ),
+        request_handoff=lambda session_id, customer_request_text, decision: None,
+        handle_media_error=lambda session_id, message, metadata: None,
+    )
+
+    async def _fake_write_audio_packet(current_actor, pcm_frame: bytes) -> None:
+        del current_actor, pcm_frame
+
+    runtime._write_audio_packet = _fake_write_audio_packet  # type: ignore[method-assign]
+    pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+
+    async def _scenario() -> None:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_v2",
+                call_id="call_media_runtime_v2",
+                interaction_id="int_media_runtime_v2",
+                ai_session_id="ais_media_runtime_v2",
+                language="ru",
+                media_uuid=str(uuid.uuid4()),
+                queue_code="voice_lab_ai",
+                queue_id="que_voice_lab_ai",
+                agent_profile="voice_support",
+                voice_v2_enabled=True,
+                voice_v2_ack_mode="immediate_short",
+                voice_v2_streaming_tts=True,
+                voice_v2_partial_asr=False,
+            ),
+            reader=asyncio.StreamReader(),
+            writer=None,  # type: ignore[arg-type]
+            vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=400),
+            frame_ms=20,
+            frame_bytes=320,
+        )
+        await runtime._process_utterance(actor, pcm_frame, False)
+
+    asyncio.run(_scenario())
+
+    assert [item[2] for item in planned] == ["ack", "reply"]
+    assert planned[0][1] == "Сейчас сориентирую."
+    assert planned[1][1].startswith("Подскажите, пожалуйста")
+    assert delivered == [
+        ("avs_media_runtime_v2", "Сейчас сориентирую.", False),
+        ("avs_media_runtime_v2", "Подскажите, пожалуйста, какой город вас интересует?", False),
+    ]
+
+
+def test_media_runtime_voice_v2_uses_partial_asr_to_start_ack_before_full_asr():
+    timings: dict[str, float] = {}
+    speak_events: list[tuple[str, float]] = []
+
+    class _PartialAwareASRProvider(ASRProvider):
+        name = "partial-aware-asr"
+
+        def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+            assert audio_bytes
+            timings["partial_ready"] = time.monotonic()
+            return ASRTranscription(text="work schedule", language=language_hint or "ru", confidence=0.8)
+
+        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+            assert audio_bytes
+            timings["full_started"] = time.monotonic()
+            time.sleep(0.35)
+            timings["full_finished"] = time.monotonic()
+            return ASRTranscription(text="work schedule", language=language_hint or "ru", confidence=0.9)
+
+    runtime = AudioSocketMediaRuntime(
+        enabled=True,
+        host="127.0.0.1",
+        port=0,
+        frame_ms=20,
+        idle_timeout_seconds=2.0,
+        registration_wait_timeout_seconds=0.5,
+        min_speech_ms=40,
+        trailing_silence_ms=40,
+        max_turn_ms=2000,
+        asr_provider=_PartialAwareASRProvider(),
+        tts_provider=_StubTTSProvider(),
+        load_registration_by_media_uuid=lambda value: None,
+        mark_media_connected=lambda session_id, value: None,
+        mark_media_ended=lambda session_id, reason: None,
+        touch_media_frame=lambda session_id: None,
+        set_state=lambda session_id, state, handoff_reason, metadata: None,
+        get_pending_greeting=lambda session_id: None,
+        mark_reply_delivered=lambda session_id, text, is_greeting: None,
+        plan_reply=lambda session_id, text, metadata, kind: None,
+        process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
+            language=language or "ru",
+            intent="clarification",
+            reply_text="Подскажите, какой город вас интересует?",
+            confidence=0.9,
+            needs_handoff=False,
+            handoff_reason=None,
+            case_action="keep_open",
+            kb_refs=[],
+            summary_text="reply ready",
+            model="stub-voice",
+            latency_ms=1,
+            status="active",
+        ),
+        request_handoff=lambda session_id, customer_request_text, decision: None,
+        handle_media_error=lambda session_id, message, metadata: None,
+    )
+
+    async def _fake_speak_text(current_actor, text: str, *, is_greeting: bool) -> None:
+        del current_actor, is_greeting
+        speak_events.append((text, time.monotonic()))
+        await asyncio.sleep(0)
+
+    runtime._speak_text = _fake_speak_text  # type: ignore[method-assign]
+
+    async def _scenario() -> None:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_v2_partial",
+                call_id="call_media_runtime_v2_partial",
+                interaction_id="int_media_runtime_v2_partial",
+                ai_session_id="ais_media_runtime_v2_partial",
+                language="ru",
+                media_uuid=str(uuid.uuid4()),
+                queue_code="voice_lab_ai",
+                queue_id="que_voice_lab_ai",
+                agent_profile="voice_support",
+                voice_v2_enabled=True,
+                voice_v2_ack_mode="immediate_short",
+                voice_v2_streaming_tts=True,
+                voice_v2_partial_asr=True,
+            ),
+            reader=asyncio.StreamReader(),
+            writer=None,  # type: ignore[arg-type]
+            vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
+            frame_ms=20,
+            frame_bytes=320,
+            state="listening",
+        )
+        speech_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        silence_frame = b"\x00\x00" * 160
+        for _ in range(35):
+            await runtime._handle_pcm(actor, speech_frame)
+        for _ in range(20):
+            if actor.partial_transcript:
+                break
+            await asyncio.sleep(0.01)
+        assert actor.partial_transcript == "work schedule"
+        for _ in range(2):
+            await runtime._handle_pcm(actor, silence_frame)
+        pcm_bytes, barge_in = await asyncio.wait_for(actor.turn_queue.get(), timeout=0.2)
+        await runtime._process_utterance(actor, pcm_bytes, barge_in)
+
+    asyncio.run(_scenario())
+
+    assert timings["partial_ready"] < timings["full_finished"]
+    assert speak_events
+    assert speak_events[0][0] == runtime._ack_text("ru", "understanding")
+    assert speak_events[0][1] < timings["full_finished"]