import json import threading import time from fastapi.testclient import TestClient from sqlalchemy import select import services.ai_voice_runtime_service.app as runtime_module from services.shared.core import utc_now_iso from services.shared.db import get_session from services.shared.models import VoiceAIStartIn from services.shared.sql_models import VoiceAISessionRow, VoiceTranscriptSegmentRow, VoiceTTSSettingsRow from services.shared.voice_tts_config import save_voice_tts_config, voice_tts_default_config def _admin_headers() -> dict[str, str]: return {"X-User": "admin", "X-Role": "admin"} def _reset_voice_tts_settings() -> None: session = get_session() try: row = session.execute(select(VoiceTTSSettingsRow)).scalar_one_or_none() if row is not None: session.delete(row) session.commit() finally: session.close() def test_create_voice_ai_session_records_greeting_without_blocking_on_orchestrator(monkeypatch): allow_start = threading.Event() def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0): assert method == "POST" assert path.endswith("/start") assert allow_start.wait(timeout=2.0) return { "session_id": "ais_voice_demo", "language": "ru", "greeting_text": "Я AI-оператор компании. Здравствуйте.", "disclosure_required": True, } monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request) client = TestClient(runtime_module.app) response_holder: dict[str, object] = {} def _request() -> None: response_holder["response"] = client.post( "/internal/voice-ai/sessions", headers=_admin_headers(), json={ "call_id": "call_voice_runtime_1", "linked_id": "linked_voice_runtime_1", "interaction_id": "int_voice_runtime_1", "queue_id": "que_voice_runtime_1", "caller_number": "+77010000031", "caller_name": "Runtime Caller", "agent_profile": "voice_support", "language_hint": "ru", "handoff_queue_id": "que_voice_runtime_1", "metadata": {"queue_code": "voice_lab"}, }, ) request_thread = threading.Thread(target=_request, daemon=True) request_thread.start() request_thread.join(timeout=0.3) assert not request_thread.is_alive() response = response_holder["response"] assert response.status_code == 200 payload = response.json() assert payload["ai_session_id"] is None assert payload["status"] == "greeting" session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1") ).scalar_one() assert voice_session.status == "greeting" assert voice_session.disclosure_played_at is None assert voice_session.last_ai_reply_at is None greeting_segment = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id) .where(VoiceTranscriptSegmentRow.speaker == "assistant") ).scalar_one() assert "AI-оператор" in greeting_segment.text assert greeting_segment.is_final is False finally: session.close() allow_start.set() deadline = time.time() + 2.0 while time.time() < deadline: session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1") ).scalar_one() if voice_session.ai_session_id == "ais_voice_demo": greetings = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id) .where(VoiceTranscriptSegmentRow.speaker == "assistant") ).scalars().all() assert len(greetings) == 1 break finally: session.close() time.sleep(0.05) else: raise AssertionError("background voice-session start did not finish in time") def test_create_voice_ai_session_records_greeting_without_blocking_on_orchestrator(monkeypatch): allow_start = threading.Event() def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0): assert method == "POST" assert path.endswith("/start") assert allow_start.wait(timeout=2.0) return { "session_id": "ais_voice_demo", "language": "ru", "greeting_text": "Здравствуйте. Подскажите, пожалуйста, чем помочь.", "disclosure_required": True, } monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request) client = TestClient(runtime_module.app) response_holder: dict[str, object] = {} def _request() -> None: response_holder["response"] = client.post( "/internal/voice-ai/sessions", headers=_admin_headers(), json={ "call_id": "call_voice_runtime_1", "linked_id": "linked_voice_runtime_1", "interaction_id": "int_voice_runtime_1", "queue_id": "que_voice_runtime_1", "caller_number": "+77010000031", "caller_name": "Runtime Caller", "agent_profile": "voice_support", "language_hint": "ru", "handoff_queue_id": "que_voice_runtime_1", "metadata": {"queue_code": "voice_lab"}, }, ) request_thread = threading.Thread(target=_request, daemon=True) request_thread.start() request_thread.join(timeout=0.3) assert not request_thread.is_alive() response = response_holder["response"] assert response.status_code == 200 payload = response.json() assert payload["ai_session_id"] is None assert payload["status"] == "greeting" session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1") ).scalar_one() assert voice_session.status == "greeting" assert voice_session.disclosure_played_at is None assert voice_session.last_ai_reply_at is None greeting_segment = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id) .where(VoiceTranscriptSegmentRow.speaker == "assistant") ).scalar_one() assert "ai" not in greeting_segment.text.lower() assert greeting_segment.is_final is False finally: session.close() allow_start.set() deadline = time.time() + 2.0 while time.time() < deadline: session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1") ).scalar_one() if voice_session.ai_session_id == "ais_voice_demo": greetings = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id) .where(VoiceTranscriptSegmentRow.speaker == "assistant") ).scalars().all() assert len(greetings) == 1 break finally: session.close() time.sleep(0.05) else: raise AssertionError("background voice-session start did not finish in time") def test_create_voice_ai_session_uses_database_selected_tts_provider(monkeypatch): _reset_voice_tts_settings() session = get_session() try: payload = voice_tts_default_config().model_dump() payload["provider"] = "elevenlabs" save_voice_tts_config(session, payload) finally: session.close() def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0): assert method == "POST" assert path.endswith("/start") return { "session_id": "ais_voice_tts_runtime", "language": "ru", "greeting_text": "Здравствуйте. Чем помочь?", "disclosure_required": False, } monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request) client = TestClient(runtime_module.app) response = client.post( "/internal/voice-ai/sessions", headers=_admin_headers(), json={ "call_id": "call_voice_runtime_tts_provider", "linked_id": "linked_voice_runtime_tts_provider", "interaction_id": "int_voice_runtime_tts_provider", "queue_id": "que_voice_runtime_tts_provider", "caller_number": "+77010000032", "caller_name": "Runtime Caller", "agent_profile": "voice_support", "language_hint": "ru", "handoff_queue_id": "que_voice_runtime_tts_provider", "metadata": {"queue_code": "voice_lab_ai"}, }, ) assert response.status_code == 200 session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_tts_provider") ).scalar_one() assert voice_session.tts_provider == "elevenlabs" finally: session.close() _reset_voice_tts_settings() def test_mark_reply_delivered_finalizes_only_matching_pending_assistant_segment(): now = utc_now_iso() session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_runtime_delivery", call_id="call_runtime_delivery", linked_id="linked_runtime_delivery", interaction_id="int_runtime_delivery", customer_id=None, queue_id="que_runtime_delivery", ai_session_id="ais_runtime_delivery", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="speaking", handoff_reason=None, handoff_target_queue_id="que_runtime_delivery", disclosure_played_at=None, last_user_utterance_at=None, last_ai_reply_at=None, started_at=now, updated_at=now, ended_at=None, ) ) session.add( VoiceTranscriptSegmentRow( segment_id="vts_runtime_delivery_greeting", session_id="avs_runtime_delivery", call_id="call_runtime_delivery", interaction_id="int_runtime_delivery", speaker="assistant", source_type="tts", sequence_no=1, text="greeting", confidence=None, is_final=False, barge_in_interrupted=False, payload_json=json.dumps({"kind": "greeting", "delivery_status": "planned"}), created_at=now, ) ) session.add( VoiceTranscriptSegmentRow( segment_id="vts_runtime_delivery_pending", session_id="avs_runtime_delivery", call_id="call_runtime_delivery", interaction_id="int_runtime_delivery", speaker="assistant", source_type="tts", sequence_no=2, text="reply that was spoken", confidence=None, is_final=False, barge_in_interrupted=False, payload_json=json.dumps({"delivery_status": "planned"}), created_at=now, ) ) session.commit() finally: session.close() runtime_module._mark_reply_delivered( "avs_runtime_delivery", "reply that was spoken", False, ) session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_runtime_delivery") ).scalar_one() segments = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == "avs_runtime_delivery") .order_by(VoiceTranscriptSegmentRow.sequence_no.asc()) ).scalars().all() assert voice_session.last_ai_reply_at is not None assert [segment.is_final for segment in segments] == [False, True] payloads = [json.loads(segment.payload_json or "{}") for segment in segments] assert payloads[0]["delivery_status"] == "planned" assert payloads[1]["delivery_status"] == "delivered" finally: session.close() def test_process_voice_ai_turn_triggers_bridge_handoff(monkeypatch): handoffs: list[dict] = [] def _fake_bridge_request(method: str, path: str, *, payload=None, timeout=10.0): handoffs.append({"method": method, "path": path, "payload": payload, "timeout": timeout}) return {"ok": True} def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0): assert method == "POST" assert path.endswith("/turns") return { "language": "ru", "intent": "handoff_request", "reply_text": "Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора.", "confidence": 0.2, "needs_handoff": True, "handoff_reason": "Нужен живой оператор.", "case_action": "keep_open", "kb_refs": [], "summary_text": "AI собрал контекст и запросил человека.", "model": "stub-voice", "latency_ms": 1, "status": "handoff_requested", "metadata": { "voice_start_language": "ru", "customer_name_status": "name_obtained", "customer_name_value": "Айдос", "customer_name_source": "voice_followup", }, } monkeypatch.setattr(runtime_module, "_bridge_request", _fake_bridge_request) monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request) session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_runtime_handoff", call_id="call_voice_runtime_handoff", linked_id="linked_voice_runtime_handoff", interaction_id="int_voice_runtime_handoff", customer_id=None, queue_id="que_voice_runtime_handoff", ai_session_id="ais_runtime_handoff", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="listening", handoff_reason=None, handoff_target_queue_id="que_voice_runtime_handoff", disclosure_played_at=utc_now_iso(), last_user_utterance_at=None, last_ai_reply_at=None, started_at=utc_now_iso(), updated_at=utc_now_iso(), ended_at=None, ) ) session.commit() finally: session.close() client = TestClient(runtime_module.app) response = client.post( "/internal/voice-ai/sessions/avs_runtime_handoff/turns", headers=_admin_headers(), json={ "voice_session_id": "avs_runtime_handoff", "call_id": "call_voice_runtime_handoff", "interaction_id": "int_voice_runtime_handoff", "transcript_text": "Соедините с оператором", "language": "ru", "sequence_no": 1, "barge_in": False, "metadata": {"turn_duration_ms": 3200}, }, ) assert response.status_code == 200 assert response.json()["needs_handoff"] is True assert handoffs assert handoffs[0]["path"] == "/internal/voice-ai/calls/call_voice_runtime_handoff/handoff" assert handoffs[0]["payload"]["summary"]["customer_name_status"] == "name_obtained" assert handoffs[0]["payload"]["summary"]["customer_name_value"] == "Айдос" assert handoffs[0]["payload"]["summary"]["customer_name_source"] == "voice_followup" assert handoffs[0]["payload"]["reason"] == "Нужен живой оператор." session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_runtime_handoff") ).scalar_one() assert voice_session.status == "handoff_requested" segments = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == "avs_runtime_handoff") .order_by(VoiceTranscriptSegmentRow.sequence_no.asc()) ).scalars().all() assert [item.speaker for item in segments] == ["caller", "assistant"] assert [item.is_final for item in segments] == [True, False] finally: session.close() def test_register_media_bridge_updates_voice_session(monkeypatch): closed: list[tuple[str, str]] = [] monkeypatch.setattr( runtime_module._MEDIA_RUNTIME, "close_session_sync", lambda session_id, *, reason: closed.append((session_id, reason)), ) session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_media_bridge_runtime", call_id="call_media_bridge_runtime", linked_id="linked_media_bridge_runtime", interaction_id="int_media_bridge_runtime", customer_id=None, queue_id="que_media_bridge_runtime", ai_session_id="ais_media_bridge_runtime", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="greeting", handoff_reason=None, handoff_target_queue_id="que_media_bridge_runtime", media_uuid=None, media_status=None, media_connected_at=None, media_ended_at=None, last_media_frame_at=None, disclosure_played_at=None, last_user_utterance_at=None, last_ai_reply_at=None, started_at=utc_now_iso(), updated_at=utc_now_iso(), ended_at=None, ) ) session.commit() finally: session.close() def test_complete_voice_ai_session_start_persists_voice_start_result_and_requests_handoff(monkeypatch): bridge_calls: list[dict] = [] def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0): assert method == "POST" assert path.endswith("/start") return { "session_id": "ais_voice_start_done", "language": "kz", "greeting_text": "", "disclosure_required": False, "needs_handoff": True, "handoff_reason": "voice_start_completed", "summary_text": "Voice start completed with status name_followup_required and candidate name Айдос.", "start_result": { "language": "kz", "customer_id": "cus_voice_start", "customer_name_status": "name_followup_required", "customer_name_value": "Айдос", "customer_name_source": "voice_start", "downstream_queue_id": "que_voice_support_kz", "downstream_queue_code": "voice_support_kz", "resolved_at": "2040-01-01T10:00:00+00:00", }, } def _fake_bridge_request(method: str, path: str, *, payload=None, timeout=10.0): bridge_calls.append({"method": method, "path": path, "payload": payload, "timeout": timeout}) return {"ok": True} monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request) monkeypatch.setattr(runtime_module, "_bridge_request", _fake_bridge_request) now = utc_now_iso() session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_voice_start_complete", call_id="call_voice_start_complete", linked_id="linked_voice_start_complete", interaction_id="int_voice_start_complete", customer_id=None, queue_id="que_voice_start_kz", ai_session_id=None, agent_profile="voice_start", language="kz", asr_provider="openai", tts_provider="openai", status="greeting", handoff_reason=None, handoff_target_queue_id="que_voice_support_kz", disclosure_played_at=None, last_user_utterance_at=None, last_ai_reply_at=None, started_at=now, updated_at=now, ended_at=None, ) ) session.commit() finally: session.close() runtime_module._complete_voice_ai_session_start( "avs_voice_start_complete", VoiceAIStartIn( voice_session_id="avs_voice_start_complete", call_id="call_voice_start_complete", interaction_id="int_voice_start_complete", customer_id=None, language_hint="kz", agent_profile="voice_start", metadata={"stage": "voice_start", "next_queue_code": "voice_support_kz"}, ), ) session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_voice_start_complete") ).scalar_one() assert voice_session.ai_session_id == "ais_voice_start_done" assert voice_session.status == "handoff_requested" assert voice_session.voice_start_language == "kz" assert voice_session.customer_name_status == "name_followup_required" assert voice_session.customer_name_value == "Айдос" assert voice_session.customer_name_source == "voice_start" finally: session.close() relevant_calls = [call for call in bridge_calls if "call_voice_start_complete" in call["path"]] assert [call["path"] for call in relevant_calls] == [ "/internal/voice-ai/calls/call_voice_start_complete/state", "/internal/voice-ai/calls/call_voice_start_complete/handoff", ] assert relevant_calls[1]["payload"]["metadata"]["customer_name_status"] == "name_followup_required" assert relevant_calls[1]["payload"]["metadata"]["customer_name_value"] == "Айдос" def test_complete_voice_ai_session_start_handles_disabled_name_collection_without_greeting(monkeypatch): bridge_calls: list[dict] = [] def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0): assert method == "POST" assert path.endswith("/start") return { "session_id": "ais_voice_start_disabled", "language": "ru", "greeting_text": "", "disclosure_required": False, "needs_handoff": True, "handoff_reason": "voice_start_completed", "summary_text": "Voice start name collection is disabled and handed off without a name.", "start_result": { "language": "ru", "customer_id": None, "customer_name_status": "name_not_obtained", "customer_name_value": None, "customer_name_source": "none", "downstream_queue_id": "que_voice_support_disabled", "downstream_queue_code": "voice_support_disabled", "resolved_at": "2040-01-01T11:00:00+00:00", }, } def _fake_bridge_request(method: str, path: str, *, payload=None, timeout=10.0): bridge_calls.append({"method": method, "path": path, "payload": payload, "timeout": timeout}) return {"ok": True} monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request) monkeypatch.setattr(runtime_module, "_bridge_request", _fake_bridge_request) now = utc_now_iso() session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_voice_start_disabled", call_id="call_voice_start_disabled", linked_id="linked_voice_start_disabled", interaction_id="int_voice_start_disabled", customer_id=None, queue_id="que_voice_start_disabled", ai_session_id=None, agent_profile="voice_start", language="ru", asr_provider="openai", tts_provider="openai", status="greeting", handoff_reason=None, handoff_target_queue_id="que_voice_support_disabled", disclosure_played_at=None, last_user_utterance_at=None, last_ai_reply_at=None, started_at=now, updated_at=now, ended_at=None, ) ) session.commit() finally: session.close() runtime_module._complete_voice_ai_session_start( "avs_voice_start_disabled", VoiceAIStartIn( voice_session_id="avs_voice_start_disabled", call_id="call_voice_start_disabled", interaction_id="int_voice_start_disabled", customer_id=None, language_hint="ru", agent_profile="voice_start", metadata={"stage": "voice_start", "next_queue_code": "voice_support_disabled"}, ), ) session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_voice_start_disabled") ).scalar_one() assert voice_session.ai_session_id == "ais_voice_start_disabled" assert voice_session.status == "handoff_requested" assert voice_session.customer_name_status == "name_not_obtained" segments = session.execute( select(VoiceTranscriptSegmentRow) .where(VoiceTranscriptSegmentRow.session_id == "avs_voice_start_disabled") .where(VoiceTranscriptSegmentRow.speaker == "assistant") ).scalars().all() assert segments == [] finally: session.close() relevant_calls = [call for call in bridge_calls if "call_voice_start_disabled" in call["path"]] assert [call["path"] for call in relevant_calls] == [ "/internal/voice-ai/calls/call_voice_start_disabled/state", "/internal/voice-ai/calls/call_voice_start_disabled/handoff", ] def test_register_media_bridge_updates_voice_session_round_trip(monkeypatch): closed: list[tuple[str, str]] = [] monkeypatch.setattr( runtime_module._MEDIA_RUNTIME, "close_session_sync", lambda session_id, *, reason: closed.append((session_id, reason)), ) session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_media_bridge_runtime_rt", call_id="call_media_bridge_runtime_rt", linked_id="linked_media_bridge_runtime_rt", interaction_id="int_media_bridge_runtime_rt", customer_id=None, queue_id="que_media_bridge_runtime_rt", ai_session_id="ais_media_bridge_runtime_rt", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="greeting", handoff_reason=None, handoff_target_queue_id="que_media_bridge_runtime_rt", media_uuid=None, media_status=None, media_connected_at=None, media_ended_at=None, last_media_frame_at=None, disclosure_played_at=None, last_user_utterance_at=None, last_ai_reply_at=None, started_at=utc_now_iso(), updated_at=utc_now_iso(), ended_at=None, ) ) session.commit() finally: session.close() client = TestClient(runtime_module.app) requested = client.post( "/internal/voice-ai/sessions/avs_media_bridge_runtime_rt/media-bridge", headers=_admin_headers(), json={ "event_type": "requested", "media_uuid": "75f4d61f-f674-4bb4-91c1-8ddfcf2fc2b4", "call_id": "call_media_bridge_runtime_rt", "linked_id": "linked_media_bridge_runtime_rt", "channel": "PJSIP/1001-00000001", "service_address": "127.0.0.1:9019", }, ) assert requested.status_code == 200 session = get_session() try: voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_media_bridge_runtime_rt") ).scalar_one() assert voice_session.media_uuid == "75f4d61f-f674-4bb4-91c1-8ddfcf2fc2b4" assert voice_session.media_status == "requested" finally: session.close() ended = client.post( "/internal/voice-ai/sessions/avs_media_bridge_runtime_rt/media-bridge", headers=_admin_headers(), json={ "event_type": "ended", "media_uuid": "75f4d61f-f674-4bb4-91c1-8ddfcf2fc2b4", "call_id": "call_media_bridge_runtime_rt", "linked_id": "linked_media_bridge_runtime_rt", "channel": "PJSIP/1001-00000001", "service_address": "127.0.0.1:9019", "reason": "audiosocket_closed", }, ) assert ended.status_code == 200 assert closed == [("avs_media_bridge_runtime_rt", "audiosocket_closed")] def test_push_voice_ai_telephony_event_call_ended_returns_detached_safe_payload(monkeypatch): closed: list[tuple[str, str]] = [] orchestrator_calls: list[tuple[str, str]] = [] monkeypatch.setattr( runtime_module._MEDIA_RUNTIME, "close_session_sync", lambda session_id, *, reason: closed.append((session_id, reason)), ) monkeypatch.setattr( runtime_module, "_orchestrator_request", lambda method, path, *, payload=None, timeout=10.0: orchestrator_calls.append((method, path)) or {"ok": True}, ) session = get_session() try: session.add( VoiceAISessionRow( session_id="avs_runtime_call_ended", call_id="call_voice_runtime_call_ended", linked_id="linked_voice_runtime_call_ended", interaction_id="int_voice_runtime_call_ended", customer_id=None, queue_id="que_voice_runtime_call_ended", ai_session_id="ais_runtime_call_ended", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="listening", handoff_reason=None, handoff_target_queue_id="que_voice_runtime_call_ended", media_uuid="2fc59167-fdc7-43a0-b4e5-c4948f7c24d8", media_status="connected", media_connected_at=utc_now_iso(), media_ended_at=None, last_media_frame_at=utc_now_iso(), disclosure_played_at=utc_now_iso(), last_user_utterance_at=None, last_ai_reply_at=None, started_at=utc_now_iso(), updated_at=utc_now_iso(), ended_at=None, ) ) session.commit() finally: session.close() client = TestClient(runtime_module.app) response = client.post( "/internal/voice-ai/sessions/avs_runtime_call_ended/telephony-events", headers=_admin_headers(), json={ "event_type": "call.ended", "payload": {"hangup_cause": "16"}, }, ) assert response.status_code == 200 assert response.json() == { "voice_session_id": "avs_runtime_call_ended", "ai_session_id": "ais_runtime_call_ended", "status": "completed", } assert closed == [("avs_runtime_call_ended", "call_ended")] assert orchestrator_calls == [("POST", "/ai/voice/sessions/avs_runtime_call_ended/close")] def test_media_registration_emotive_ack_rotation_is_not_gated_on_v2_queue_eligibility(monkeypatch): # AI_VOICE_V2_QUEUE_CODES defaults to "voice_lab_ai" only, so a queue outside # that allowlist runs the plain v1 ack path. Emotive-ack rotation and prebaked-ack # caching must still apply there — otherwise every filler collapses to the single # fixed "Секунду." fallback string instead of rotating, and gets synthesized live # via ElevenLabs each time instead of served from the pre-cached ack bank, adding # real TTS round-trip time to the one phrase whose job is to hide that latency. monkeypatch.delenv("AI_VOICE_V2_QUEUE_CODES", raising=False) monkeypatch.delenv("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", raising=False) monkeypatch.delenv("AI_VOICE_V2_PREBAKED_ACK_ENABLED", raising=False) row = VoiceAISessionRow( session_id="avs_runtime_emotive_ack_v1", call_id="call_runtime_emotive_ack_v1", interaction_id="int_runtime_emotive_ack_v1", ai_session_id="ais_runtime_emotive_ack_v1", language="ru", media_uuid="media-emotive-ack-v1", queue_id="que_not_v2_eligible", agent_profile="voice_support", ) registration = runtime_module._media_registration_from_row(row, queue_code="queue_outside_v2_allowlist") assert registration.voice_v2_enabled is False assert registration.voice_v2_emotive_ack is True assert registration.voice_v2_prebaked_ack is True