import json from datetime import datetime, timedelta, timezone import pytest from fastapi.testclient import TestClient from sqlalchemy import select import services.asterisk_bridge_service.app as bridge_module import services.asterisk_bridge_service.ivr_fastagi as bridge_ivr_fastagi from services.shared.db import get_session from services.shared.core import utc_now_iso from services.shared.models import VoiceAICallStateUpdateIn, VoiceAIHandoffRequestIn from services.shared.sql_models import ( AISessionRow, AsteriskCallActionLogRow, AsteriskCallLinkRow, AsteriskEventLogRow, CallRecordingRow, VoiceAISessionRow, VoiceEventRow, VoiceTranscriptSegmentRow, ) def _admin_headers() -> dict[str, str]: return {"X-User": "admin", "X-Role": "admin"} def _operator_headers(user: str = "operator_a") -> dict[str, str]: return {"X-User": user, "X-Role": "operator"} @pytest.fixture(autouse=True) def _cleanup_bridge_background_threads(): yield bridge_module._shutdown() def _wait_until_stopped(stop_event=None): if stop_event is not None: stop_event.wait(60) def test_asterisk_status_reports_queue_mapping(monkeypatch): monkeypatch.setenv("ASTERISK_BRIDGE_ENABLED", "1") monkeypatch.setenv("ASTERISK_AMI_HOST", "10.10.10.10") monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"sales":"que_sales","support":"que_support"}') monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "bearer_first") monkeypatch.setenv("ASTERISK_WEBRTC_ENABLED", "1") monkeypatch.setenv("ASTERISK_WEBRTC_WS_URL", "wss://pbx.example.test:8089/ws") client = TestClient(bridge_module.app) response = client.get("/asterisk/status", headers=_admin_headers()) assert response.status_code == 200 payload = response.json() assert payload["status"] == "ok" assert payload["ami_host"] == "10.10.10.10" assert payload["queue_codes_loaded"] == ["sales", "support"] assert payload["bridge_auth_mode"] == "bearer_first" assert payload["callcontrol_enabled"] is False assert payload["webrtc_enabled"] is True assert payload["webrtc_ws_url"] == "wss://pbx.example.test:8089/ws" def test_browser_softphone_config_route_returns_mapped_config(monkeypatch): monkeypatch.setenv("ASTERISK_WEBRTC_ENABLED", "1") monkeypatch.setenv("ASTERISK_WEBRTC_WS_URL", "wss://pbx.example.test:8089/ws") monkeypatch.setenv("ASTERISK_AMI_HOST", "pbx.example.test") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator":"2001"}') monkeypatch.setenv( "ASTERISK_BROWSER_SIP_MAP_JSON", '{"operator":{"authorization_username":"2001","password":"secret-2001","display_name":"Operator Browser"}}', ) monkeypatch.setenv( "ASTERISK_WEBRTC_ICE_SERVERS_JSON", '[{"urls":["stun:stun.example.test:3478"]}]', ) client = TestClient(bridge_module.app) response = client.get("/asterisk/browser-softphone/config", headers={"X-User": "operator", "X-Role": "operator"}) assert response.status_code == 200 payload = response.json() assert payload["enabled"] is True assert payload["ws_url"] == "wss://pbx.example.test:8089/ws" assert payload["authorization_username"] == "2001" assert payload["password"] == "secret-2001" assert payload["display_name"] == "Operator Browser" assert payload["operator_extension"] == "2001" assert payload["sip_uri"] == "sip:2001@pbx.example.test" assert payload["ice_servers"] == [{"urls": ["stun:stun.example.test:3478"]}] def test_browser_softphone_config_route_returns_403_when_mapping_missing(monkeypatch): monkeypatch.setenv("ASTERISK_WEBRTC_ENABLED", "1") monkeypatch.setenv("ASTERISK_WEBRTC_WS_URL", "wss://pbx.example.test:8089/ws") monkeypatch.setenv("ASTERISK_BROWSER_SIP_MAP_JSON", "{}") client = TestClient(bridge_module.app) response = client.get("/asterisk/browser-softphone/config", headers={"X-User": "operator", "X-Role": "operator"}) assert response.status_code == 403 assert response.json()["detail"] == "Browser softphone is not configured for this user" def test_post_json_bearer_first_falls_back_to_legacy(monkeypatch): class DummyResponse: def __init__(self, status_code: int, body: dict): self.status_code = status_code self._body = body def raise_for_status(self): if self.status_code >= 400: raise bridge_module.httpx.HTTPStatusError( "error", request=None, response=self, ) def json(self): return self._body class DummyClient: seen_headers = [] def __init__(self, **kwargs): self.kwargs = kwargs def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def request(self, method, url, headers=None, **kwargs): DummyClient.seen_headers.append(headers or {}) if headers and headers.get("Authorization"): return DummyResponse(401, {"detail": "unauthorized"}) return DummyResponse(200, {"ok": True}) monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "bearer_first") monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_FALLBACK_LEGACY", "1") monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_USER", "ast-bridge") monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_ROLE", "admin") monkeypatch.setenv("APP_TOKEN_SECRET", "track9-test-secret") monkeypatch.setattr(bridge_module.httpx, "Client", DummyClient) result = bridge_module._post_json("http://example.internal/endpoint", {"a": 1}) assert result == {"ok": True} assert len(DummyClient.seen_headers) == 2 assert "Authorization" in DummyClient.seen_headers[0] assert DummyClient.seen_headers[1]["X-User"] == "ast-bridge" assert DummyClient.seen_headers[1]["X-Role"] == "admin" def test_assign_interaction_uses_patch(monkeypatch): class DummyResponse: status_code = 200 def raise_for_status(self): return None def json(self): return {"ok": True} class DummyClient: seen: list[dict[str, str]] = [] def __init__(self, **kwargs): self.kwargs = kwargs def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def request(self, method, url, headers=None, **kwargs): DummyClient.seen.append({"method": method, "url": url, "timeout": self.kwargs.get("timeout")}) return DummyResponse() monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "legacy_headers") monkeypatch.setattr(bridge_module.httpx, "Client", DummyClient) result = bridge_module._assign_interaction(interaction_id="int_demo", assignee="operator_a") assert result == {"ok": True} assert DummyClient.seen == [ {"method": "PATCH", "url": "http://localhost:8004/interactions/int_demo/assign", "timeout": 2.5} ] def test_assign_interaction_uses_fast_callcontrol_request_policy(monkeypatch): captured: list[dict[str, object]] = [] def _fake_patch_json(url, payload, **kwargs): captured.append({"url": url, "payload": payload, **kwargs}) return {"ok": True} monkeypatch.setattr(bridge_module, "_patch_json", _fake_patch_json) result = bridge_module._assign_interaction(interaction_id="int_fast", assignee="operator_a") assert result == {"ok": True} assert captured == [ { "url": "http://localhost:8004/interactions/int_fast/assign", "payload": {"assignee": "operator_a"}, "timeout_seconds": 2.5, "max_attempts": 1, "retry_backoff_seconds": 0.0, } ] def test_ivr_fastagi_runtime_completes_and_sets_transfer_target(monkeypatch): class DummyBridge: def _queue_map(self): return {"voice_lab_ivr": "q_ivr", "support": "q_support", "voice_lab": "q_voice"} def _ivr_dtmf_timeout_seconds(self): return 5 def _ivr_max_no_input_retries(self): return 2 def _ivr_max_invalid_retries(self): return 2 class DummyChannel: def __init__(self): self.variables: dict[str, str | None] = {} self.playbacks: list[str] = [] self.waits: list[int] = [] self.verbose_messages: list[str] = [] def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None: self.playbacks.append(prompt_audio_key) return None def wait_for_digit(self, timeout_ms: int) -> str | None: self.waits.append(timeout_ms) return "1" def set_variable(self, name: str, value: str | None) -> None: self.variables[name] = value def verbose(self, message: str, *, level: int = 1) -> None: self.verbose_messages.append(f"{level}:{message}") calls: list[tuple[str, object]] = [] monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge()) monkeypatch.setattr( bridge_ivr_fastagi, "_wait_for_call_link_context", lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"}, ) monkeypatch.setattr( bridge_ivr_fastagi, "_resolve_transfer_extension", lambda queue_code: {"support": "7010", "voice_lab": "7000"}.get(queue_code), ) monkeypatch.setattr( bridge_ivr_fastagi, "_start_session", lambda call_context, interaction_id, queue_id: calls.append(("start", queue_id)) or { "session": {"session_id": "ivs_fastagi"}, "current_node": { "node_id": "root", "prompt_audio_key": "ivr/demo-root", "options": [{"digit": "1", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr( bridge_ivr_fastagi, "_send_dtmf", lambda session_id, digit: calls.append(("dtmf", digit)) or { "session": { "session_id": session_id, "status": "completed", "resolved_queue_id": "q_support", "resolved_queue_code": "support", }, "current_node": {"node_id": "support", "is_terminal": True}, "completed": True, }, ) monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: calls.append(("abandon", session_id))) channel = DummyChannel() request = bridge_ivr_fastagi.AgiRequest( env={}, args=["call_fastagi", "linked_fastagi", "voice_lab_ivr", "voice_lab", "7200"], ) bridge_ivr_fastagi.run_ivr_call(channel, request) assert calls == [("start", "q_ivr"), ("dtmf", "1")] assert channel.playbacks == ["ivr/demo-root"] assert channel.waits == [5000] assert channel.variables["MVPCC_IVR_SESSION_ID"] == "ivs_fastagi" assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "support" assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7010" assert channel.variables["MVPCC_IVR_RESULT"] == "completed" def test_ivr_fastagi_runtime_falls_back_after_no_input_retries(monkeypatch): class DummyBridge: def _queue_map(self): return {"voice_lab_ivr": "q_ivr", "voice_lab": "q_voice"} def _ivr_dtmf_timeout_seconds(self): return 5 def _ivr_max_no_input_retries(self): return 2 def _ivr_max_invalid_retries(self): return 2 class DummyChannel: def __init__(self): self.variables: dict[str, str | None] = {} self.wait_calls = 0 def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None: return None def wait_for_digit(self, timeout_ms: int) -> str | None: self.wait_calls += 1 return None def set_variable(self, name: str, value: str | None) -> None: self.variables[name] = value def verbose(self, message: str, *, level: int = 1) -> None: return None no_input_calls: list[str] = [] abandoned: list[str] = [] monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge()) monkeypatch.setattr( bridge_ivr_fastagi, "_wait_for_call_link_context", lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"}, ) monkeypatch.setattr( bridge_ivr_fastagi, "_resolve_transfer_extension", lambda queue_code: {"voice_lab": "7000"}.get(queue_code), ) monkeypatch.setattr( bridge_ivr_fastagi, "_start_session", lambda call_context, interaction_id, queue_id: { "session": {"session_id": "ivs_no_input"}, "current_node": { "node_id": "root", "prompt_audio_key": "ivr/demo-root", "options": [{"digit": "1", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr( bridge_ivr_fastagi, "_send_no_input", lambda session_id: no_input_calls.append(session_id) or { "session": {"session_id": session_id, "status": "active"}, "current_node": { "node_id": "root", "prompt_audio_key": "ivr/demo-root", "options": [{"digit": "1", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: abandoned.append(session_id)) channel = DummyChannel() request = bridge_ivr_fastagi.AgiRequest( env={}, args=["call_no_input", "linked_no_input", "voice_lab_ivr", "voice_lab", "7200"], ) bridge_ivr_fastagi.run_ivr_call(channel, request) assert no_input_calls == ["ivs_no_input", "ivs_no_input", "ivs_no_input"] assert abandoned == ["ivs_no_input"] assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "voice_lab" assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7000" assert channel.variables["MVPCC_IVR_RESULT"] == "fallback_no_input" def test_ivr_fastagi_runtime_falls_back_after_invalid_digit_retries(monkeypatch): class DummyBridge: def _queue_map(self): return {"voice_lab_ivr": "q_ivr", "voice_lab": "q_voice"} def _ivr_dtmf_timeout_seconds(self): return 5 def _ivr_max_no_input_retries(self): return 2 def _ivr_max_invalid_retries(self): return 2 class DummyChannel: def __init__(self): self.variables: dict[str, str | None] = {} self.digits = ["9", "9", "9"] def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None: return None def wait_for_digit(self, timeout_ms: int) -> str | None: return self.digits.pop(0) def set_variable(self, name: str, value: str | None) -> None: self.variables[name] = value def verbose(self, message: str, *, level: int = 1) -> None: return None dtmf_calls: list[str] = [] abandoned: list[str] = [] monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge()) monkeypatch.setattr( bridge_ivr_fastagi, "_wait_for_call_link_context", lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"}, ) monkeypatch.setattr( bridge_ivr_fastagi, "_resolve_transfer_extension", lambda queue_code: {"voice_lab": "7000"}.get(queue_code), ) monkeypatch.setattr( bridge_ivr_fastagi, "_start_session", lambda call_context, interaction_id, queue_id: { "session": {"session_id": "ivs_invalid"}, "current_node": { "node_id": "root", "prompt_audio_key": "ivr/demo-root", "options": [{"digit": "1", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr( bridge_ivr_fastagi, "_send_dtmf", lambda session_id, digit: dtmf_calls.append(digit) or { "session": {"session_id": session_id, "status": "active"}, "current_node": { "node_id": "root", "prompt_audio_key": "ivr/demo-root", "options": [{"digit": "1", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: abandoned.append(session_id)) channel = DummyChannel() request = bridge_ivr_fastagi.AgiRequest( env={}, args=["call_invalid", "linked_invalid", "voice_lab_ivr", "voice_lab", "7200"], ) bridge_ivr_fastagi.run_ivr_call(channel, request) assert dtmf_calls == ["9", "9", "9"] assert abandoned == ["ivs_invalid"] assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "voice_lab" assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7000" assert channel.variables["MVPCC_IVR_RESULT"] == "fallback_invalid" def test_ivr_fastagi_runtime_abandons_on_hangup(monkeypatch): class DummyBridge: def _queue_map(self): return {"voice_lab_ivr": "q_ivr"} def _ivr_dtmf_timeout_seconds(self): return 5 def _ivr_max_no_input_retries(self): return 2 def _ivr_max_invalid_retries(self): return 2 class DummyChannel: def __init__(self): self.variables: dict[str, str | None] = {} def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None: return None def wait_for_digit(self, timeout_ms: int) -> str | None: raise bridge_ivr_fastagi.AgiHangup("caller hung up") def set_variable(self, name: str, value: str | None) -> None: self.variables[name] = value def verbose(self, message: str, *, level: int = 1) -> None: return None abandoned: list[str] = [] monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge()) monkeypatch.setattr( bridge_ivr_fastagi, "_wait_for_call_link_context", lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"}, ) monkeypatch.setattr( bridge_ivr_fastagi, "_start_session", lambda call_context, interaction_id, queue_id: { "session": {"session_id": "ivs_hangup"}, "current_node": {"node_id": "root", "options": []}, "completed": False, }, ) monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: abandoned.append(session_id)) channel = DummyChannel() request = bridge_ivr_fastagi.AgiRequest( env={}, args=["call_hangup", "linked_hangup", "voice_lab_ivr", "voice_lab", "7200"], ) bridge_ivr_fastagi.run_ivr_call(channel, request) assert abandoned == ["ivs_hangup"] def test_ivr_fastagi_runtime_uses_digit_captured_during_prompt(monkeypatch): class DummyBridge: def _queue_map(self): return {"voice_lab_ivr": "q_ivr", "support": "q_support", "voice_lab": "q_voice"} def _ivr_dtmf_timeout_seconds(self): return 5 def _ivr_max_no_input_retries(self): return 2 def _ivr_max_invalid_retries(self): return 2 class DummyChannel: def __init__(self): self.variables: dict[str, str | None] = {} self.waits: list[int] = [] self.playbacks: list[str] = [] def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None: self.playbacks.append(prompt_audio_key) return "2" def wait_for_digit(self, timeout_ms: int) -> str | None: self.waits.append(timeout_ms) return None def set_variable(self, name: str, value: str | None) -> None: self.variables[name] = value def verbose(self, message: str, *, level: int = 1) -> None: return None dtmf_calls: list[str] = [] monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge()) monkeypatch.setattr( bridge_ivr_fastagi, "_wait_for_call_link_context", lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"}, ) monkeypatch.setattr( bridge_ivr_fastagi, "_resolve_transfer_extension", lambda queue_code: {"support": "7010", "voice_lab": "7000"}.get(queue_code), ) monkeypatch.setattr( bridge_ivr_fastagi, "_start_session", lambda call_context, interaction_id, queue_id: { "session": {"session_id": "ivs_prompt_digit"}, "current_node": { "node_id": "root", "prompt_audio_key": "ivr/demo-root", "options": [{"digit": "2", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr( bridge_ivr_fastagi, "_send_dtmf", lambda session_id, digit: dtmf_calls.append(digit) or { "session": { "session_id": session_id, "status": "completed", "resolved_queue_id": "q_support", "resolved_queue_code": "support", }, "current_node": {"node_id": "support", "is_terminal": True}, "completed": True, }, ) monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: None) channel = DummyChannel() request = bridge_ivr_fastagi.AgiRequest( env={}, args=["call_prompt_digit", "linked_prompt_digit", "voice_lab_ivr", "voice_lab", "7200"], ) bridge_ivr_fastagi.run_ivr_call(channel, request) assert channel.playbacks == ["ivr/demo-root"] assert channel.waits == [] assert dtmf_calls == ["2"] assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "support" assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7010" def test_ivr_fastagi_runtime_plays_prompt_sequence_before_wait(monkeypatch): class DummyBridge: def _queue_map(self): return {"voice_lab_ivr": "q_ivr", "support": "q_support", "voice_lab": "q_voice"} def _ivr_dtmf_timeout_seconds(self): return 5 def _ivr_max_no_input_retries(self): return 2 def _ivr_max_invalid_retries(self): return 2 class DummyChannel: def __init__(self): self.variables: dict[str, str | None] = {} self.playbacks: list[tuple[str, int]] = [] self.waits: list[int] = [] def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None: self.playbacks.append((prompt_audio_key, timeout_ms)) if prompt_audio_key == "ivr/demo-language-kz": return "2" return None def wait_for_digit(self, timeout_ms: int) -> str | None: self.waits.append(timeout_ms) return None def set_variable(self, name: str, value: str | None) -> None: self.variables[name] = value def verbose(self, message: str, *, level: int = 1) -> None: return None dtmf_calls: list[str] = [] monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge()) monkeypatch.setattr( bridge_ivr_fastagi, "_wait_for_call_link_context", lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"}, ) monkeypatch.setattr( bridge_ivr_fastagi, "_resolve_transfer_extension", lambda queue_code: {"support": "7010", "voice_lab": "7000"}.get(queue_code), ) monkeypatch.setattr( bridge_ivr_fastagi, "_start_session", lambda call_context, interaction_id, queue_id: { "session": {"session_id": "ivs_prompt_sequence"}, "current_node": { "node_id": "root", "prompt_text": "Combined bilingual prompt", "prompt_sequence": [ {"prompt_audio_key": "ivr/demo-language-ru", "prompt_text": "Russian segment", "language": "ru"}, {"prompt_audio_key": "ivr/demo-language-kz", "prompt_text": "Kazakh segment", "language": "kz"}, ], "options": [{"digit": "2", "target_node_id": "support"}], }, "completed": False, }, ) monkeypatch.setattr( bridge_ivr_fastagi, "_send_dtmf", lambda session_id, digit: dtmf_calls.append(digit) or { "session": { "session_id": session_id, "status": "completed", "resolved_queue_id": "q_support", "resolved_queue_code": "support", }, "current_node": {"node_id": "support", "is_terminal": True}, "completed": True, }, ) monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: None) channel = DummyChannel() request = bridge_ivr_fastagi.AgiRequest( env={}, args=["call_prompt_sequence", "linked_prompt_sequence", "voice_lab_ivr", "voice_lab", "7200"], ) bridge_ivr_fastagi.run_ivr_call(channel, request) assert channel.playbacks == [ ("ivr/demo-language-ru", 1), ("ivr/demo-language-kz", 1), ] assert channel.waits == [] assert dtmf_calls == ["2"] assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "support" assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7010" def test_bridge_process_call_started_creates_interaction_and_call_link(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}') interaction_id = f"int_ast_{tmp_path.name}" emitted: dict[str, object] = {} monkeypatch.setattr( bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": interaction_id}, ) monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.update(kwargs) or {"event_id": "vev_ast"}, ) call_id = f"call_ast_{tmp_path.name}" session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_demo", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_demo", "CallerNumber": "+77010000001", "CallerName": "Lab Caller", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) bridge_module._process_bridge_row(session, row) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one_or_none() assert row.forward_status == "forwarded" assert row.interaction_id == interaction_id assert link is not None assert link.queue_code == "lab" assert link.queue_id == "que_lab" assert link.interaction_id == interaction_id assert emitted["event_type"] == "call.started" assert emitted["interaction_id"] == interaction_id assert emitted["payload"]["source"] == "asterisk" finally: session.close() client = TestClient(bridge_module.app) response = client.get( f"/asterisk/live-calls/{call_id}/ai-summary", headers=_admin_headers(), ) assert response.status_code == 200 payload = response.json() def test_call_started_uses_api_fallback_after_create_failure(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}') interaction_id = f"int_ast_api_{tmp_path.name}" emitted: dict[str, object] = {} def _raise_create_error(**kwargs): raise RuntimeError("interaction create failed") monkeypatch.setattr(bridge_module, "_create_interaction", _raise_create_error) monkeypatch.setattr( bridge_module, "_find_interaction_by_call_id", lambda call_id: {"interaction_id": interaction_id, "subject": f"Inbound call [{call_id}]"}, ) monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.update(kwargs) or {"event_id": "vev_ast_api"}, ) call_id = f"call_ast_api_{tmp_path.name}" session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_demo_api", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_demo_api", "CallerNumber": "+77010000021", "CallerName": "API Fallback Caller", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) bridge_module._process_bridge_row(session, row) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one_or_none() assert row.forward_status == "forwarded" assert row.interaction_id == interaction_id assert link is not None assert link.interaction_id == interaction_id assert emitted["event_type"] == "call.started" assert emitted["interaction_id"] == interaction_id finally: session.close() def test_voice_ai_summary_exposes_customer_name_state_for_operator(tmp_path): now = utc_now_iso() call_id = f"call_voice_summary_{tmp_path.name}" session_id = f"avs_voice_summary_{tmp_path.name}" ai_session_id = f"ais_voice_summary_{tmp_path.name}" interaction_id = f"int_voice_summary_{tmp_path.name}" session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id=f"linked_voice_summary_{tmp_path.name}", queue_code="voice_support", queue_id="que_voice_support", interaction_id=interaction_id, caller_number="+77010001122", caller_name="Summary Caller", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-000099", started_at=now, connected_at=now, ended_at=None, updated_at=now, voice_session_id=session_id, ai_session_id=ai_session_id, ai_state="handoff_required", ai_handoff_reason="customer requested operator", ai_last_model_at=now, voice_start_language="ru", customer_name_status="name_followup_required", customer_name_value="Айдос", customer_name_source="voice_start", customer_name_resolved_at=now, ) ) session.add( AISessionRow( session_id=ai_session_id, channel="voice", call_id=call_id, thread_id=None, interaction_id=interaction_id, customer_id="cus_voice_summary", agent_profile="voice_support", language="ru", status="handoff_required", summary_text="AI collected context and asked for operator handoff.", last_user_message_id=None, last_ai_message_id=None, handoff_reason="customer requested operator", created_at=now, updated_at=now, closed_at=None, ) ) session.add( VoiceAISessionRow( session_id=session_id, call_id=call_id, linked_id=f"linked_voice_summary_{tmp_path.name}", interaction_id=interaction_id, customer_id="cus_voice_summary", queue_id="que_voice_support", ai_session_id=ai_session_id, agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="yandex", status="handoff_requested", handoff_reason="customer requested operator", handoff_target_queue_id="que_voice_support", disclosure_played_at=now, last_user_utterance_at=now, last_ai_reply_at=now, started_at=now, updated_at=now, ended_at=None, voice_start_language="ru", customer_name_status="name_followup_required", customer_name_value="Айдос", customer_name_source="voice_start", customer_name_resolved_at=now, ) ) session.add_all( [ VoiceTranscriptSegmentRow( segment_id=f"{session_id}_seg_1", session_id=session_id, call_id=call_id, interaction_id=interaction_id, sequence_no=1, speaker="caller", source_type="voice_asr", text="Соедините с оператором", confidence=0.96, is_final=True, barge_in_interrupted=False, payload_json="{}", created_at=now, ), VoiceTranscriptSegmentRow( segment_id=f"{session_id}_seg_2", session_id=session_id, call_id=call_id, interaction_id=interaction_id, sequence_no=2, speaker="assistant", source_type="voice_policy", text="Сейчас переведу вас на оператора.", confidence=None, is_final=True, barge_in_interrupted=False, payload_json="{}", created_at=now, ), ] ) session.commit() finally: session.close() summary = bridge_module._voice_ai_summary_for_call(call_id) assert summary is not None assert summary.customer_name_status == "name_followup_required" assert summary.customer_name_value == "Айдос" assert summary.customer_name_source == "voice_start" assert summary.voice_start_language == "ru" def test_call_started_re_raises_original_create_error_when_all_fallbacks_fail(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}') def _raise_create_error(**kwargs): raise RuntimeError("interaction create failed") monkeypatch.setattr(bridge_module, "_create_interaction", _raise_create_error) monkeypatch.setattr(bridge_module, "_find_interaction_by_call_id", lambda call_id: None) call_id = f"call_ast_fail_{tmp_path.name}" session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_demo_fail", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_demo_fail", "CallerNumber": "+77010000022", "CallerName": "Failed Fallback Caller", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) with pytest.raises(RuntimeError, match="interaction create failed"): bridge_module._process_bridge_row(session, row) finally: session.close() def test_reconcile_keeps_live_call_active_when_ami_still_sees_channels(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1") call_id = f"call_reconcile_live_{tmp_path.name}" stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_reconcile_live", queue_code="lab", queue_id="que_lab", interaction_id=f"int_reconcile_live_{tmp_path.name}", caller_number="+77010000004", caller_name="Track11 Reconcile Live", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-stale", started_at=stale_started, connected_at=stale_started, ended_at=None, updated_at=stale_started, ) session.add(link) session.commit() finally: session.close() emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_live"}, ) monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: ["PJSIP/2001-00009999"], ) bridge_module._reconcile_stale_calls_once() session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.status == "active" assert link.telephony_status == "connected" assert link.ended_at is None assert link.channel_name == "PJSIP/2001-00009999" assert [item for item in emitted if item["call_id"] == call_id] == [] finally: session.close() def test_reconcile_skips_synthetic_end_when_channel_lookup_is_unavailable(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1") call_id = f"call_reconcile_unavailable_{tmp_path.name}" stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_reconcile_unavailable", queue_code="lab", queue_id="que_lab", interaction_id=f"int_reconcile_unavailable_{tmp_path.name}", caller_number="+77010000024", caller_name="Track11 Reconcile Unavailable", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-00009998", started_at=stale_started, connected_at=stale_started, ended_at=None, updated_at=stale_started, ) session.add(link) session.commit() finally: session.close() emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_unavailable"}, ) monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: None, ) bridge_module._reconcile_stale_calls_once() session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.status == "active" assert link.telephony_status == "connected" assert link.ended_at is None assert [item for item in emitted if item["call_id"] == call_id] == [] finally: session.close() def test_reconcile_emits_synthetic_end_and_closes_call(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1") call_id = f"call_reconcile_soft_{tmp_path.name}" stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_reconcile_soft", queue_code="lab", queue_id="que_lab", interaction_id=f"int_reconcile_soft_{tmp_path.name}", caller_number="+77010000005", caller_name="Track11 Reconcile Soft", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-00000010", started_at=stale_started, connected_at=stale_started, ended_at=None, updated_at=stale_started, ) session.add(link) session.commit() finally: session.close() emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_soft"}, ) monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: [], ) bridge_module._reconcile_stale_calls_once() session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.status == "ended" assert link.telephony_status == "ended" assert link.ended_at is not None matching = [item for item in emitted if item["call_id"] == call_id] assert len(matching) == 1 assert matching[0]["event_type"] == "call.ended" assert matching[0]["payload"]["reconciled"] is True finally: session.close() def test_reconcile_keeps_recent_active_ai_call_open(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1") call_id = f"call_reconcile_ai_live_{tmp_path.name}" session_id = f"avs_reconcile_ai_live_{tmp_path.name}" stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() recent_media = (datetime.now(timezone.utc) - timedelta(seconds=2)).isoformat() session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_reconcile_ai_live", queue_code="voice_lab_ai", queue_id="que_voice_lab", interaction_id=f"int_reconcile_ai_live_{tmp_path.name}", caller_number="+77010000055", caller_name="Track16 Reconcile AI", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-0000AI55", voice_session_id=session_id, ai_session_id=f"ais_reconcile_ai_live_{tmp_path.name}", ai_state="speaking", ai_handoff_reason=None, ai_last_model_at=recent_media, started_at=stale_started, connected_at=stale_started, ended_at=None, updated_at=stale_started, ) session.add(link) session.add( VoiceAISessionRow( session_id=session_id, call_id=call_id, linked_id="linked_reconcile_ai_live", interaction_id=link.interaction_id, customer_id=None, queue_id="que_voice_lab", ai_session_id=f"ais_reconcile_ai_live_{tmp_path.name}", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="yandex", status="active", handoff_reason=None, handoff_target_queue_id="que_voice_lab", media_uuid="11111111-1111-1111-1111-111111111111", media_status="connected", media_connected_at=stale_started, media_ended_at=None, last_media_frame_at=recent_media, disclosure_played_at=stale_started, last_user_utterance_at=recent_media, last_ai_reply_at=recent_media, started_at=stale_started, updated_at=recent_media, ended_at=None, ) ) session.commit() finally: session.close() emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_ai_live"}, ) monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: [], ) bridge_module._reconcile_stale_calls_once() session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.status == "active" assert link.telephony_status == "connected" assert link.ended_at is None assert [item for item in emitted if item["call_id"] == call_id] == [] finally: session.close() def test_reconcile_closes_active_call_without_duplicate_end_event(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1") call_id = f"call_reconcile_existing_end_{tmp_path.name}" stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_reconcile_existing_end", queue_code="lab", queue_id="que_lab", interaction_id=f"int_reconcile_existing_end_{tmp_path.name}", caller_number="+77010000023", caller_name="Track11 Reconcile Existing End", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-00000011", started_at=stale_started, connected_at=stale_started, ended_at=None, updated_at=stale_started, ) session.add(link) session.add( VoiceEventRow( event_id=f"vev_reconcile_existing_end_{tmp_path.name}", event_type="call.ended", call_id=call_id, interaction_id=link.interaction_id, payload_json=json.dumps({"hangup_cause": "normal_clearing"}), created_at=stale_started, ) ) session.commit() finally: session.close() emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_existing_end"}, ) monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: [], ) bridge_module._reconcile_stale_calls_once() session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.status == "ended" assert link.telephony_status == "ended" assert link.ended_at is not None assert [item for item in emitted if item["call_id"] == call_id] == [] finally: session.close() def test_reconcile_prioritizes_oldest_unresolved_ended_call(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1") monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_SCAN_LIMIT", "1") older_iso = (datetime.now(timezone.utc) - timedelta(seconds=120)).isoformat() newer_iso = (datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat() older_call_id = f"call_reconcile_oldest_{tmp_path.name}" newer_call_id = f"call_reconcile_newer_with_recording_{tmp_path.name}" recording_path = tmp_path / "fixtures" / "reconcile-oldest.wav" recording_path.parent.mkdir(parents=True, exist_ok=True) recording_path.write_bytes(b"RIFF" + b"\x00" * 32) session = get_session() try: session.query(CallRecordingRow).delete() session.query(AsteriskCallLinkRow).delete() session.commit() older_link = AsteriskCallLinkRow( call_id=older_call_id, linked_id="linked_reconcile_oldest", queue_code="lab", queue_id="que_lab", interaction_id=f"int_reconcile_oldest_{tmp_path.name}", caller_number="+77010000025", caller_name="Track11 Reconcile Oldest", status="ended", telephony_status="ended", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-oldest", started_at=older_iso, connected_at=older_iso, ended_at=older_iso, updated_at=older_iso, ) newer_link = AsteriskCallLinkRow( call_id=newer_call_id, linked_id="linked_reconcile_newer", queue_code="lab", queue_id="que_lab", interaction_id=f"int_reconcile_newer_{tmp_path.name}", caller_number="+77010000026", caller_name="Track11 Reconcile Newer", status="ended", telephony_status="ended", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-newer", started_at=newer_iso, connected_at=newer_iso, ended_at=newer_iso, updated_at=newer_iso, ) session.add_all([older_link, newer_link]) session.flush() session.add( CallRecordingRow( recording_id=f"rec_reconcile_newer_{tmp_path.name}", channel="voice", call_id=newer_call_id, interaction_id=newer_link.interaction_id, source_event_id=f"src_reconcile_newer_{tmp_path.name}", file_name="newer.wav", storage_backend="local_fs", storage_path=str(recording_path), mime_type="audio/wav", size_bytes=recording_path.stat().st_size, duration_seconds=4, checksum_sha256="abc123", status="ready", recorded_at=newer_iso, created_at=newer_iso, updated_at=newer_iso, archived_at=None, ) ) session.commit() finally: session.close() uploaded: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_find_remote_recording_for_call", lambda call_id: ( ("/remote/oldest.wav", "oldest.wav", "audio/wav", datetime.now(timezone.utc) - timedelta(seconds=60)) if call_id == older_call_id else None ), ) monkeypatch.setattr( bridge_module, "_fetch_recording_file", lambda remote_path, file_name: (recording_path, False), ) monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": f"vev_{kwargs['call_id']}"}, ) monkeypatch.setattr( bridge_module, "_upload_recording", lambda **kwargs: uploaded.append(kwargs) or {"recording_id": f"rec_uploaded_{kwargs['call_id']}"}, ) bridge_module._reconcile_stale_calls_once() assert [entry["call_id"] for entry in uploaded] == [older_call_id] def test_recent_calls_route_returns_recent_transferred_call(tmp_path): client = TestClient(bridge_module.app) now = utc_now_iso() call_id = f"call_recent_transfer_{tmp_path.name}" session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_recent_transfer", queue_code="voice_lab", queue_id="que_recent_transfer", interaction_id=f"int_recent_transfer_{tmp_path.name}", caller_number="+77010000006", caller_name="Recent Transfer", status="ended", telephony_status="ended", claimed_by_user="operator", claimed_at=now, operator_extension="2001", channel_name="PJSIP/2001-00010000", started_at=now, connected_at=now, ended_at=now, updated_at=now, ) session.add(link) session.add( AsteriskCallActionLogRow( action_id=f"aca_recent_transfer_{tmp_path.name}", call_id=call_id, interaction_id=link.interaction_id, action_type="blind-transfer", actor_user="operator", actor_role="operator", request_json=json.dumps({"target_value": "2002", "resolved_extension": "2002"}), result_status="ok", ami_action_id="ami_recent_transfer", error=None, created_at=now, ) ) session.add( VoiceEventRow( event_id=f"vev_recent_transfer_{tmp_path.name}", event_type="call.ended", call_id=call_id, interaction_id=link.interaction_id, payload_json=json.dumps({"hangup_cause": "normal_clearing"}), created_at=now, ) ) session.commit() finally: session.close() response = client.get("/asterisk/recent-calls", headers=_admin_headers()) assert response.status_code == 200 payload = response.json() item = next(entry for entry in payload if entry["call_id"] == call_id) assert item["terminal_action"] == "blind-transfer" assert item["terminal_target"] == "2002" assert item["hangup_cause"] == "normal_clearing" assert item["last_transition_at"] == now def test_recent_calls_route_skips_stale_calls(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_RECENT_CALLS_WINDOW_SECONDS", "120") client = TestClient(bridge_module.app) old_iso = (datetime.now(timezone.utc) - timedelta(seconds=600)).isoformat() call_id = f"call_recent_stale_{tmp_path.name}" session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id="linked_recent_stale", queue_code="voice_lab", queue_id="que_recent_stale", interaction_id=f"int_recent_stale_{tmp_path.name}", caller_number="+77010000007", caller_name="Recent Stale", status="ended", telephony_status="ended", claimed_by_user="operator", claimed_at=old_iso, operator_extension="2001", channel_name="PJSIP/2001-00010001", started_at=old_iso, connected_at=old_iso, ended_at=old_iso, updated_at=old_iso, ) ) session.commit() finally: session.close() response = client.get("/asterisk/recent-calls", headers=_admin_headers()) assert response.status_code == 200 payload = response.json() assert all(entry["call_id"] != call_id for entry in payload) def test_retry_endpoint_replays_failed_bridge_event(monkeypatch, tmp_path): call_id = f"call_retry_{tmp_path.name}" monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", "{}") session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_retry", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_retry", "CallerNumber": "+77010000002", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) try: bridge_module._process_bridge_row(session, row) except Exception as exc: # noqa: BLE001 bridge_module._mark_log_failed(session, row, str(exc)) session.commit() bridge_event_id = row.bridge_event_id assert row.forward_status == "failed" finally: session.close() monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_retry"}') monkeypatch.setattr( bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_retry_{tmp_path.name}"}, ) emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_retry"}, ) client = TestClient(bridge_module.app) retried = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers()) repeated = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers()) assert retried.status_code == 200 assert repeated.status_code == 200 payload = retried.json() assert payload["forward_status"] == "forwarded" assert payload["interaction_id"].startswith("int_retry_") assert repeated.json()["forward_status"] == "forwarded" assert len(emitted) == 1 def test_retry_endpoint_does_not_double_process_row_already_claimed_by_background_worker(monkeypatch, tmp_path): call_id = f"call_retry_claimed_{tmp_path.name}" monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_retry_claimed"}') session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_retry_claimed", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_retry_claimed", "CallerNumber": "+77010000027", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) bridge_module._mark_log_failed(session, row, "retry me") session.commit() bridge_event_id = row.bridge_event_id finally: session.close() emitted: list[dict[str, object]] = [] monkeypatch.setattr( bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_retry_claimed_{tmp_path.name}"}, ) monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_retry_claimed"}, ) assert bridge_module._claim_bridge_event_for_processing(bridge_event_id) is True client = TestClient(bridge_module.app) blocked = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers()) assert blocked.status_code == 200 assert blocked.json()["forward_status"] == "processing" assert emitted == [] processed = bridge_module._process_claimed_bridge_event(bridge_event_id) assert processed is not None assert processed.forward_status == "forwarded" assert len(emitted) == 1 def test_retry_recovers_after_voice_event_was_sent_before_db_failure(monkeypatch, tmp_path): call_id = f"call_retry_voice_commit_{tmp_path.name}" monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_retry_voice_commit"}') emitted_by_source: dict[str, dict[str, str]] = {} emitted_calls: list[str] = [] def _emit_once_per_source(**kwargs): source_event_id = str(kwargs["source_event_id"]) emitted_calls.append(source_event_id) return emitted_by_source.setdefault( source_event_id, {"event_id": f"vev_retry_voice_{len(emitted_by_source) + 1}"}, ) original_mark_log_forwarded = bridge_module._mark_log_forwarded should_fail_once = {"value": True} def _mark_log_forwarded_once(session, row, **kwargs): original_mark_log_forwarded(session, row, **kwargs) if should_fail_once["value"]: should_fail_once["value"] = False raise RuntimeError("db write failed after voice event") monkeypatch.setattr( bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_retry_voice_commit_{tmp_path.name}"}, ) monkeypatch.setattr(bridge_module, "_emit_voice_event", _emit_once_per_source) monkeypatch.setattr(bridge_module, "_mark_log_forwarded", _mark_log_forwarded_once) bridge_module._record_ami_payload( { "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_retry_voice_commit", "CallerNumber": "+77010000028", "CallerName": "Retry Voice Commit", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", } ) session = get_session() try: row = session.execute( select(AsteriskEventLogRow).where(AsteriskEventLogRow.call_id == call_id) ).scalar_one() bridge_event_id = row.bridge_event_id assert row.forward_status == "failed" finally: session.close() client = TestClient(bridge_module.app) retried = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers()) assert retried.status_code == 200 assert retried.json()["forward_status"] == "forwarded" assert len(emitted_calls) == 2 assert len(emitted_by_source) == 1 def test_retry_recovers_after_recording_upload_before_db_failure(monkeypatch, tmp_path): call_id = f"call_retry_recording_commit_{tmp_path.name}" ready_iso = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat() recording_fixture = tmp_path / "fixtures" / "retry-recording.wav" recording_fixture.parent.mkdir(parents=True, exist_ok=True) recording_fixture.write_bytes(b"RIFF" + b"\x00" * 64) session = get_session() try: link = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_retry_recording_commit", queue_code="lab", queue_id="que_lab", interaction_id=f"int_retry_recording_commit_{tmp_path.name}", caller_number="+77010000029", caller_name="Retry Recording Commit", status="ended", telephony_status="ended", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-recording", started_at=ready_iso, connected_at=ready_iso, ended_at=ready_iso, updated_at=ready_iso, ) session.add(link) session.add( VoiceEventRow( event_id=f"vev_retry_recording_ended_{tmp_path.name}", event_type="call.ended", call_id=call_id, interaction_id=link.interaction_id, source_event_id=f"src_retry_recording_ended_{tmp_path.name}", payload_json=json.dumps({"hangup_cause": "normal_clearing"}), created_at=ready_iso, ) ) session.add( VoiceEventRow( event_id=f"vev_retry_recording_ready_{tmp_path.name}", event_type="recording.ready", call_id=call_id, interaction_id=link.interaction_id, source_event_id=f"src_retry_recording_ready_{tmp_path.name}", payload_json=json.dumps({"remote_path": "/remote/retry-recording.wav"}), created_at=ready_iso, ) ) session.commit() finally: session.close() uploaded_by_source: dict[str, dict[str, str]] = {} upload_calls: list[str] = [] def _upload_once_per_source(**kwargs): source_event_id = str(kwargs["source_event_id"]) upload_calls.append(source_event_id) return uploaded_by_source.setdefault( source_event_id, {"recording_id": f"rec_retry_recording_{len(uploaded_by_source) + 1}"}, ) original_mark_log_forwarded = bridge_module._mark_log_forwarded should_fail_once = {"value": True} def _mark_log_forwarded_once(session, row, **kwargs): original_mark_log_forwarded(session, row, **kwargs) if should_fail_once["value"]: should_fail_once["value"] = False raise RuntimeError("db write failed after recording upload") monkeypatch.setattr( bridge_module, "_fetch_recording_file", lambda remote_path, file_name: (recording_fixture, False), ) monkeypatch.setattr(bridge_module, "_upload_recording", _upload_once_per_source) monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": f"vev_{kwargs['event_type']}"}, ) monkeypatch.setattr(bridge_module, "_mark_log_forwarded", _mark_log_forwarded_once) bridge_module._record_ami_payload( { "UserEvent": "MVPCCRecordingReady", "CallID": call_id, "LinkedID": "linked_retry_recording_commit", "RemotePath": "/remote/retry-recording.wav", "FileName": "retry-recording.wav", "MimeType": "audio/wav", "DurationSeconds": "7", } ) session = get_session() try: row = session.execute( select(AsteriskEventLogRow) .where(AsteriskEventLogRow.call_id == call_id) .where(AsteriskEventLogRow.ami_event_name == "MVPCCRecordingReady") ).scalar_one() bridge_event_id = row.bridge_event_id assert row.forward_status == "failed" finally: session.close() client = TestClient(bridge_module.app) retried = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers()) assert retried.status_code == 200 assert retried.json()["forward_status"] == "forwarded" assert len(upload_calls) == 2 assert len(uploaded_by_source) == 1 def test_startup_fails_fast_when_singleton_guard_is_already_held(monkeypatch): monkeypatch.setenv("ASTERISK_BRIDGE_ENABLED", "1") monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped) monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped) monkeypatch.setattr(bridge_module, "_try_acquire_bridge_singleton_guard", lambda: False) with pytest.raises(RuntimeError, match="replicas=1, workers=1"): bridge_module._startup() def test_disabled_bridge_startup_does_not_require_singleton_guard(monkeypatch): monkeypatch.delenv("ASTERISK_BRIDGE_ENABLED", raising=False) monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped) monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped) monkeypatch.setattr( bridge_module, "_try_acquire_bridge_singleton_guard", lambda: (_ for _ in ()).throw(AssertionError("singleton guard should not be acquired")), ) bridge_module._startup() assert len(bridge_module._background_threads_alive()) == 2 def test_shutdown_releases_singleton_guard(monkeypatch): monkeypatch.setenv("ASTERISK_BRIDGE_ENABLED", "1") monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped) monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped) bridge_module._startup() try: assert bridge_module._BRIDGE_SINGLETON_FILE_HANDLE is not None finally: bridge_module._shutdown() assert bridge_module._BRIDGE_SINGLETON_FILE_HANDLE is None assert bridge_module._try_acquire_bridge_singleton_guard() is True bridge_module._release_bridge_singleton_guard() def test_claim_endpoint_updates_call_link_and_action_log(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}') monkeypatch.setattr(bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_claim_{tmp_path.name}"}) monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_claim"}) monkeypatch.setattr( bridge_module, "_resolve_channel_name", lambda session, link, prefer_operator=False, refresh=False, snapshot=None: "PJSIP/1001-0000001", ) monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_claim"}) monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"}) call_id = f"call_claim_{tmp_path.name}" session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_claim", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_claim", "CallerNumber": "+77010000003", "CallerName": "Track11 Claim", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", "Channel": "PJSIP/1001-0000001", }, ) bridge_module._process_bridge_row(session, row) session.commit() finally: session.close() client = TestClient(bridge_module.app) claimed = client.post( f"/asterisk/live-calls/{call_id}/claim", headers=_operator_headers("operator_a"), json={}, ) assert claimed.status_code == 200 payload = claimed.json() assert payload["claimed_by_user"] == "operator_a" assert payload["operator_extension"] == "2001" assert payload["telephony_status"] == "claimed" session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.claimed_by_user == "operator_a" assert link.operator_extension == "2001" action = session.execute( select(AsteriskCallActionLogRow) .where(AsteriskCallActionLogRow.call_id == call_id) .where(AsteriskCallActionLogRow.action_type == "claim") .order_by(AsteriskCallActionLogRow.id.desc()) ).scalar_one() assert action.result_status == "ok" assert action.ami_action_id == "ami_claim" finally: session.close() def test_claim_resolves_live_channels_once_on_happy_path(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_claim_fast_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_claim_fast", queue_code="lab", queue_id="que_lab", interaction_id=f"int_claim_fast_{tmp_path.name}", caller_number="+77010000013", caller_name="Track12 Claim Fast", status="active", telephony_status="ringing", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-0000013", started_at=utc_now_iso(), connected_at=None, ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() channel_calls = [] def _fake_channels(**kwargs): channel_calls.append(kwargs) return ["PJSIP/1001-0000013"] monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", _fake_channels) monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"}) monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_claim_fast"}) client = TestClient(bridge_module.app) claimed = client.post( f"/asterisk/live-calls/{call_id}/claim", headers=_operator_headers("operator_a"), json={}, ) assert claimed.status_code == 200 assert len(channel_calls) == 1 def test_claim_skips_redirect_when_operator_leg_already_exists(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_claim_existing_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_claim_existing", queue_code="lab", queue_id="que_lab", interaction_id=f"int_claim_existing_{tmp_path.name}", caller_number="+77010000007", caller_name="Track11 Existing Leg", status="active", telephony_status="ringing", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-0000004", started_at=utc_now_iso(), connected_at=None, ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: ["PJSIP/1001-0000004", "PJSIP/2001-0000005"], ) monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"}) def _unexpected_ami(*args, **kwargs): raise AssertionError("Redirect should be skipped when operator leg already exists") monkeypatch.setattr(bridge_module, "_ami_action", _unexpected_ami) client = TestClient(bridge_module.app) claimed = client.post( f"/asterisk/live-calls/{call_id}/claim", headers=_operator_headers("operator_a"), json={}, ) assert claimed.status_code == 200 payload = claimed.json() assert payload["claimed_by_user"] == "operator_a" assert payload["operator_extension"] == "2001" assert payload["telephony_status"] == "claimed" def test_claim_connected_call_becomes_idempotent_success(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_claim_connected_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_claim_connected", queue_code="lab", queue_id="que_lab", interaction_id=f"int_claim_connected_{tmp_path.name}", caller_number="+77010000010", caller_name="Track11 Connected Claim", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-0000009", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: []) monkeypatch.setattr( bridge_module, "_latest_event_channel", lambda session, *, call_id, event_name: "PJSIP/2001-0000009" if event_name.endswith("OperatorConnected") else "PJSIP/1001-0000008", ) monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"}) def _unexpected_ami(*args, **kwargs): raise AssertionError("Redirect should be skipped when call is already connected") monkeypatch.setattr(bridge_module, "_ami_action", _unexpected_ami) client = TestClient(bridge_module.app) claimed = client.post( f"/asterisk/live-calls/{call_id}/claim", headers=_operator_headers("operator_a"), json={}, ) assert claimed.status_code == 200 payload = claimed.json() assert payload["claimed_by_user"] == "operator_a" assert payload["operator_extension"] == "2001" assert payload["telephony_status"] == "connected" def test_operator_cannot_hangup_foreign_claim(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") call_id = f"call_foreign_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_foreign", queue_code="lab", queue_id="que_lab", interaction_id=f"int_foreign_{tmp_path.name}", caller_number="+77010000004", caller_name="Track11 Foreign", status="active", telephony_status="claimed", claimed_by_user="operator_b", claimed_at=utc_now_iso(), operator_extension="2002", channel_name="PJSIP/1001-0000002", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() client = TestClient(bridge_module.app) denied = client.post( f"/asterisk/live-calls/{call_id}/hangup", headers=_operator_headers("operator_a"), json={}, ) assert denied.status_code == 403 def test_operator_can_hangup_own_extension_without_claimed_user(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_own_ext_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_own_ext", queue_code="lab", queue_id="que_lab", interaction_id=f"int_own_ext_{tmp_path.name}", caller_number="+77010000008", caller_name="Track11 Own Ext", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension="2001", channel_name="PJSIP/2001-0000006", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr( bridge_module, "_resolve_channel_name", lambda session, link, prefer_operator=False, refresh=False, snapshot=None: "PJSIP/2001-0000006", ) monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_hangup"}) client = TestClient(bridge_module.app) response = client.post( f"/asterisk/live-calls/{call_id}/hangup", headers=_operator_headers("operator_a"), json={}, ) assert response.status_code == 200 assert response.json()["telephony_status"] == "ended" def test_hangup_falls_back_to_another_live_leg(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_hangup_fallback_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_hangup_fallback", queue_code="lab", queue_id="que_lab", interaction_id=f"int_hangup_fallback_{tmp_path.name}", caller_number="+77010000011", caller_name="Track11 Hangup Fallback", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at=utc_now_iso(), operator_extension="2001", channel_name="PJSIP/2001-0000010", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr( bridge_module, "_candidate_channels_for_call", lambda session, link, operator_extension=None, prefer_operator=False, snapshot=None: [ "PJSIP/2001-0000010", "PJSIP/1001-0000011", ], ) calls = [] def _ami_action(action, params): calls.append(params["Channel"]) if params["Channel"] == "PJSIP/2001-0000010": raise RuntimeError("No such channel") return {"action_id": "ami_hangup_fallback"} monkeypatch.setattr(bridge_module, "_ami_action", _ami_action) client = TestClient(bridge_module.app) response = client.post( f"/asterisk/live-calls/{call_id}/hangup", headers=_operator_headers("operator_a"), json={}, ) assert response.status_code == 200 assert response.json()["telephony_status"] == "ended" assert calls == ["PJSIP/2001-0000010", "PJSIP/1001-0000011"] def test_hangup_becomes_idempotent_success_when_channel_already_closed(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_hangup_closed_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_hangup_closed", queue_code="lab", queue_id="que_lab", interaction_id=f"int_hangup_closed_{tmp_path.name}", caller_number="+77010000012", caller_name="Track11 Hangup Closed", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at=utc_now_iso(), operator_extension="2001", channel_name="PJSIP/2001-0000012", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr( bridge_module, "_candidate_channels_for_call", lambda session, link, operator_extension=None, prefer_operator=False, snapshot=None: ["PJSIP/2001-0000012"], ) monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: []) def _ami_action(action, params): raise RuntimeError("No such channel") monkeypatch.setattr(bridge_module, "_ami_action", _ami_action) client = TestClient(bridge_module.app) response = client.post( f"/asterisk/live-calls/{call_id}/hangup", headers=_operator_headers("operator_a"), json={}, ) assert response.status_code == 200 assert response.json()["telephony_status"] == "ended" session = get_session() try: action = session.execute( select(AsteriskCallActionLogRow) .where(AsteriskCallActionLogRow.call_id == call_id) .where(AsteriskCallActionLogRow.action_type == "hangup") .order_by(AsteriskCallActionLogRow.id.desc()) ).scalar_one() assert action.result_status == "ok" assert action.error == "channel_already_closed" finally: session.close() def test_hangup_resolves_live_channels_once_on_happy_path(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_hangup_fast_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_hangup_fast", queue_code="lab", queue_id="que_lab", interaction_id=f"int_hangup_fast_{tmp_path.name}", caller_number="+77010000014", caller_name="Track12 Hangup Fast", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at=utc_now_iso(), operator_extension="2001", channel_name="PJSIP/2001-0000014", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() channel_calls = [] def _fake_channels(**kwargs): channel_calls.append(kwargs) return ["PJSIP/2001-0000014"] monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", _fake_channels) monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_hangup_fast"}) client = TestClient(bridge_module.app) response = client.post( f"/asterisk/live-calls/{call_id}/hangup", headers=_operator_headers("operator_a"), json={}, ) assert response.status_code == 200 assert len(channel_calls) == 1 def test_hangup_hangs_up_all_live_legs_on_success(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') call_id = f"call_hangup_all_legs_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_hangup_all_legs", queue_code="lab", queue_id="que_lab", interaction_id=f"int_hangup_all_legs_{tmp_path.name}", caller_number="+77010000015", caller_name="Track14 Hangup All Legs", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at=utc_now_iso(), operator_extension="2001", channel_name="PJSIP/2001-0000015", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr( bridge_module, "_candidate_channels_for_call", lambda session, link, operator_extension=None, prefer_operator=False, snapshot=None: [ "PJSIP/2001-0000015", "PJSIP/1001-0000016", ], ) calls = [] def _ami_action(action, params): calls.append(params["Channel"]) return {"action_id": f"ami_{len(calls)}"} monkeypatch.setattr(bridge_module, "_ami_action", _ami_action) client = TestClient(bridge_module.app) response = client.post( f"/asterisk/live-calls/{call_id}/hangup", headers=_operator_headers("operator_a"), json={}, ) assert response.status_code == 200 assert response.json()["telephony_status"] == "ended" assert calls == ["PJSIP/2001-0000015", "PJSIP/1001-0000016"] def test_blind_transfer_emits_transferred_voice_event(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1") monkeypatch.setenv("ASTERISK_TRANSFER_TARGET_MAP_JSON", '{"voice_lab":"2001"}') events: list[dict] = [] call_id = f"call_transfer_{tmp_path.name}" session = get_session() try: row = AsteriskCallLinkRow( call_id=call_id, linked_id="linked_transfer", queue_code="voice_lab", queue_id="que_lab", interaction_id=f"int_transfer_{tmp_path.name}", caller_number="+77010000005", caller_name="Track11 Transfer", status="active", telephony_status="connected", claimed_by_user="operator_a", claimed_at=utc_now_iso(), operator_extension="2001", channel_name="PJSIP/1001-0000003", started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) session.add(row) session.commit() finally: session.close() monkeypatch.setattr( bridge_module, "_resolve_channel_name", lambda session, link, prefer_operator=False, refresh=False, snapshot=None: "PJSIP/1001-0000003", ) monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_transfer"}) monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: events.append(kwargs) or {"event_id": "vev_transfer"}, ) client = TestClient(bridge_module.app) moved = client.post( f"/asterisk/live-calls/{call_id}/blind-transfer", headers=_operator_headers("operator_a"), json={"target_type": "queue_code", "target_value": "voice_lab"}, ) assert moved.status_code == 200 assert moved.json()["call_id"] == call_id assert events assert events[-1]["event_type"] == "call.transferred" def test_operator_connected_event_sets_connected_status(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}') monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}') monkeypatch.setattr(bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_conn_{tmp_path.name}"}) captured: list[dict] = [] monkeypatch.setattr( bridge_module, "_emit_voice_event", lambda **kwargs: captured.append(kwargs) or {"event_id": "vev_conn"}, ) call_id = f"call_conn_{tmp_path.name}" session = get_session() try: started = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_conn", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_conn", "CallerNumber": "+77010000006", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) bridge_module._process_bridge_row(session, started) session.commit() connected = bridge_module._create_bridge_log( session, ami_event_name="MVPCCOperatorConnected", call_id=call_id, linked_id="linked_conn", payload={ "UserEvent": "MVPCCOperatorConnected", "CallID": call_id, "LinkedID": "linked_conn", "OperatorExtension": "2001", "ClaimedByUser": "s", "Channel": "PJSIP/1001-0000010", }, ) bridge_module._process_bridge_row(session, connected) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.telephony_status == "connected" assert link.operator_extension == "2001" assert link.claimed_by_user == "operator_a" finally: session.close() assert any(item.get("event_type") == "call.connected" for item in captured) def test_operator_connected_does_not_guess_claimed_user_for_ambiguous_extension(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}') monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator":"2001","admin":"2001"}') monkeypatch.setattr(bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_conn_amb_{tmp_path.name}"}) monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_conn_amb"}) call_id = f"call_conn_amb_{tmp_path.name}" session = get_session() try: started = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id="linked_conn_amb", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": "linked_conn_amb", "CallerNumber": "+77010000009", "QueueCode": "lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) bridge_module._process_bridge_row(session, started) session.commit() connected = bridge_module._create_bridge_log( session, ami_event_name="MVPCCOperatorConnected", call_id=call_id, linked_id="linked_conn_amb", payload={ "UserEvent": "MVPCCOperatorConnected", "CallID": call_id, "LinkedID": "linked_conn_amb", "OperatorExtension": "2001", "ClaimedByUser": "s", "Channel": "PJSIP/1001-0000011", }, ) bridge_module._process_bridge_row(session, connected) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.telephony_status == "connected" assert link.operator_extension == "2001" assert link.claimed_by_user is None finally: session.close() def test_app_module_exposes_refactor_compatibility_surface(): assert bridge_module.app is not None assert callable(bridge_module._startup) assert callable(bridge_module._shutdown) assert callable(bridge_module._release_bridge_singleton_guard) assert bridge_module._process_bridge_row.__module__ == "services.asterisk_bridge_service.bridge_processing" assert bridge_module._live_call_to_out.__module__ == "services.asterisk_bridge_service.presenters" assert bridge_module._find_remote_recording_for_call.__module__ == "services.asterisk_bridge_service.recording_io" def test_private_shims_remain_monkeypatchable(monkeypatch): monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "shim"}) result = bridge_module._emit_voice_event( event_type="call.started", call_id="call_shim", interaction_id=None, payload={}, ) assert result == {"event_id": "shim"} def test_call_started_with_ai_queue_creates_voice_ai_session(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"voice_lab":"que_voice_lab"}') monkeypatch.setenv("AI_VOICE_ENABLED", "1") monkeypatch.setenv( "AI_VOICE_QUEUE_CONFIG_JSON", '{"voice_lab":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}}', ) monkeypatch.setattr( bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_voice_ai_{tmp_path.name}"}, ) monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_voice_ai"}) monkeypatch.setattr( bridge_module, "_start_voice_ai_session", lambda **kwargs: { "voice_session_id": f"avs_voice_ai_{tmp_path.name}", "ai_session_id": f"ais_voice_ai_{tmp_path.name}", "status": "greeting", }, ) call_id = f"call_voice_ai_{tmp_path.name}" session = get_session() try: row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id=f"linked_voice_ai_{tmp_path.name}", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": f"linked_voice_ai_{tmp_path.name}", "CallerNumber": "+77010000041", "CallerName": "Voice AI Caller", "QueueCode": "voice_lab", "Extension": "7000", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) bridge_module._process_bridge_row(session, row) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.voice_session_id == f"avs_voice_ai_{tmp_path.name}" assert link.ai_session_id == f"ais_voice_ai_{tmp_path.name}" assert link.ai_state == "greeting" finally: session.close() def test_call_started_ai_does_not_flush_db_before_start_request(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"voice_lab":"que_voice_lab"}') monkeypatch.setenv("AI_VOICE_ENABLED", "1") monkeypatch.setenv( "AI_VOICE_QUEUE_CONFIG_JSON", '{"voice_lab":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}}', ) monkeypatch.setattr( bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_voice_ai_flush_{tmp_path.name}"}, ) monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_voice_ai_flush"}) flush_count = {"value": 0} session = get_session() original_flush = type(session).flush def _tracked_flush(*args, **kwargs): flush_count["value"] += 1 return original_flush(session, *args, **kwargs) call_id = f"call_voice_ai_flush_{tmp_path.name}" try: monkeypatch.setattr(session, "flush", _tracked_flush) row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCCallStarted", call_id=call_id, linked_id=f"linked_voice_ai_flush_{tmp_path.name}", payload={ "UserEvent": "MVPCCCallStarted", "CallID": call_id, "LinkedID": f"linked_voice_ai_flush_{tmp_path.name}", "CallerNumber": "+77010000042", "CallerName": "Voice AI Flush Caller", "QueueCode": "voice_lab", "Extension": "7100", "Context": "mvpcc-inbound", "Direction": "inbound", }, ) flush_count["value"] = 0 def _start_voice_ai_session(**kwargs): del kwargs assert flush_count["value"] == 0 return { "voice_session_id": f"avs_voice_ai_flush_{tmp_path.name}", "ai_session_id": f"ais_voice_ai_flush_{tmp_path.name}", "status": "greeting", } monkeypatch.setattr(bridge_module, "_start_voice_ai_session", _start_voice_ai_session) bridge_module._process_bridge_row(session, row) session.commit() finally: session.close() def test_audio_bridge_requested_registers_media_runtime(monkeypatch, tmp_path): recorded: list[dict] = [] monkeypatch.setattr( bridge_module, "_register_voice_ai_media_bridge", lambda **kwargs: recorded.append(kwargs) or {"ok": True}, ) call_id = f"call_audio_bridge_{tmp_path.name}" session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id=f"linked_audio_bridge_{tmp_path.name}", queue_code="voice_lab", queue_id="que_voice_lab", interaction_id=f"int_audio_bridge_{tmp_path.name}", caller_number="+77010000051", caller_name="Audio Bridge Caller", status="active", telephony_status="ringing", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name=None, voice_session_id=f"avs_audio_bridge_{tmp_path.name}", ai_session_id=f"ais_audio_bridge_{tmp_path.name}", ai_state="queued", ai_handoff_reason=None, ai_last_model_at=None, started_at=utc_now_iso(), connected_at=None, ended_at=None, updated_at=utc_now_iso(), ) ) session.commit() row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCAIAudioBridgeRequested", call_id=call_id, linked_id=f"linked_audio_bridge_{tmp_path.name}", payload={ "UserEvent": "MVPCCAIAudioBridgeRequested", "CallID": call_id, "LinkedID": f"linked_audio_bridge_{tmp_path.name}", "MediaUUID": "d64fa2f7-0063-4fe1-a36f-e1b372980ec1", "AudioSocketService": "127.0.0.1:9019", "Channel": "PJSIP/1001-00000011", }, ) bridge_module._process_bridge_row(session, row) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.ai_state == "greeting" finally: session.close() assert recorded == [ { "voice_session_id": f"avs_audio_bridge_{tmp_path.name}", "event_type": "requested", "media_uuid": "d64fa2f7-0063-4fe1-a36f-e1b372980ec1", "call_id": call_id, "linked_id": f"linked_audio_bridge_{tmp_path.name}", "channel": "PJSIP/1001-00000011", "service_address": "127.0.0.1:9019", } ] def test_audio_bridge_requested_recovers_voice_session_from_db(monkeypatch, tmp_path): recorded: list[dict] = [] monkeypatch.setattr( bridge_module, "_register_voice_ai_media_bridge", lambda **kwargs: recorded.append(kwargs) or {"ok": True}, ) call_id = f"call_audio_bridge_recover_{tmp_path.name}" linked_id = f"linked_audio_bridge_recover_{tmp_path.name}" voice_session_id = f"avs_audio_bridge_recover_{tmp_path.name}" ai_session_id = f"ais_audio_bridge_recover_{tmp_path.name}" interaction_id = f"int_audio_bridge_recover_{tmp_path.name}" session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id=linked_id, queue_code="voice_lab_ai", queue_id="que_voice_lab", interaction_id=interaction_id, caller_number="+77010000061", caller_name="Recovered Audio Bridge Caller", status="active", telephony_status="ringing", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name=None, voice_session_id=None, ai_session_id=None, ai_state="queued", ai_handoff_reason=None, ai_last_model_at=None, started_at=utc_now_iso(), connected_at=None, ended_at=None, updated_at=utc_now_iso(), ) ) session.add( VoiceAISessionRow( session_id=voice_session_id, call_id=call_id, linked_id=linked_id, interaction_id=interaction_id, customer_id=None, queue_id="que_voice_lab", ai_session_id=ai_session_id, agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="greeting", handoff_reason=None, handoff_target_queue_id="que_voice_lab", 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() row = bridge_module._create_bridge_log( session, ami_event_name="MVPCCAIAudioBridgeRequested", call_id=call_id, linked_id=linked_id, payload={ "UserEvent": "MVPCCAIAudioBridgeRequested", "CallID": call_id, "LinkedID": linked_id, "MediaUUID": "9f23dd6f-3cd7-4aaf-8a4d-71154a3626d5", "AudioSocketService": "127.0.0.1:9019", "Channel": "PJSIP/1001-00000021", }, ) bridge_module._process_bridge_row(session, row) session.commit() link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.voice_session_id == voice_session_id assert link.ai_session_id == ai_session_id assert link.ai_state == "greeting" finally: session.close() assert recorded == [ { "voice_session_id": voice_session_id, "event_type": "requested", "media_uuid": "9f23dd6f-3cd7-4aaf-8a4d-71154a3626d5", "call_id": call_id, "linked_id": linked_id, "channel": "PJSIP/1001-00000021", "service_address": "127.0.0.1:9019", } ] def test_voice_ai_update_call_state_updates_live_link(tmp_path): call_id = f"call_voice_state_{tmp_path.name}" session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id=f"linked_voice_state_{tmp_path.name}", queue_code="voice_lab", queue_id="que_voice_lab", interaction_id=f"int_voice_state_{tmp_path.name}", caller_number="+77010000052", caller_name="State Caller", status="active", telephony_status="connected", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-00000012", voice_session_id=f"avs_voice_state_{tmp_path.name}", ai_session_id=f"ais_voice_state_{tmp_path.name}", ai_state="greeting", ai_handoff_reason=None, ai_last_model_at=None, started_at=utc_now_iso(), connected_at=utc_now_iso(), ended_at=None, updated_at=utc_now_iso(), ) ) session.commit() finally: session.close() result = bridge_module._update_voice_ai_call_state( call_id, VoiceAICallStateUpdateIn( voice_session_id=f"avs_voice_state_{tmp_path.name}", ai_session_id=f"ais_voice_state_{tmp_path.name}", ai_state="speaking", handoff_reason=None, metadata={"media_uuid": "11111111-1111-1111-1111-111111111111"}, ), actor={"auth_source": "service", "sub": "svc:ai-voice-runtime", "user": "ai-voice-runtime", "role": "admin"}, ) assert result.ai_state == "speaking" session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.ai_state == "speaking" assert link.ai_last_model_at is not None finally: session.close() def test_live_call_ai_summary_route_returns_handoff_context(tmp_path): call_id = f"call_voice_summary_{tmp_path.name}" session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id=f"linked_voice_summary_{tmp_path.name}", queue_code="voice_lab", queue_id="que_voice_lab", interaction_id=f"int_voice_summary_{tmp_path.name}", caller_number="+77010000042", caller_name="Summary Caller", status="active", telephony_status="ringing", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-0000042", voice_session_id=f"avs_voice_summary_{tmp_path.name}", ai_session_id=f"ais_voice_summary_{tmp_path.name}", ai_state="handoff_required", ai_handoff_reason="Нужен живой оператор.", ai_last_model_at=utc_now_iso(), started_at=utc_now_iso(), connected_at=None, ended_at=None, updated_at=utc_now_iso(), ) ) session.add( VoiceAISessionRow( session_id=f"avs_voice_summary_{tmp_path.name}", call_id=call_id, linked_id=f"linked_voice_summary_{tmp_path.name}", interaction_id=f"int_voice_summary_{tmp_path.name}", customer_id=None, queue_id="que_voice_lab", ai_session_id=f"ais_voice_summary_{tmp_path.name}", agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="openai", status="handoff_requested", handoff_reason="Нужен живой оператор.", handoff_target_queue_id="que_voice_lab", disclosure_played_at=utc_now_iso(), last_user_utterance_at=utc_now_iso(), last_ai_reply_at=utc_now_iso(), started_at=utc_now_iso(), updated_at=utc_now_iso(), ended_at=None, ) ) session.add( AISessionRow( session_id=f"ais_voice_summary_{tmp_path.name}", channel="voice", call_id=call_id, thread_id=None, interaction_id=f"int_voice_summary_{tmp_path.name}", customer_id=None, agent_profile="voice_support", language="ru", status="handoff_required", summary_text="AI собрал контекст и передал звонок оператору.", last_user_message_id=None, last_ai_message_id=None, handoff_reason="Нужен живой оператор.", created_at=utc_now_iso(), updated_at=utc_now_iso(), closed_at=None, ) ) session.add( VoiceTranscriptSegmentRow( segment_id=f"vts_voice_summary_customer_{tmp_path.name}", session_id=f"avs_voice_summary_{tmp_path.name}", call_id=call_id, interaction_id=f"int_voice_summary_{tmp_path.name}", speaker="caller", source_type="asr", sequence_no=1, text="Соедините меня с оператором", confidence=0.91, is_final=True, barge_in_interrupted=False, payload_json="{}", created_at=utc_now_iso(), ) ) session.add( VoiceTranscriptSegmentRow( segment_id=f"vts_voice_summary_ai_{tmp_path.name}", session_id=f"avs_voice_summary_{tmp_path.name}", call_id=call_id, interaction_id=f"int_voice_summary_{tmp_path.name}", speaker="assistant", source_type="tts", sequence_no=2, text="Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора.", confidence=None, is_final=True, barge_in_interrupted=False, payload_json="{}", created_at=utc_now_iso(), ) ) session.add( VoiceTranscriptSegmentRow( segment_id=f"vts_voice_summary_ai_pending_{tmp_path.name}", session_id=f"avs_voice_summary_{tmp_path.name}", call_id=call_id, interaction_id=f"int_voice_summary_{tmp_path.name}", speaker="assistant", source_type="tts", sequence_no=3, text="Плановая недоставленная реплика", confidence=None, is_final=False, barge_in_interrupted=True, payload_json="{}", created_at=utc_now_iso(), ) ) session.commit() finally: session.close() def test_voice_ai_handoff_revives_false_ended_call_when_channel_is_resolvable(monkeypatch, tmp_path): monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"ivr_support":"que_ivr_support","voice_lab_ai":"que_voice_lab"}') monkeypatch.setenv("ASTERISK_TRANSFER_TARGET_MAP_JSON", '{"ivr_support":"2001","voice_lab_ai":"2001"}') call_id = f"call_ai_handoff_revive_{tmp_path.name}" interaction_id = f"int_ai_handoff_revive_{tmp_path.name}" session_id = f"avs_ai_handoff_revive_{tmp_path.name}" ai_session_id = f"ais_ai_handoff_revive_{tmp_path.name}" now = utc_now_iso() session = get_session() try: session.add( AsteriskCallLinkRow( call_id=call_id, linked_id=f"linked_ai_handoff_revive_{tmp_path.name}", queue_code="voice_lab_ai", queue_id="que_voice_lab", interaction_id=interaction_id, caller_number="+77010000077", caller_name="Track16 Handoff Revive", status="ended", telephony_status="ended", claimed_by_user=None, claimed_at=None, operator_extension=None, channel_name="PJSIP/1001-0000077", voice_session_id=session_id, ai_session_id=ai_session_id, ai_state="handoff_requested", ai_handoff_reason="Нужен живой оператор.", ai_last_model_at=now, started_at=now, connected_at=None, ended_at=now, updated_at=now, ) ) session.add( VoiceAISessionRow( session_id=session_id, call_id=call_id, linked_id=f"linked_ai_handoff_revive_{tmp_path.name}", interaction_id=interaction_id, customer_id=None, queue_id="que_voice_lab", ai_session_id=ai_session_id, agent_profile="voice_support", language="ru", asr_provider="openai", tts_provider="yandex", status="handoff_requested", handoff_reason="Нужен живой оператор.", handoff_target_queue_id="que_ivr_support", media_uuid="22222222-2222-2222-2222-222222222222", media_status="connected", media_connected_at=now, media_ended_at=None, last_media_frame_at=now, disclosure_played_at=now, last_user_utterance_at=now, last_ai_reply_at=now, started_at=now, updated_at=now, ended_at=None, ) ) session.commit() finally: session.close() ami_calls: list[tuple[str, dict]] = [] monkeypatch.setattr( bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: ["PJSIP/1001-0000077"], ) monkeypatch.setattr( bridge_module, "_ami_action", lambda action, params: ami_calls.append((action, params)) or {"action_id": "ami_handoff_revive"}, ) result = bridge_module._request_voice_ai_handoff( call_id, VoiceAIHandoffRequestIn( voice_session_id=session_id, ai_session_id=ai_session_id, interaction_id=interaction_id, target_queue_id="que_ivr_support", reason="Нужен живой оператор.", summary={"customer_request_text": "Соедините меня с оператором"}, ), actor={"auth_source": "service", "sub": "svc:ai-voice-runtime", "user": "ai-voice-runtime", "role": "admin"}, ) assert result.call_id == call_id assert ami_calls assert [call[0] for call in ami_calls[:4]] == ["Setvar", "Setvar", "Setvar", "Setvar"] assert ami_calls[0][1]["Variable"] == "MVPCC_START_LANGUAGE" assert ami_calls[3][1]["Variable"] == "MVPCC_CUSTOMER_NAME_SOURCE" assert ami_calls[-1][0] == "Redirect" assert ami_calls[-1][1]["Channel"] == "PJSIP/1001-0000077" assert ami_calls[-1][1]["Exten"] == "2001" payload = { "status_label": "AI передал звонок оператору", "customer_request_text": "Соедините меня с оператором", "ai_outcome_text": "Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора.", "handoff_reason": "Нужен живой оператор.", } session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id) ).scalar_one() assert link.status == "active" assert link.telephony_status == "connected" assert link.ended_at is None assert link.ai_state == "handoff_required" finally: session.close() assert payload["status_label"] == "AI передал звонок оператору" assert payload["customer_request_text"] == "Соедините меня с оператором" assert payload["ai_outcome_text"] == "Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора." assert payload["handoff_reason"] == "Нужен живой оператор."