From 179b2adee4c201dec17c03896defbf1af910cbba Mon Sep 17 00:00:00 2001 From: Tejas Narayan Date: Tue, 1 Sep 2026 16:25:49 +0000 Subject: [PATCH 1/5] feat: updating plivo imports --- .serena/project.local.yml | 5 + app/api/v1/api.py | 1 + app/api/v1/media.py | 2 +- app/api/v1/routes/call_imports.py | 38 +- app/api/v1/routes/telephony.py | 40 ++- app/api/v1/routes/vobiz_telephony.py | 21 +- app/api/v1/routes/voice_agent.py | 8 + app/app_factory.py | 2 +- .../telephony/call_recording_lifecycle.py | 4 +- .../telephony/inbound_stream_answer.py | 134 +++++++ .../telephony/number_import_service.py | 123 +++++-- app/services/telephony/plivo_client.py | 23 +- app/services/telephony/plivo_webhook_urls.py | 35 ++ app/services/telephony/plivo_xml.py | 2 +- app/services/telephony/telephony_service.py | 65 ++-- app/services/telephony/vobiz_agent_context.py | 73 +++- app/services/telephony/webhook_auth.py | 54 ++- .../voice_agent/llm_voice_providers.py | 339 ++++++++++++++++++ app/services/voice_agent/voice_bundle.py | 74 ++-- .../telephony_credential_rate_limit.py | 5 +- app/workers/tasks/process_call_import_row.py | 16 +- docs/telephony-media.md | 2 +- .../src/pages/agents/AgentWorkspaceDetail.tsx | 64 ++-- frontend/src/pages/agents/AgentsWorkspace.tsx | 1 + .../pages/agents/components/AgentEditForm.tsx | 54 +-- pytest_out1.txt | 29 ++ tests/conftest.py | 2 +- tests/test_api/test_call_imports_routes.py | 24 ++ tests/test_api/test_telephony_webhooks.py | 131 +++++-- tests/test_api/test_vobiz_telephony.py | 2 +- tests/test_app_factory_telephony_edge.py | 10 +- .../test_setup_flexprice_meters.py | 51 ++- .../test_evaluator_inbound_enqueue.py | 4 +- tests/test_services/test_media_urls.py | 14 +- .../test_number_import_service.py | 98 ++++- .../test_telephony/test_plivo_client.py | 36 ++ .../test_telephony/test_plivo_webhook_urls.py | 36 ++ .../test_telephony/test_plivo_xml.py | 29 ++ .../test_telephony/test_vobiz.py | 4 +- .../test_llm_voice_providers.py | 47 +++ .../test_process_call_import_row.py | 14 +- .../test_telephony_credential_rate_limit.py | 4 +- 42 files changed, 1459 insertions(+), 261 deletions(-) create mode 100644 .serena/project.local.yml create mode 100644 app/services/telephony/inbound_stream_answer.py create mode 100644 app/services/telephony/plivo_webhook_urls.py create mode 100644 app/services/voice_agent/llm_voice_providers.py create mode 100644 pytest_out1.txt create mode 100644 tests/test_services/test_telephony/test_plivo_client.py create mode 100644 tests/test_services/test_telephony/test_plivo_webhook_urls.py create mode 100644 tests/test_services/test_telephony/test_plivo_xml.py create mode 100644 tests/test_services/test_voice_agent/test_llm_voice_providers.py diff --git a/.serena/project.local.yml b/.serena/project.local.yml new file mode 100644 index 00000000..36cd3def --- /dev/null +++ b/.serena/project.local.yml @@ -0,0 +1,5 @@ +# This file allows you to locally override settings in project.yml for development purposes. +# +# Use the same keys as in project.yml here. Any setting you specify will override the corresponding +# setting in project.yml, allowing you to customise the configuration for your local development environment +# without affecting the project configuration in project.yml (which is intended to be versioned). diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 055bc147..e85452e2 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -87,6 +87,7 @@ api_router.include_router(prompt_partials.router) api_router.include_router(prompt_optimization.router) api_router.include_router(telephony.router) +api_router.include_router(telephony.plivo_webhook_router) api_router.include_router(vobiz_telephony.router) api_router.include_router(call_imports.router) api_router.include_router(call_import_schemas.router) diff --git a/app/api/v1/media.py b/app/api/v1/media.py index d35d99a3..56dcb536 100644 --- a/app/api/v1/media.py +++ b/app/api/v1/media.py @@ -6,5 +6,5 @@ media_router = APIRouter() media_router.include_router(vobiz_telephony.webhook_router) -media_router.include_router(vobiz_telephony.ws_router) +media_router.include_router(vobiz_telephony.carrier_ws_router) media_router.include_router(voice_agent.ws_router) diff --git a/app/api/v1/routes/call_imports.py b/app/api/v1/routes/call_imports.py index 70551508..f9a14ce7 100644 --- a/app/api/v1/routes/call_imports.py +++ b/app/api/v1/routes/call_imports.py @@ -1360,11 +1360,14 @@ def _validate_direct_url_import_ready( ) -def _validate_exotel_import_ready( +def _validate_credentialed_import_ready( + provider: str, parameters: List[CallImportSchemaParameter], parameter_mapping: Dict[str, Any], ) -> None: - """Ensure Exotel credentialed import has a mapped recording_url column.""" + """Ensure credentialed import has a mapped recording_url column.""" + provider_key = (provider or "").lower() + provider_label = provider_key.capitalize() rec_url_param = next( ( p @@ -1377,7 +1380,7 @@ def _validate_exotel_import_ready( raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( - "Exotel import requires a schema parameter of type " + f"{provider_label} import requires a schema parameter of type " "'recording_url'." ), ) @@ -1386,12 +1389,20 @@ def _validate_exotel_import_ready( raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( - "Exotel import requires the 'recording_url' parameter to " + f"{provider_label} import requires the 'recording_url' parameter to " "be mapped to a source column." ), ) +def _validate_exotel_import_ready( + parameters: List[CallImportSchemaParameter], + parameter_mapping: Dict[str, Any], +) -> None: + """Ensure Exotel credentialed import has a mapped recording_url column.""" + _validate_credentialed_import_ready("exotel", parameters, parameter_mapping) + + def _is_manual_audio_call_import(call_import: CallImport) -> bool: """True for batches created via manual audio upload (recordings already in S3).""" return (call_import.source_format or "").lower() == "audio" @@ -2150,9 +2161,11 @@ async def start_call_import( payload.provider or "", ) _validate_telephony_credentials_live(db, organization_id, integration) - if (integration.provider or "").lower() == "exotel": - _validate_exotel_import_ready( - parameters, dict(call_import.parameter_mapping or {}) + if (integration.provider or "").lower() in {"exotel", "plivo"}: + _validate_credentialed_import_ready( + integration.provider, + parameters, + dict(call_import.parameter_mapping or {}), ) else: _validate_direct_url_import_ready( @@ -2345,8 +2358,10 @@ async def upload_call_import_csv( integration = _resolve_telephony_integration( db, organization_id, telephony_integration_id, provider or "" ) - if (integration.provider or "").lower() == "exotel": - _validate_exotel_import_ready(parameters, cleaned_mapping) + if (integration.provider or "").lower() in {"exotel", "plivo"}: + _validate_credentialed_import_ready( + integration.provider, parameters, cleaned_mapping + ) else: _validate_direct_url_import_ready(parameters, cleaned_mapping) integration = None @@ -3527,14 +3542,15 @@ async def retry_failed_call_import_rows( _validate_telephony_credentials_live(db, organization_id, integration) call_import.provider = integration.provider call_import.telephony_integration_id = integration.id - if (integration.provider or "").lower() == "exotel": + if (integration.provider or "").lower() in {"exotel", "plivo"}: schema = _resolve_schema( db, organization_id, call_import.workspace_id, call_import.schema_id, ) - _validate_exotel_import_ready( + _validate_credentialed_import_ready( + integration.provider, list(schema.parameters), dict(call_import.parameter_mapping or {}), ) diff --git a/app/api/v1/routes/telephony.py b/app/api/v1/routes/telephony.py index 6d4f8b9a..be00d481 100644 --- a/app/api/v1/routes/telephony.py +++ b/app/api/v1/routes/telephony.py @@ -33,6 +33,7 @@ from app.services.telephony.platform_outbound_pool import outbound_pool_api_payload router = APIRouter(prefix="/telephony", tags=["Telephony"]) +plivo_webhook_router = APIRouter(prefix="/telephony/plivo", tags=["Plivo Telephony Webhooks"]) class TelephonyAvailableNumberResponse(BaseModel): @@ -557,25 +558,52 @@ async def _read_webhook_params(request: Request) -> Dict[str, Any]: return params -@router.post("/webhooks/answer") -async def telephony_answer_webhook(request: Request, db: Session = Depends(get_db)): +async def _handle_plivo_answer_webhook(request: Request, db: Session) -> Response: params = await _read_webhook_params(request) verify_plivo_webhook(request, params, "answer", db) xml = telephony_service.handle_answer_webhook(params, db) return Response(content=xml, media_type="application/xml") -@router.post("/webhooks/events") -async def telephony_events_webhook(request: Request, db: Session = Depends(get_db)): +async def _handle_plivo_events_webhook(request: Request, db: Session) -> Dict[str, str]: params = await _read_webhook_params(request) verify_plivo_webhook(request, params, "events", db) telephony_service.handle_event_webhook(params, db) return {"status": "ok"} -@router.post("/webhooks/masking") -async def telephony_masking_webhook(request: Request, db: Session = Depends(get_db)): +async def _handle_plivo_masking_webhook(request: Request, db: Session) -> Response: params = await _read_webhook_params(request) verify_plivo_webhook(request, params, "masking", db) xml = telephony_service.handle_masking_webhook(params, db) return Response(content=xml, media_type="application/xml") + + +@plivo_webhook_router.post("/webhooks/answer") +async def plivo_answer_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_answer_webhook(request, db) + + +@plivo_webhook_router.post("/webhooks/events") +async def plivo_events_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_events_webhook(request, db) + + +@plivo_webhook_router.post("/webhooks/masking") +async def plivo_masking_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_masking_webhook(request, db) + + +@router.post("/webhooks/answer") +async def telephony_answer_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_answer_webhook(request, db) + + +@router.post("/webhooks/events") +async def telephony_events_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_events_webhook(request, db) + + +@router.post("/webhooks/masking") +async def telephony_masking_webhook(request: Request, db: Session = Depends(get_db)): + return await _handle_plivo_masking_webhook(request, db) diff --git a/app/api/v1/routes/vobiz_telephony.py b/app/api/v1/routes/vobiz_telephony.py index 052d03b9..08355d81 100644 --- a/app/api/v1/routes/vobiz_telephony.py +++ b/app/api/v1/routes/vobiz_telephony.py @@ -19,7 +19,7 @@ from app.services.telephony.phone_routing import resolve_inbound_agent_for_number from app.services.telephony.plivo_client import normalize_e164 from app.services.telephony.vobiz_agent_context import ( - build_vobiz_ws_url, + build_carrier_ws_url, extract_webhook_params, resolve_vobiz_agent_context, vobiz_webhook_base_url, @@ -57,7 +57,8 @@ router = APIRouter(prefix="/telephony/vobiz", tags=["Vobiz Telephony"]) webhook_router = APIRouter(prefix="/telephony/vobiz", tags=["Vobiz Telephony Webhooks"]) -ws_router = APIRouter(prefix="/telephony/vobiz", tags=["Vobiz Telephony Media"]) +carrier_ws_router = APIRouter(prefix="/telephony/carrier", tags=["Carrier Telephony Media"]) +ws_router = carrier_ws_router class VobizOutboundCallRequest(BaseModel): @@ -478,7 +479,7 @@ async def vobiz_answer_webhook( if session_token and call_uuid: link_provider_call_id(db, call_ref=session_token, provider_call_id=call_uuid) - ws_url = build_vobiz_ws_url( + ws_url = build_carrier_ws_url( agent_id=str(agent_id), session=session_token, persona_id=persona_id, @@ -612,8 +613,8 @@ async def vobiz_recording_ready_webhook( return {"status": "ok"} -@ws_router.websocket("/ws") -async def vobiz_media_websocket(websocket: WebSocket): +@carrier_ws_router.websocket("/ws") +async def carrier_media_websocket(websocket: WebSocket): agent_id = websocket.query_params.get("agent_id") session_token = websocket.query_params.get("session") persona_id = websocket.query_params.get("persona_id") @@ -655,7 +656,7 @@ async def vobiz_media_websocket(websocket: WebSocket): # endregion if not call_short_id: logger.warning( - "No CallRecording for Vobiz session {}; live transcript and recording will not be linked", + "No CallRecording for carrier media session {}; live transcript and recording will not be linked", session_token, ) try: @@ -707,6 +708,8 @@ async def vobiz_media_websocket(websocket: WebSocket): stt_api_key=context.stt_api_key, tts_api_key=context.tts_api_key, llm_api_key=context.llm_api_key, + llm_endpoint_url=context.llm_endpoint_url, + llm_base_url=context.llm_base_url, serializer=serializer, telephony_mode=True, call_short_id=call_short_id, @@ -734,12 +737,12 @@ async def vobiz_media_websocket(websocket: WebSocket): silence_hangup_secs=hangup_secs, ) except ValueError as e: - logger.error("Vobiz media websocket setup failed: {}", e) + logger.error("Carrier media websocket setup failed: {}", e) await websocket.close(code=1011, reason=str(e)) except WebSocketDisconnect: - logger.info("Vobiz media websocket disconnected") + logger.info("Carrier media websocket disconnected") except Exception as e: - logger.error("Vobiz media websocket error: {}", e, exc_info=True) + logger.error("Carrier media websocket error: {}", e, exc_info=True) try: await websocket.close(code=1011, reason="Server error") except Exception: diff --git a/app/api/v1/routes/voice_agent.py b/app/api/v1/routes/voice_agent.py index 9657c763..5dbcda08 100644 --- a/app/api/v1/routes/voice_agent.py +++ b/app/api/v1/routes/voice_agent.py @@ -503,6 +503,13 @@ def resolve_azure_endpoint_for_provider(provider: ModelProvider) -> str | None: ).lower() == "azure" else None ) + from app.services.voice_agent.llm_voice_providers import resolve_voice_llm_base_url + + llm_base_url = ( + resolve_voice_llm_base_url(db, organization_id, voice_bundle, llm_provider) + if llm_provider and voice_bundle + else None + ) # If in bridge mode, we need to bridge test agent to Retell call # For now, we'll run the voice bundle normally and note that bridging @@ -534,6 +541,7 @@ def resolve_azure_endpoint_for_provider(provider: ModelProvider) -> str | None: tts_api_key=tts_api_key, llm_api_key=llm_api_key, llm_endpoint_url=llm_endpoint_url, + llm_base_url=llm_base_url, silence_hangup_secs=agent_silence_hangup_secs, ) else: diff --git a/app/app_factory.py b/app/app_factory.py index c2af3e09..4a83d2a2 100644 --- a/app/app_factory.py +++ b/app/app_factory.py @@ -215,7 +215,7 @@ def create_app() -> FastAPI: ) if _service_mode() == "media": logger.info( - "Vobiz telephony edge: /telephony/vobiz/webhooks/* and /telephony/vobiz/ws" + "Carrier telephony edge: /telephony/vobiz/webhooks/* and /telephony/carrier/ws" ) if settings.SERVICE_MODE == "api": logger.info( diff --git a/app/services/telephony/call_recording_lifecycle.py b/app/services/telephony/call_recording_lifecycle.py index f5121e36..a7c417a0 100644 --- a/app/services/telephony/call_recording_lifecycle.py +++ b/app/services/telephony/call_recording_lifecycle.py @@ -183,7 +183,6 @@ def _normalize_call_event(status: Optional[str]) -> str: def _find_by_call_ref(db: Session, call_ref: str) -> Optional[CallRecording]: rows = ( db.query(CallRecording) - .filter(CallRecording.provider_platform == "vobiz") .order_by(CallRecording.created_at.desc()) .limit(200) .all() @@ -237,6 +236,7 @@ def create_inbound_call_recording( provider_call_id: Optional[str] = None, evaluator_id: Optional[UUID] = None, evaluator_result_id: Optional[UUID] = None, + provider_platform: str = "vobiz", ) -> CallRecording: existing = find_call_recording(db, call_ref=call_ref, provider_call_id=provider_call_id) if existing: @@ -263,7 +263,7 @@ def create_inbound_call_recording( call_event="call_started", call_data=call_data, provider_call_id=provider_call_id, - provider_platform="vobiz", + provider_platform=provider_platform, agent_id=agent.id, evaluator_result_id=evaluator_result_id, ) diff --git a/app/services/telephony/inbound_stream_answer.py b/app/services/telephony/inbound_stream_answer.py new file mode 100644 index 00000000..d5a19a8f --- /dev/null +++ b/app/services/telephony/inbound_stream_answer.py @@ -0,0 +1,134 @@ +"""Build WebSocket stream XML for inbound carrier answer webhooks.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional +from uuid import UUID + +from loguru import logger +from sqlalchemy.orm import Session + +from app.config import settings +from app.models.database import Agent +from app.services.telephony.call_recording_lifecycle import ( + create_inbound_call_recording, + link_provider_call_id, +) +from app.services.telephony.phone_routing import resolve_inbound_agent_for_number +from app.services.telephony.plivo_xml import reject_call, speak_and_hangup +from app.services.telephony.vobiz_agent_context import build_carrier_ws_url, vobiz_webhook_base_url +from app.services.telephony.vobiz_session import create_call_session +from app.services.telephony.vobiz_xml import stream_to_agent + + +def build_inbound_stream_answer_xml( + db: Session, + params: Dict[str, Any], + *, + provider_platform: str = "plivo", +) -> str: + """Return Plivo/Vobiz-compatible XML that streams inbound audio to the voice agent.""" + to_number = params.get("To") or params.get("to") + from_number = params.get("From") or params.get("from") + call_uuid = ( + params.get("CallUUID") + or params.get("CallSid") + or params.get("call_sid") + or params.get("Sid") + ) + + if not to_number: + return speak_and_hangup("Call could not be routed.") + + agent_id, organization_id = resolve_inbound_agent_for_number(db, to_number) + if not agent_id or not organization_id: + logger.warning( + "Inbound stream answer miss: to={} from={} call_uuid={}", + to_number, + from_number, + call_uuid, + ) + return reject_call("No active routing found for this number.") + + agent = db.query(Agent).filter(Agent.id == agent_id).first() + if not agent: + return reject_call("No active routing found for this number.") + + inbound_evaluator_id: Optional[UUID] = None + inbound_evaluator_result_id: Optional[UUID] = None + inbound_persona_id: Optional[UUID] = None + inbound_scenario_id: Optional[UUID] = None + if agent.workspace_id: + from app.services.evaluators.evaluator_inbound_service import ( + consume_inbound_evaluator_combination, + create_inbound_evaluator_result, + find_inbound_suite_for_agent, + ) + + suite = find_inbound_suite_for_agent( + db, agent, organization_id, agent.workspace_id + ) + if suite: + selected, _idx, _next_idx = consume_inbound_evaluator_combination(db, suite) + inbound_evaluator_id = selected.id + inbound_persona_id = selected.persona_id + inbound_scenario_id = selected.scenario_id + result_row = create_inbound_evaluator_result( + db, + organization_id, + agent.workspace_id, + selected, + ) + inbound_evaluator_result_id = result_row.id + + session = create_call_session( + agent_id=str(agent_id), + organization_id=str(organization_id), + direction="inbound", + from_number=from_number, + to_number=to_number, + persona_id=str(inbound_persona_id) if inbound_persona_id else None, + scenario_id=str(inbound_scenario_id) if inbound_scenario_id else None, + evaluator_id=str(inbound_evaluator_id) if inbound_evaluator_id else None, + ) + + create_inbound_call_recording( + db, + agent=agent, + organization_id=organization_id, + call_ref=session.call_ref, + from_number=from_number, + to_number=to_number, + provider_call_id=call_uuid, + evaluator_id=inbound_evaluator_id, + evaluator_result_id=inbound_evaluator_result_id, + provider_platform=provider_platform, + ) + + if call_uuid: + link_provider_call_id(db, call_ref=session.call_ref, provider_call_id=call_uuid) + + ws_url = build_carrier_ws_url( + agent_id=str(agent_id), + session=session.call_ref, + persona_id=str(inbound_persona_id) if inbound_persona_id else None, + scenario_id=str(inbound_scenario_id) if inbound_scenario_id else None, + ) + + record_action_url = None + if settings.VOBIZ_CARRIER_SESSION_RECORDING and provider_platform == "vobiz": + record_action_url = ( + f"{vobiz_webhook_base_url()}{settings.API_V1_PREFIX}/telephony/vobiz/webhooks/recording-ready" + f"?call_ref={session.call_ref}" + ) + + logger.info( + "Inbound stream answer agent_id={} session={} to={} from={} call_uuid={}", + agent_id, + session.call_ref, + to_number, + from_number, + call_uuid, + ) + + return stream_to_agent(ws_url, record_action_url=record_action_url) diff --git a/app/services/telephony/number_import_service.py b/app/services/telephony/number_import_service.py index a5389527..7711af3a 100644 --- a/app/services/telephony/number_import_service.py +++ b/app/services/telephony/number_import_service.py @@ -13,6 +13,11 @@ from app.models.enums import TelephonyProvider from app.services.telephony.phone_routing import sync_agent_telephony_number_link from app.services.telephony.plivo_client import PlivoClient, expand_phone_candidates, normalize_e164 +from app.services.telephony.plivo_webhook_urls import ( + legacy_answer_webhook_url, + plivo_answer_webhook_url, + plivo_hangup_webhook_url, +) from app.services.telephony.telephony_service import telephony_service from app.services.telephony.vobiz_agent_context import vobiz_webhook_base_url from app.services.telephony.vobiz_client import VobizClient, build_vobiz_client_for_org @@ -42,21 +47,15 @@ def _vobiz_answer_webhook_url() -> str: def _plivo_answer_webhook_url() -> str: - base = (settings.PLIVO_WEBHOOK_BASE_URL or "").strip().rstrip("/") - if not base: - raise ValueError( - "Plivo webhook base URL is not configured. Set plivo.webhook_base_url in platform config." - ) - return f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/answer" + return plivo_answer_webhook_url() def _plivo_hangup_webhook_url() -> str: - base = (settings.PLIVO_WEBHOOK_BASE_URL or "").strip().rstrip("/") - return f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/events" + return plivo_hangup_webhook_url() def _exotel_voice_webhook_url() -> str: - return _plivo_answer_webhook_url() + return legacy_answer_webhook_url() def _normalize_remote_number(provider: str, item: Dict[str, Any]) -> Optional[str]: @@ -161,27 +160,111 @@ def _list_remote_numbers( raise ValueError(f"Unsupported provider: {provider}") -def _remote_metadata(provider: str, item: Dict[str, Any]) -> Dict[str, Any]: +def _normalize_country_iso2( + raw: Optional[str], + *, + e164: Optional[str] = None, +) -> Optional[str]: + """Normalize provider country values to ISO-3166 alpha-2 for storage.""" + value = (raw or "").strip() + if not value: + return _country_iso2_from_e164(e164) + + if len(value) == 2 and value.isalpha(): + return value.upper() + + mapped = _COUNTRY_NAME_TO_ISO2.get(value.lower()) + if mapped: + return mapped + + return _country_iso2_from_e164(e164) + + +def _country_iso2_from_e164(e164: Optional[str]) -> Optional[str]: + if not e164: + return None + digits = str(e164).strip().lstrip("+") + if not digits.isdigit(): + return None + for length in (3, 2, 1): + prefix = digits[:length] + iso2 = _CALLING_CODE_TO_ISO2.get(prefix) + if iso2: + return iso2 + return None + + +def _extract_application_id(value: Optional[str]) -> Optional[str]: + """Extract a Plivo-style application id from a URI or raw id.""" + if not value: + return None + text = str(value).strip().rstrip("/") + if "/Application/" in text: + tail = text.rsplit("/Application/", 1)[-1] + app_id = tail.split("/", 1)[0].strip() + return app_id or None + return text or None + + +_COUNTRY_NAME_TO_ISO2 = { + "india": "IN", + "united states": "US", + "united states of america": "US", + "usa": "US", + "united kingdom": "GB", + "uk": "GB", + "canada": "CA", + "australia": "AU", + "germany": "DE", + "france": "FR", + "singapore": "SG", +} + +_CALLING_CODE_TO_ISO2 = { + "91": "IN", + "1": "US", + "44": "GB", + "61": "AU", + "49": "DE", + "33": "FR", + "65": "SG", + "971": "AE", +} + + +def _remote_metadata( + provider: str, + item: Dict[str, Any], + *, + e164: Optional[str] = None, +) -> Dict[str, Any]: provider_key = provider.lower() if provider_key == TelephonyProvider.EXOTEL.value: + raw_country = item.get("Country") or item.get("country") return { "provider_number_id": item.get("Sid") or item.get("sid"), - "country": item.get("Country") or item.get("country"), + "country": raw_country, + "country_iso2": _normalize_country_iso2(raw_country, e164=e164), "region": item.get("Region") or item.get("region") or item.get("FriendlyName"), "capabilities": None, "status": item.get("Status") or item.get("status"), "application_id": item.get("VoiceUrl") or item.get("voice_url"), } + raw_country = item.get("country") or item.get("Country") + application_id = _extract_application_id( + item.get("application_id") + or item.get("app_id") + or item.get("Application") + or item.get("application") + ) return { "provider_number_id": item.get("id") or item.get("Sid") or item.get("sid"), - "country": item.get("country") or item.get("Country"), + "country": raw_country, + "country_iso2": _normalize_country_iso2(raw_country, e164=e164), "region": item.get("region") or item.get("Region"), "capabilities": item.get("capabilities"), "status": item.get("status") or item.get("Status"), - "application_id": item.get("application_id") - or item.get("app_id") - or item.get("Application") - or item.get("application"), + "application_id": application_id, } @@ -250,7 +333,7 @@ def list_available_numbers( if not e164: continue imported = imported_by_phone.get(e164) - meta = _remote_metadata(provider_key, item) + meta = _remote_metadata(provider_key, item, e164=e164) results.append( { "e164": e164, @@ -295,7 +378,7 @@ def import_numbers( e164 = _normalize_remote_number(provider_key, item) if e164: enriched = dict(item) - enriched.update(_remote_metadata(provider_key, item)) + enriched.update(_remote_metadata(provider_key, item, e164=e164)) remote_by_e164[e164] = enriched agent: Optional[Agent] = None @@ -374,7 +457,7 @@ def import_numbers( row.inbound_enabled = True row.outbound_enabled = True row.source = "imported" - row.country_iso2 = remote.get("country") or row.country_iso2 + row.country_iso2 = remote.get("country_iso2") or row.country_iso2 row.region = remote.get("region") or row.region row.capabilities = remote.get("capabilities") or row.capabilities row.provider_app_id = remote.get("application_id") or row.provider_app_id @@ -385,7 +468,7 @@ def import_numbers( organization_id=org_id, telephony_integration_id=integration_id, phone_number=e164, - country_iso2=remote.get("country"), + country_iso2=remote.get("country_iso2"), region=remote.get("region"), capabilities=remote.get("capabilities"), provider_app_id=remote.get("application_id"), diff --git a/app/services/telephony/plivo_client.py b/app/services/telephony/plivo_client.py index 8e750c18..499a2ee1 100644 --- a/app/services/telephony/plivo_client.py +++ b/app/services/telephony/plivo_client.py @@ -1,6 +1,6 @@ """Thin Plivo SDK wrapper for telephony operations.""" -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from loguru import logger try: @@ -85,12 +85,21 @@ def _add(value: Optional[str]) -> None: class PlivoClient: """Wrapper around plivo.RestClient that returns normalized dictionaries.""" - def __init__(self, auth_id: str, auth_token: str): + def __init__( + self, + auth_id: str, + auth_token: str, + *, + credential_fingerprint: Optional[str] = None, + ): if plivo is None: raise ValueError( "Plivo SDK is not installed. Install it with `pip install -e .` or `pip install plivo`." ) self.client = plivo.RestClient(auth_id=auth_id, auth_token=auth_token) + self._auth_id = auth_id + self._auth_token = auth_token + self._credential_fingerprint = credential_fingerprint @staticmethod def _to_dict(data: Any) -> Dict[str, Any]: @@ -211,6 +220,16 @@ def get_call_details(self, call_uuid: str) -> Dict[str, Any]: logger.exception("Failed to fetch call details") raise ValueError(f"Failed to fetch call details: {str(e)}") + def download_recording(self, recording_url: str) -> Tuple[bytes, str]: + """Download a recording from Plivo with HTTP Basic auth.""" + from app.services.telephony.recording_download import download_recording_url + + return download_recording_url( + recording_url, + auth=(self._auth_id, self._auth_token), + credential_fingerprint=self._credential_fingerprint, + ) + def start_voice_verification( self, recipient: str, app_uuid: str, callback_url: Optional[str] = None ) -> Dict[str, Any]: diff --git a/app/services/telephony/plivo_webhook_urls.py b/app/services/telephony/plivo_webhook_urls.py new file mode 100644 index 00000000..97c6d89d --- /dev/null +++ b/app/services/telephony/plivo_webhook_urls.py @@ -0,0 +1,35 @@ +"""Public HTTPS callback URLs for native Plivo telephony webhooks.""" + +from __future__ import annotations + +from app.config import settings + + +def plivo_webhook_base() -> str: + base = (settings.PLIVO_WEBHOOK_BASE_URL or "").strip().rstrip("/") + if not base: + raise ValueError( + "Plivo webhook base URL is not configured. Set plivo.webhook_base_url in platform config." + ) + return base + + +def plivo_answer_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/plivo/webhooks/answer" + + +def plivo_hangup_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/plivo/webhooks/events" + + +def plivo_masking_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/plivo/webhooks/masking" + + +def legacy_answer_webhook_url() -> str: + """Legacy generic path kept for Exotel and already-imported Plivo numbers.""" + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/webhooks/answer" + + +def legacy_events_webhook_url() -> str: + return f"{plivo_webhook_base()}{settings.API_V1_PREFIX}/telephony/webhooks/events" diff --git a/app/services/telephony/plivo_xml.py b/app/services/telephony/plivo_xml.py index deb01b7c..9fa44ae8 100644 --- a/app/services/telephony/plivo_xml.py +++ b/app/services/telephony/plivo_xml.py @@ -23,7 +23,7 @@ def dial_number(to_number: str, caller_id: str) -> str: """Build XML to dial a target number with caller ID.""" plivoxml = _get_plivoxml() response = plivoxml.ResponseElement() - dial = plivoxml.DialElement(callerId=caller_id) + dial = plivoxml.DialElement(caller_id=caller_id) dial.add(plivoxml.NumberElement(to_number)) response.add(dial) return response.to_string() diff --git a/app/services/telephony/telephony_service.py b/app/services/telephony/telephony_service.py index 6374ab49..a8037a98 100644 --- a/app/services/telephony/telephony_service.py +++ b/app/services/telephony/telephony_service.py @@ -27,7 +27,13 @@ from app.services.credentials.resolver import clear_other_defaults from app.services.telephony.exotel_client import build_exotel_client_from_integration from app.services.telephony.plivo_client import PlivoClient, normalize_e164 -from app.services.telephony.plivo_xml import dial_number, reject_call, speak_and_hangup +from app.services.telephony.plivo_webhook_urls import ( + legacy_events_webhook_url, + plivo_answer_webhook_url, + plivo_hangup_webhook_url, +) +from app.services.telephony.inbound_stream_answer import build_inbound_stream_answer_xml +from app.services.telephony.plivo_xml import reject_call from app.services.telephony.vobiz_client import build_vobiz_client_for_org from app.services.telephony.vobiz_xml import ( speak_and_hangup as vobiz_speak_and_hangup, @@ -88,7 +94,15 @@ def get_provider_client( auth_token = decrypt_api_key(integration.auth_token) provider_key = (integration.provider or provider or "").lower() if provider_key == "plivo": - return PlivoClient(auth_id=auth_id, auth_token=auth_token) + from app.workers.concurrency.telephony_credential_rate_limit import ( + fingerprint_for_integration, + ) + + return PlivoClient( + auth_id=auth_id, + auth_token=auth_token, + credential_fingerprint=fingerprint_for_integration(integration), + ) if provider_key == "exotel": from app.workers.concurrency.telephony_credential_rate_limit import ( fingerprint_for_integration, @@ -492,9 +506,8 @@ def initiate_outbound_call( raise ValueError("No active telephony integration found for from_number") client = self.get_provider_client(org_id, db, provider=integration.provider) - base = settings.PLIVO_WEBHOOK_BASE_URL.rstrip("/") - answer_url = f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/answer" - hangup_url = f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/events" + answer_url = plivo_answer_webhook_url() + hangup_url = plivo_hangup_webhook_url() response = client.create_outbound_call( from_=from_number, to_=to_number, answer_url=answer_url, hangup_url=hangup_url ) @@ -551,8 +564,7 @@ def start_voice_otp( recipient = normalize_e164(phone_number) callback_url = None if settings.PLIVO_WEBHOOK_BASE_URL: - base = settings.PLIVO_WEBHOOK_BASE_URL.rstrip("/") - callback_url = f"{base}{settings.API_V1_PREFIX}/telephony/webhooks/events" + callback_url = legacy_events_webhook_url() response = self.get_provider_client(org_id, db, provider=provider).start_voice_verification( recipient=recipient, app_uuid=app_uuid, callback_url=callback_url @@ -700,27 +712,17 @@ def handle_answer_webhook(self, params: Dict[str, Any], db: Session) -> str: or params.get("call_sid") or params.get("Sid") ) - logger.info("Telephony answer webhook call_uuid={} to={} from={}", call_uuid, to_number, from_number) - - if not to_number: - return speak_and_hangup("Call could not be routed.") - - to_number = normalize_e164(to_number) - number = db.query(TelephonyPhoneNumber).filter(TelephonyPhoneNumber.phone_number == to_number).first() - if not number: - return reject_call("This number is not configured.") - - if number.agent_id: - agent = db.query(Agent).filter(Agent.id == number.agent_id).first() - if agent and agent.phone_number: - try: - return dial_number(normalize_e164(agent.phone_number), to_number) - except ValueError: - logger.warning("Agent {} has non-E.164 phone number", agent.id) - - return speak_and_hangup("No active routing found for this number.") + logger.info( + "Telephony answer webhook call_uuid={} to={} from={}", + call_uuid, + to_number, + from_number, + ) + return build_inbound_stream_answer_xml(db, params, provider_platform="plivo") def handle_event_webhook(self, params: Dict[str, Any], db: Session) -> None: + from app.services.telephony.call_recording_lifecycle import update_call_from_vobiz_event + call_uuid = ( params.get("CallUUID") or params.get("RequestUUID") @@ -732,6 +734,15 @@ def handle_event_webhook(self, params: Dict[str, Any], db: Session) -> None: if not call_uuid: return + updated = update_call_from_vobiz_event( + db, + provider_call_id=call_uuid, + call_status=call_status, + payload=params, + ) + if updated: + return + row = db.query(CallRecording).filter(CallRecording.provider_call_id == call_uuid).first() if not row: return @@ -771,6 +782,8 @@ def handle_masking_webhook(self, params: Dict[str, Any], db: Session) -> str: return reject_call() target = session.party_b_number if from_number == session.party_a_number else session.party_a_number + from app.services.telephony.plivo_xml import dial_number + return dial_number(target, to_number) diff --git a/app/services/telephony/vobiz_agent_context.py b/app/services/telephony/vobiz_agent_context.py index 9a6970bc..2276bd04 100644 --- a/app/services/telephony/vobiz_agent_context.py +++ b/app/services/telephony/vobiz_agent_context.py @@ -37,6 +37,60 @@ class VobizAgentContext: stt_api_key: Optional[str] tts_api_key: Optional[str] llm_api_key: Optional[str] + llm_endpoint_url: Optional[str] = None + llm_base_url: Optional[str] = None + + +def _resolve_azure_endpoint_for_provider( + db: Session, + organization_id: UUID, + provider: ModelProvider, +) -> Optional[str]: + from app.services.ai.llm_service import _resolve_azure_endpoint_from_provider + + provider_value = provider.value if hasattr(provider, "value") else str(provider) + ai_provider_rec = db.query(AIProvider).filter( + AIProvider.organization_id == organization_id, + AIProvider.provider == provider_value, + AIProvider.is_active.is_(True), + ).first() + if not ai_provider_rec: + ai_provider_rec = db.query(AIProvider).filter( + AIProvider.organization_id == organization_id, + func.lower(AIProvider.provider) == provider_value.lower(), + AIProvider.is_active.is_(True), + ).first() + if not ai_provider_rec: + return None + return _resolve_azure_endpoint_from_provider(ai_provider_rec, None) + + +def _resolve_voice_llm_urls( + db: Session, + organization_id: UUID, + voice_bundle: Optional[VoiceBundle], +) -> tuple[Optional[str], Optional[str]]: + if not voice_bundle or not voice_bundle.llm_provider: + return None, None + + llm_provider = voice_bundle.llm_provider + provider_key = ( + llm_provider.value if hasattr(llm_provider, "value") else str(llm_provider) + ).lower() + llm_endpoint_url = ( + _resolve_azure_endpoint_for_provider(db, organization_id, llm_provider) + if provider_key == "azure" + else None + ) + from app.services.voice_agent.llm_voice_providers import resolve_voice_llm_base_url + + llm_base_url = resolve_voice_llm_base_url( + db, + organization_id, + voice_bundle, + llm_provider, + ) + return llm_endpoint_url, llm_base_url def _resolve_api_key_for_provider(db: Session, organization_id: UUID, provider: ModelProvider) -> Optional[str]: @@ -210,6 +264,8 @@ def resolve_vobiz_agent_context( stt_api_key = None tts_api_key = None llm_api_key = None + llm_endpoint_url = None + llm_base_url = None if use_voice_bundle_pipeline and voice_bundle: if voice_bundle.stt_provider: @@ -218,6 +274,11 @@ def resolve_vobiz_agent_context( tts_api_key = _resolve_api_key_for_provider(db, organization_id, voice_bundle.tts_provider) if voice_bundle.llm_provider: llm_api_key = _resolve_api_key_for_provider(db, organization_id, voice_bundle.llm_provider) + llm_endpoint_url, llm_base_url = _resolve_voice_llm_urls( + db, + organization_id, + voice_bundle, + ) else: ai_provider = None if agent.ai_provider_id: @@ -260,6 +321,8 @@ def resolve_vobiz_agent_context( stt_api_key=stt_api_key, tts_api_key=tts_api_key, llm_api_key=llm_api_key, + llm_endpoint_url=llm_endpoint_url, + llm_base_url=llm_base_url, ) @@ -277,13 +340,14 @@ def vobiz_webhook_base_url() -> str: return base.rstrip("/") -def build_vobiz_ws_url( +def build_carrier_ws_url( *, agent_id: str, session: str, persona_id: Optional[str] = None, scenario_id: Optional[str] = None, ) -> str: + """Build the WebSocket URL for live carrier audio (Plivo, Vobiz, etc.).""" from app.config import settings from app.services.media_urls import carrier_media_ws_base_url from urllib.parse import quote @@ -297,7 +361,12 @@ def build_vobiz_ws_url( query += f"&persona_id={quote(persona_id)}" if scenario_id: query += f"&scenario_id={quote(scenario_id)}" - return f"{ws_base}{settings.API_V1_PREFIX}/telephony/vobiz/ws?{query}" + return f"{ws_base}{settings.API_V1_PREFIX}/telephony/carrier/ws?{query}" + + +def build_vobiz_ws_url(**kwargs) -> str: + """Deprecated alias for :func:`build_carrier_ws_url`.""" + return build_carrier_ws_url(**kwargs) def extract_webhook_params(payload: Dict[str, Any]) -> Dict[str, Any]: diff --git a/app/services/telephony/webhook_auth.py b/app/services/telephony/webhook_auth.py index 16a7c1ad..cc6808a8 100644 --- a/app/services/telephony/webhook_auth.py +++ b/app/services/telephony/webhook_auth.py @@ -135,6 +135,38 @@ def _auth_token_for_vobiz_org(db: Session, org_id: UUID) -> Optional[str]: return _platform_vobiz_auth_token() +def _platform_plivo_auth_token() -> Optional[str]: + token = (settings.PLIVO_AUTH_TOKEN or "").strip() + return token or None + + +def _resolve_plivo_auth_token_by_auth_id(auth_id: str, db: Session) -> Optional[str]: + candidate = (auth_id or "").strip() + if not candidate: + return None + + platform_id = (settings.PLIVO_AUTH_ID or "").strip() + if platform_id and candidate.lower() == platform_id.lower(): + return _platform_plivo_auth_token() + + rows = ( + db.query(TelephonyIntegration) + .filter( + TelephonyIntegration.provider == "plivo", + TelephonyIntegration.is_active.is_(True), + ) + .all() + ) + for row in rows: + try: + stored_id = decrypt_api_key(row.auth_id).strip() + except Exception: + continue + if stored_id.lower() == candidate.lower(): + return decrypt_api_key(row.auth_token).strip() + return None + + def _resolve_auth_token_for_call_event( params: Dict[str, Any], db: Session, @@ -181,7 +213,27 @@ def resolve_plivo_auth_token( phone_number = params.get("To") or params.get("to") return _resolve_auth_token_for_phone(phone_number, db) if webhook_kind == "events": - return _resolve_auth_token_for_call_event(params, db) + token = _resolve_auth_token_for_call_event(params, db) + if token: + return token + + parent_auth_id = ( + params.get("ParentAuthID") + or params.get("AuthID") + or params.get("auth_id") + ) + if parent_auth_id: + token = _resolve_plivo_auth_token_by_auth_id(str(parent_auth_id), db) + if token: + return token + + phone_number = ( + params.get("To") + or params.get("to") + or params.get("From") + or params.get("from") + ) + return _resolve_auth_token_for_phone(phone_number, db) return None diff --git a/app/services/voice_agent/llm_voice_providers.py b/app/services/voice_agent/llm_voice_providers.py new file mode 100644 index 00000000..7cf8848b --- /dev/null +++ b/app/services/voice_agent/llm_voice_providers.py @@ -0,0 +1,339 @@ +"""Live voice pipeline LLM provider registry and service factory. + +Mirrors the LLM-capable providers exposed in Voice Bundles / Integrations +(see ``app/services/judge_alignment/model_catalog.py``). +""" + +from __future__ import annotations + +import json +import os +from typing import Any, Callable, Dict, Optional + +from loguru import logger + +# Providers selectable for the LLM leg of STT+LLM+TTS voice bundles. +LLM_VOICE_PROVIDER_KEYS = frozenset( + { + "openai", + "anthropic", + "google", + "xai", + "fireworks", + "cohere", + "mistral", + "meta", + "together", + "perplexity", + "azure", + "aws", + "openrouter", + "custom", + "sarvam", + } +) + +_DEFAULT_LLM_MODELS: Dict[str, str] = { + "openai": "gpt-4.1", + "google": "gemini-2.5-flash", + "anthropic": "claude-sonnet-4.6", + "xai": "grok-3-beta", + "fireworks": "accounts/fireworks/models/llama-v3p1-8b-instruct", + "cohere": "command-r-plus-08-2024", + "mistral": "mistral-small-latest", + "meta": "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", + "together": "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", + "perplexity": "sonar", + "azure": "gpt-4.1", + "aws": "amazon.nova-lite-v1:0", + "openrouter": "openai/gpt-4o-2024-11-20", + "custom": "gpt-4o-mini", + "sarvam": "sarvam-30b", +} + +_ENV_KEYS: Dict[str, str] = { + "openai": "OPENAI_API_KEY", + "google": "GOOGLE_API_KEY", + "anthropic": "ANTHROPIC_API_KEY", + "xai": "XAI_API_KEY", + "fireworks": "FIREWORKS_API_KEY", + "cohere": "COHERE_API_KEY", + "mistral": "MISTRAL_API_KEY", + "meta": "TOGETHER_API_KEY", + "together": "TOGETHER_API_KEY", + "perplexity": "PERPLEXITY_API_KEY", + "azure": "AZURE_OPENAI_API_KEY", + "aws": "AWS_ACCESS_KEY_ID", + "openrouter": "OPENROUTER_API_KEY", + "custom": "OPENAI_API_KEY", + "sarvam": "SARVAM_API_KEY", +} + + +def normalize_llm_model(provider: str, model: str) -> str: + """Normalize catalog model ids for provider-specific APIs.""" + provider_key = (provider or "").strip().lower() + if not model: + return model + if provider_key == "fireworks" and not model.startswith("accounts/"): + return f"accounts/fireworks/models/{model}" + if provider_key == "azure": + from app.services.ai.llm_service import _azure_deployment_name + + return _azure_deployment_name(model) + return model + + +def default_llm_model(provider: str) -> str: + return _DEFAULT_LLM_MODELS.get((provider or "").strip().lower(), "gpt-4.1") + + +def llm_env_key(provider: str) -> str: + return _ENV_KEYS.get((provider or "").strip().lower(), "OPENAI_API_KEY") + + +def _parse_aws_credentials(api_key: str) -> Dict[str, Any]: + """Parse AWS credential JSON or fall back to access key + env secret.""" + try: + parsed = json.loads(api_key) + if isinstance(parsed, dict): + access = ( + parsed.get("aws_access_key_id") + or parsed.get("access_key_id") + or parsed.get("aws_access_key") + ) + secret = ( + parsed.get("aws_secret_access_key") + or parsed.get("secret_access_key") + or parsed.get("aws_secret_key") + ) + return { + "aws_access_key": access, + "aws_secret_key": secret, + "aws_session_token": parsed.get("aws_session_token") + or parsed.get("session_token"), + "aws_region": parsed.get("aws_region") + or parsed.get("region") + or os.getenv("AWS_REGION", "us-east-1"), + } + except (json.JSONDecodeError, TypeError): + pass + return { + "aws_access_key": api_key, + "aws_secret_key": os.getenv("AWS_SECRET_ACCESS_KEY"), + "aws_region": os.getenv("AWS_REGION", "us-east-1"), + } + + +def get_llm_provider_registry(get_service: Callable[[str], Any]) -> Dict[str, Dict[str, Any]]: + """Build the LLM provider registry used by ``run_voice_bundle_fastapi``.""" + + def _openai_factory(api_key, model, params=None, base_url=None): + kwargs: Dict[str, Any] = {"api_key": api_key, "model": normalize_llm_model("openai", model)} + if params: + kwargs["params"] = params + if base_url: + kwargs["base_url"] = base_url + return get_service("OpenAILLMService")(**kwargs) + + registry: Dict[str, Dict[str, Any]] = {} + + for provider in sorted(LLM_VOICE_PROVIDER_KEYS): + env_key = llm_env_key(provider) + default_model = default_llm_model(provider) + + if provider == "openai": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, _f=_openai_factory: _f( + api_key, model, params + ), + } + elif provider == "google": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("GoogleLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "anthropic": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("AnthropicLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "fireworks": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("FireworksLLMService")( + api_key=api_key, + model=normalize_llm_model("fireworks", model), + **({"params": params} if params else {}), + ), + } + elif provider == "xai": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("GrokLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "mistral": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("MistralLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider in ("together", "meta"): + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("TogetherLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "perplexity": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("PerplexityLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "openrouter": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenRouterLLMService")( + api_key=api_key, + model=model, + **({"params": params} if params else {}), + ), + } + elif provider == "aws": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("AWSBedrockLLMService")( + model=model, + params=params, + **_parse_aws_credentials(api_key), + ), + } + elif provider == "cohere": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenAILLMService")( + api_key=api_key, + model=model, + base_url="https://api.cohere.com/compatibility/v1", + **({"params": params} if params else {}), + ), + } + elif provider == "sarvam": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, gs=get_service: gs("OpenAILLMService")( + api_key=api_key, + model=model, + base_url="https://api.sarvam.ai/v1", + **({"params": params} if params else {}), + ), + } + elif provider == "custom": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + "factory": lambda api_key, model, params=None, base_url=None, _f=_openai_factory: _f( + api_key, model, params, base_url=base_url + ), + "supports_base_url": True, + } + elif provider == "azure": + registry[provider] = { + "env_key": env_key, + "default_model": default_model, + # Azure is instantiated with endpoint metadata in run_voice_bundle_fastapi. + "factory": lambda api_key, model, params=None, _f=_openai_factory: _f( + api_key, normalize_llm_model("azure", model), params + ), + } + + return registry + + +def resolve_voice_llm_base_url(db, organization_id, voice_bundle, llm_provider) -> Optional[str]: + """Resolve an OpenAI-compatible base URL for custom / gateway-routed LLM legs.""" + provider_key = ( + llm_provider.value if hasattr(llm_provider, "value") else str(llm_provider) + ).lower() + if provider_key != "custom": + return None + + from app.services.credentials import resolve_ai_provider + + ai_provider = resolve_ai_provider( + provider_key, + db, + organization_id, + credential_id=getattr(voice_bundle, "llm_credential_id", None), + ) + if not ai_provider: + return None + + base_url = getattr(ai_provider, "gateway_base_url", None) + if base_url and str(base_url).strip(): + return str(base_url).strip() + return None + + +def instantiate_llm_service( + provider: str, + *, + get_service: Callable[[str], Any], + api_key: str, + model: str, + params: Optional[Any] = None, + base_url: Optional[str] = None, +): + """Instantiate a streaming LLM service for the live voice pipeline.""" + registry = get_llm_provider_registry(get_service) + provider_key = (provider or "").strip().lower() + cfg = registry.get(provider_key) + if cfg is None: + supported = ", ".join(sorted(registry.keys())) + raise ValueError( + f"Unsupported LLM provider '{provider_key}'. Supported providers: {supported}" + ) + + factory = cfg["factory"] + if provider_key == "custom" or cfg.get("supports_base_url"): + return factory(api_key, model, params, base_url=base_url) + if base_url: + logger.debug( + "Ignoring llm base_url for provider '{}' (not OpenAI-compatible custom routing)", + provider_key, + ) + return factory(api_key, model, params) diff --git a/app/services/voice_agent/voice_bundle.py b/app/services/voice_agent/voice_bundle.py index edebc4d0..89bc72fe 100644 --- a/app/services/voice_agent/voice_bundle.py +++ b/app/services/voice_agent/voice_bundle.py @@ -153,6 +153,30 @@ def _get_service(service_name: str): elif service_name == "AzureLLMService": from efficientai.services.azure.llm import AzureLLMService service_class = AzureLLMService + elif service_name == "AnthropicLLMService": + from efficientai.services.anthropic.llm import AnthropicLLMService + service_class = AnthropicLLMService + elif service_name == "FireworksLLMService": + from efficientai.services.fireworks.llm import FireworksLLMService + service_class = FireworksLLMService + elif service_name == "GrokLLMService": + from efficientai.services.grok.llm import GrokLLMService + service_class = GrokLLMService + elif service_name == "MistralLLMService": + from efficientai.services.mistral.llm import MistralLLMService + service_class = MistralLLMService + elif service_name == "TogetherLLMService": + from efficientai.services.together.llm import TogetherLLMService + service_class = TogetherLLMService + elif service_name == "PerplexityLLMService": + from efficientai.services.perplexity.llm import PerplexityLLMService + service_class = PerplexityLLMService + elif service_name == "OpenRouterLLMService": + from efficientai.services.openrouter.llm import OpenRouterLLMService + service_class = OpenRouterLLMService + elif service_name == "AWSBedrockLLMService": + from efficientai.services.aws.llm import AWSBedrockLLMService + service_class = AWSBedrockLLMService # Optional: Smart Turn Analyzer elif service_name == "LocalSmartTurnAnalyzerV3": @@ -347,40 +371,10 @@ def _instantiate_tts_service( def _get_llm_providers(): - """Get LLM provider registry with truly lazy-loaded service classes. - - Each provider's SDK is only loaded when that provider is actually used. - """ - return { - "openai": { - "env_key": "OPENAI_API_KEY", - "default_model": "gpt-4.1", - "factory": lambda api_key, model, params=None: _get_service("OpenAILLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - }, - "google": { - "env_key": "GOOGLE_API_KEY", - "default_model": "gemini-2.5-flash", - "factory": lambda api_key, model, params=None: _get_service("GoogleLLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - }, - "azure": { - "env_key": "AZURE_OPENAI_API_KEY", - "default_model": "gpt-4.1", - # Azure is instantiated with endpoint metadata in run_voice_bundle_fastapi. - "factory": lambda api_key, model, params=None: _get_service("OpenAILLMService")( - api_key=api_key, - model=model, - **({"params": params} if params else {}), - ), - }, - } + """Get LLM provider registry with truly lazy-loaded service classes.""" + from app.services.voice_agent.llm_voice_providers import get_llm_provider_registry + + return get_llm_provider_registry(_get_service) DEFAULT_STT_PROVIDER = None @@ -522,6 +516,7 @@ async def run_voice_bundle_fastapi( tts_api_key: str | None = None, llm_api_key: str | None = None, llm_endpoint_url: str | None = None, + llm_base_url: str | None = None, serializer=None, telephony_mode: bool = False, call_short_id: str | None = None, @@ -696,7 +691,16 @@ async def run_voice_bundle_fastapi( params=llm_params, ) else: - llm = llm_cfg["factory"](api_key=llm_api_key, model=llm_model, params=llm_params) + from app.services.voice_agent.llm_voice_providers import instantiate_llm_service + + llm = instantiate_llm_service( + llm_provider_value, + get_service=_get_service, + api_key=llm_api_key, + model=llm_model, + params=llm_params, + base_url=llm_base_url, + ) # Build context with provided system instruction or a default base_instruction = ( diff --git a/app/workers/concurrency/telephony_credential_rate_limit.py b/app/workers/concurrency/telephony_credential_rate_limit.py index c2c9c66e..e5617ffe 100644 --- a/app/workers/concurrency/telephony_credential_rate_limit.py +++ b/app/workers/concurrency/telephony_credential_rate_limit.py @@ -98,13 +98,16 @@ def fingerprint_for_integration(integration) -> str: ) +_CREDENTIALED_RECORDING_IMPORT_PROVIDERS = frozenset({"exotel", "plivo"}) + + def requires_authenticated_recording_fetch(call_import) -> bool: """True when call-import recording fetch uses provider HTTP auth.""" if call_import.telephony_integration_id is None and not ( call_import.provider or "" ).strip(): return False - return (call_import.provider or "").strip().lower() == "exotel" + return (call_import.provider or "").strip().lower() in _CREDENTIALED_RECORDING_IMPORT_PROVIDERS _PEEK_LUA = """ diff --git a/app/workers/tasks/process_call_import_row.py b/app/workers/tasks/process_call_import_row.py index d3299b3d..efc177d7 100644 --- a/app/workers/tasks/process_call_import_row.py +++ b/app/workers/tasks/process_call_import_row.py @@ -11,8 +11,11 @@ * **Exotel credentialed import**: download the CSV-supplied ``recording_url`` with HTTP Basic auth from the batch's pinned credentials. - * **Other credentialed providers** (e.g. Plivo): download the - CSV-supplied ``recording_url`` without auth (public links). + * **Plivo credentialed import**: download the CSV-supplied + ``recording_url`` with HTTP Basic auth from the batch's pinned + credentials. + * **Other credentialed providers**: download the CSV-supplied + ``recording_url`` without auth (public links). Transient download errors schedule a Celery retry; auth/4xx/oversize errors mark the row failed immediately. @@ -220,11 +223,14 @@ def _row_or_import_gone( return None +_CREDENTIALED_RECORDING_IMPORT_PROVIDERS = frozenset({"exotel", "plivo"}) + + def _use_credentialed_recording_download(call_import, client) -> bool: """True when CSV recording URLs should be fetched with provider auth.""" if client is None or not hasattr(client, "download_recording"): return False - return (call_import.provider or "").lower() == "exotel" + return (call_import.provider or "").lower() in _CREDENTIALED_RECORDING_IMPORT_PROVIDERS def _is_direct_url_import(call_import) -> bool: @@ -498,8 +504,8 @@ def process_call_import_row_task( else: # ------------------------------------------------------------------ # Credentialed import — download from the CSV-supplied URL only. - # Exotel URLs require HTTP Basic auth; other providers use public - # download (e.g. Plivo presigned links). + # Exotel and Plivo URLs require HTTP Basic auth; other providers + # use public download (e.g. presigned links). # ------------------------------------------------------------------ audio_bytes: Optional[bytes] = None content_type: Optional[str] = None diff --git a/docs/telephony-media.md b/docs/telephony-media.md index b8bc2ce9..f0347b6b 100644 --- a/docs/telephony-media.md +++ b/docs/telephony-media.md @@ -15,7 +15,7 @@ Vobiz sees **one public host** (the telephony edge): - `POST/GET /api/v1/telephony/vobiz/webhooks/answer` - `POST /api/v1/telephony/vobiz/webhooks/events` - `POST /api/v1/telephony/vobiz/webhooks/recording-ready` -- `WSS /api/v1/telephony/vobiz/ws` +- `WSS /api/v1/telephony/carrier/ws` (shared live-audio socket for Plivo, Vobiz, and other Stream-compatible carriers) Configure **`vobiz.webhook_base_url`** to that public URL (e.g. `https://telephony.staging.example.com`). When `media_ws_base_url` / `MEDIA_WS_BASE_URL` is unset, carrier answer XML reuses the same host (`https` → `wss`) via `carrier_media_ws_base_url()`. diff --git a/frontend/src/pages/agents/AgentWorkspaceDetail.tsx b/frontend/src/pages/agents/AgentWorkspaceDetail.tsx index feba5879..a0f379b0 100644 --- a/frontend/src/pages/agents/AgentWorkspaceDetail.tsx +++ b/frontend/src/pages/agents/AgentWorkspaceDetail.tsx @@ -28,6 +28,24 @@ function normalizeCallMedium(value: string | undefined | null): 'phone_call' | ' return value === 'web_call' ? 'web_call' : 'phone_call' } +function agentToFormData(agent: NonNullable>>): FormData { + return { + name: agent.name, + phone_number: agent.phone_number || '', + language: agent.language, + description: agent.description || '', + prompt_variables: agent.prompt_variables || {}, + silence_hangup_secs: agent.silence_hangup_secs ?? 15, + call_type: agent.call_type, + call_medium: normalizeCallMedium(agent.call_medium), + telephony_phone_number_id: agent.telephony_phone_number_id || '', + voice_bundle_id: agent.voice_bundle_id || '', + voice_ai_integration_id: agent.voice_ai_integration_id || '', + voice_ai_agent_id: agent.voice_ai_agent_id || '', + provider_prompt: agent.provider_prompt || '', + } +} + interface FormData { name: string phone_number: string @@ -134,24 +152,10 @@ export default function AgentWorkspaceDetail({ } useEffect(() => { - if (agent) { - setFormData({ - name: agent.name, - phone_number: agent.phone_number || '', - language: agent.language, - description: agent.description || '', - prompt_variables: agent.prompt_variables || {}, - silence_hangup_secs: agent.silence_hangup_secs ?? 15, - call_type: agent.call_type, - call_medium: normalizeCallMedium(agent.call_medium), - telephony_phone_number_id: agent.telephony_phone_number_id || '', - voice_bundle_id: agent.voice_bundle_id || '', - voice_ai_integration_id: agent.voice_ai_integration_id || '', - voice_ai_agent_id: agent.voice_ai_agent_id || '', - provider_prompt: agent.provider_prompt || '', - }) + if (agent && !isEditMode) { + setFormData(agentToFormData(agent)) } - }, [agent]) + }, [agent, isEditMode]) const updateMutation = useMutation({ mutationFn: (data: FormData) => { @@ -183,6 +187,7 @@ export default function AgentWorkspaceDetail({ onSuccess: () => { queryClient.invalidateQueries({ queryKey: ['agent', agentRouteId] }) queryClient.invalidateQueries({ queryKey: ['agents'] }) + queryClient.invalidateQueries({ queryKey: ['telephony-numbers'] }) setIsEditMode(false) showToast('Agent updated successfully!', 'success') }, @@ -272,23 +277,16 @@ export default function AgentWorkspaceDetail({ updateMutation.mutate(formData) } + const handleEditClick = () => { + if (agent) { + setFormData(agentToFormData(agent)) + } + setIsEditMode(true) + } + const handleCancelEdit = () => { if (agent) { - setFormData({ - name: agent.name, - phone_number: agent.phone_number || '', - language: agent.language, - description: agent.description || '', - prompt_variables: agent.prompt_variables || {}, - silence_hangup_secs: agent.silence_hangup_secs ?? 15, - call_type: agent.call_type, - call_medium: normalizeCallMedium(agent.call_medium), - telephony_phone_number_id: agent.telephony_phone_number_id || '', - voice_bundle_id: agent.voice_bundle_id || '', - voice_ai_integration_id: agent.voice_ai_integration_id || '', - voice_ai_agent_id: agent.voice_ai_agent_id || '', - provider_prompt: agent.provider_prompt || '', - }) + setFormData(agentToFormData(agent)) } setIsEditMode(false) } @@ -392,7 +390,7 @@ export default function AgentWorkspaceDetail({ agentId={agent.agent_id} isEditMode={isEditMode} isPending={updateMutation.isPending} - onEditClick={() => setIsEditMode(true)} + onEditClick={handleEditClick} onCancelEdit={handleCancelEdit} onSave={handleSave} /> diff --git a/frontend/src/pages/agents/AgentsWorkspace.tsx b/frontend/src/pages/agents/AgentsWorkspace.tsx index d242762e..48195acf 100644 --- a/frontend/src/pages/agents/AgentsWorkspace.tsx +++ b/frontend/src/pages/agents/AgentsWorkspace.tsx @@ -116,6 +116,7 @@ export default function AgentsWorkspace() { const handleCreateSuccess = () => { queryClient.invalidateQueries({ queryKey: ['agents'] }) + queryClient.invalidateQueries({ queryKey: ['telephony-numbers'] }) setShowCreateModal(false) } diff --git a/frontend/src/pages/agents/components/AgentEditForm.tsx b/frontend/src/pages/agents/components/AgentEditForm.tsx index d734ff8b..7a9b6e91 100644 --- a/frontend/src/pages/agents/components/AgentEditForm.tsx +++ b/frontend/src/pages/agents/components/AgentEditForm.tsx @@ -159,20 +159,16 @@ export default function AgentEditForm({ } if (!canUseProviderNumbers || telephonyNumbers.length === 0) { - setPhoneNumberInputMode('custom') - return - } - - const selectedExists = telephonyNumbers.some((n) => n.id === formData.telephony_phone_number_id) - if (!selectedExists && phoneNumberInputMode === 'provider') { - onChange({ ...formData, telephony_phone_number_id: '', phone_number: '' }) + if (phoneNumberInputMode === 'provider') { + setPhoneNumberInputMode('custom') + } } }, [ - formData, - onChange, + formData.call_medium, + formData.telephony_phone_number_id, phoneNumberInputMode, canUseProviderNumbers, - telephonyNumbers, + telephonyNumbers.length, ]) const generateDescriptionMutation = useMutation({ @@ -383,23 +379,27 @@ export default function AgentEditForm({ disabled={!canUseProviderNumbers || telephonyNumbers.length === 0} > - {telephonyNumbers.map((number) => ( - - ))} + {telephonyNumbers.map((number) => { + const assignedToOtherAgent = + !!number.agent_id && number.agent_id !== agentId + return ( + + ) + })} ) : ( Date: Tue, 1 Sep 2026 17:48:15 +0000 Subject: [PATCH 2/5] feat: updating plivo creds --- app/api/v1/routes/vobiz_telephony.py | 13 +-- .../telephony/carrier_media_serializer.py | 58 ++++++++++++ app/services/telephony/recording_download.py | 34 ++++--- pytest_out1.txt | 29 ------ tests/conftest.py | 1 + .../test_telephony/test_recording_download.py | 24 +++++ .../test_telephony/test_vobiz.py | 91 +++++++++++++++++++ .../test_llm_voice_providers.py | 2 +- 8 files changed, 201 insertions(+), 51 deletions(-) create mode 100644 app/services/telephony/carrier_media_serializer.py delete mode 100644 pytest_out1.txt diff --git a/app/api/v1/routes/vobiz_telephony.py b/app/api/v1/routes/vobiz_telephony.py index 08355d81..a750152a 100644 --- a/app/api/v1/routes/vobiz_telephony.py +++ b/app/api/v1/routes/vobiz_telephony.py @@ -49,8 +49,8 @@ from app.services.telephony.vobiz_xml import reject_call, speak_and_hangup, stream_to_agent from app.services.voice_agent.bot_fast_api import run_bot from app.services.voice_agent.voice_bundle import run_voice_bundle_fastapi +from app.services.telephony.carrier_media_serializer import build_carrier_frame_serializer from efficientai.runner.utils import parse_telephony_websocket -from efficientai.serializers.vobiz import VobizFrameSerializer # Exposed at module scope so tests can patch `.delay` without importing Celery tasks. initiate_vobiz_outbound_call_task = None @@ -680,15 +680,12 @@ async def carrier_media_websocket(websocket: WebSocket): persona_id=persona_id, scenario_id=scenario_id, ) - serializer = VobizFrameSerializer( + serializer = build_carrier_frame_serializer( + provider_platform=getattr(call_row, "provider_platform", None), stream_id=stream_id, call_id=call_id, - auth_id=settings.VOBIZ_AUTH_ID, - auth_token=settings.VOBIZ_AUTH_TOKEN, - params=VobizFrameSerializer.InputParams( - sample_rate=8000, - api_base=settings.VOBIZ_API_BASE, - ), + organization_id=UUID(session.organization_id), + db=db, ) if context.use_voice_bundle_pipeline: diff --git a/app/services/telephony/carrier_media_serializer.py b/app/services/telephony/carrier_media_serializer.py new file mode 100644 index 00000000..039946ff --- /dev/null +++ b/app/services/telephony/carrier_media_serializer.py @@ -0,0 +1,58 @@ +"""Choose the carrier media serializer and hangup credentials.""" + +from __future__ import annotations + +from typing import Optional +from uuid import UUID + +from sqlalchemy.orm import Session + +from app.config import settings +from app.core.encryption import decrypt_api_key +from app.services.credentials.resolver import resolve_telephony_integration +from efficientai.serializers.plivo import PlivoFrameSerializer +from efficientai.serializers.vobiz import VobizFrameSerializer + + +def _plivo_call_control_credentials( + db: Session, organization_id: UUID +) -> tuple[str, str]: + integration = resolve_telephony_integration("plivo", db, organization_id) + if integration: + return ( + decrypt_api_key(integration.auth_id).strip(), + decrypt_api_key(integration.auth_token).strip(), + ) + return (settings.PLIVO_AUTH_ID or "").strip(), (settings.PLIVO_AUTH_TOKEN or "").strip() + + +def build_carrier_frame_serializer( + *, + provider_platform: Optional[str], + stream_id: str, + call_id: Optional[str], + organization_id: UUID, + db: Session, +): + """Return the media serializer whose hangup API matches the live carrier.""" + platform = (provider_platform or "").strip().lower() + if platform == "plivo": + auth_id, auth_token = _plivo_call_control_credentials(db, organization_id) + return PlivoFrameSerializer( + stream_id=stream_id, + call_id=call_id, + auth_id=auth_id, + auth_token=auth_token, + params=PlivoFrameSerializer.InputParams(sample_rate=8000), + ) + + return VobizFrameSerializer( + stream_id=stream_id, + call_id=call_id, + auth_id=settings.VOBIZ_AUTH_ID, + auth_token=settings.VOBIZ_AUTH_TOKEN, + params=VobizFrameSerializer.InputParams( + sample_rate=8000, + api_base=settings.VOBIZ_API_BASE, + ), + ) diff --git a/app/services/telephony/recording_download.py b/app/services/telephony/recording_download.py index e4d40ec8..2340fc3f 100644 --- a/app/services/telephony/recording_download.py +++ b/app/services/telephony/recording_download.py @@ -107,7 +107,12 @@ def assert_recording_url_safe( user_supplied: bool, allowed_suffixes: Optional[List[str]] = None, ) -> None: - """Validate a recording URL before any outbound HTTP request.""" + """Validate a recording URL before any outbound HTTP request. + + Literal IP hosts are always rejected so credentialed fetches cannot send + Basic auth to a CSV-controlled address. ``user_supplied`` is kept for + call-site compatibility; redirect policy is enforced by the downloader. + """ parsed = urlparse(recording_url.strip()) if parsed.scheme not in {"http", "https"}: raise ExotelInvalidContentError( @@ -125,24 +130,27 @@ def assert_recording_url_safe( try: literal_ip = ipaddress.ip_address(hostname) + except ValueError: + literal_ip = None + + if literal_ip is not None: if _ip_is_blocked(literal_ip): raise ExotelInvalidContentError( "Recording URL targets a blocked network address" ) - if user_supplied: - raise ExotelInvalidContentError( - "User-supplied recording URLs must use allowlisted hostnames" - ) - except ValueError: - if not _hostname_allowed(hostname, suffixes): + raise ExotelInvalidContentError( + "Recording URLs must use allowlisted hostnames, not IP addresses" + ) + + if not _hostname_allowed(hostname, suffixes): + raise ExotelInvalidContentError( + f"Recording URL hostname is not allowlisted: {hostname}" + ) + for resolved_ip in _resolve_host_ips(hostname): + if _ip_is_blocked(resolved_ip): raise ExotelInvalidContentError( - f"Recording URL hostname is not allowlisted: {hostname}" + "Recording URL resolves to a blocked network address" ) - for resolved_ip in _resolve_host_ips(hostname): - if _ip_is_blocked(resolved_ip): - raise ExotelInvalidContentError( - "Recording URL resolves to a blocked network address" - ) def download_recording_url( diff --git a/pytest_out1.txt b/pytest_out1.txt deleted file mode 100644 index 69ea3785..00000000 --- a/pytest_out1.txt +++ /dev/null @@ -1,29 +0,0 @@ -.................... [100%] -=============================== warnings summary =============================== -app/models/schemas.py:2048 - /home/tejasnarayan/EfficientAI/efficientAI/app/models/schemas.py:2048: PydanticDeprecatedSince20: Pydantic V1 style `@validator` validators are deprecated. You should migrate to Pydantic V2 style `@field_validator` validators, see the migration guide for more details. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/ - @validator('supported_surfaces', 'enabled_surfaces', pre=True) - -app/models/schemas.py:2058 - /home/tejasnarayan/EfficientAI/efficientAI/app/models/schemas.py:2058: PydanticDeprecatedSince20: Pydantic V1 style `@validator` validators are deprecated. You should migrate to Pydantic V2 style `@field_validator` validators, see the migration guide for more details. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/ - @validator('metric_origin', pre=True) - -app/models/schemas.py:2836 - /home/tejasnarayan/EfficientAI/efficientAI/app/models/schemas.py:2836: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/ - class TelephonyIntegrationResponse(BaseModel): - -app/models/schemas.py:2857 - /home/tejasnarayan/EfficientAI/efficientAI/app/models/schemas.py:2857: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/ - class TelephonyPhoneNumberResponse(BaseModel): - -app/models/schemas.py:2946 - /home/tejasnarayan/EfficientAI/efficientAI/app/models/schemas.py:2946: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/ - class TelephonyMaskingSessionResponse(BaseModel): - -newenv/lib/python3.12/site-packages/requests/__init__.py:113 - /home/tejasnarayan/EfficientAI/efficientAI/newenv/lib/python3.12/site-packages/requests/__init__.py:113: RequestsDependencyWarning: urllib3 (2.5.0) or chardet (6.0.0.post1)/charset_normalizer (3.4.4) doesn't match a supported version! - warnings.warn( - --- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html -20 passed, 6 warnings in 3.98s -PYTEST1_EXIT:0 diff --git a/tests/conftest.py b/tests/conftest.py index f8ddd33b..6fa81e75 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -908,6 +908,7 @@ def _build_session_api_app(): app.include_router(voice_agent.router, prefix="/api/v1") app.include_router(voice_playground.router, prefix="/api/v1") app.include_router(telephony.router, prefix="/api/v1") + app.include_router(telephony.plivo_webhook_router, prefix="/api/v1") app.include_router(vobiz_telephony.router, prefix="/api/v1") app.include_router(call_imports.router, prefix="/api/v1") app.include_router(call_import_schemas.router, prefix="/api/v1") diff --git a/tests/test_services/test_telephony/test_recording_download.py b/tests/test_services/test_telephony/test_recording_download.py index 30218415..07068c17 100644 --- a/tests/test_services/test_telephony/test_recording_download.py +++ b/tests/test_services/test_telephony/test_recording_download.py @@ -22,6 +22,30 @@ def test_assert_recording_url_safe_rejects_metadata_ip(): ) +def test_assert_recording_url_safe_rejects_public_literal_ip_even_when_trusted(): + with pytest.raises(ExotelInvalidContentError, match="not IP addresses"): + module.assert_recording_url_safe( + "https://203.0.113.10/recordings/call.mp3", + user_supplied=False, + ) + + +def test_download_recording_url_does_not_send_credentials_to_literal_ip(monkeypatch): + mock_client = MagicMock() + mock_client.__enter__.return_value = mock_client + mock_client.__exit__.return_value = False + + with patch.object(module.httpx, "Client", return_value=mock_client): + with pytest.raises(ExotelInvalidContentError, match="not IP addresses"): + module.download_recording_url( + "https://203.0.113.10/recordings/call.mp3", + auth=("plivo-auth-id", "plivo-auth-token"), + credential_fingerprint="fp-plivo", + ) + + mock_client.get.assert_not_called() + + def test_assert_recording_url_safe_rejects_non_allowlisted_host(monkeypatch): monkeypatch.setattr( module.settings, diff --git a/tests/test_services/test_telephony/test_vobiz.py b/tests/test_services/test_telephony/test_vobiz.py index 561857b9..1e956546 100644 --- a/tests/test_services/test_telephony/test_vobiz.py +++ b/tests/test_services/test_telephony/test_vobiz.py @@ -11,7 +11,9 @@ from app.services.telephony.plivo_client import expand_phone_candidates, normalize_e164 from app.services.telephony.vobiz_client import VobizClient, build_vobiz_client_for_org from app.services.telephony.vobiz_session import create_call_session, delete_call_session, get_call_session +from app.services.telephony.carrier_media_serializer import build_carrier_frame_serializer from app.services.telephony.vobiz_xml import reject_call, speak_and_hangup, stream_to_agent +from efficientai.serializers.plivo import PlivoFrameSerializer from efficientai.serializers.vobiz import VobizFrameSerializer @@ -232,3 +234,92 @@ async def test_vobiz_serializer_round_trip_media_frame(): frame = await serializer.deserialize(media_message) assert frame is not None assert len(frame.audio) > 0 + + +def test_carrier_frame_serializer_uses_plivo_credentials_for_plivo_calls(monkeypatch): + org_id = uuid4() + integration = MagicMock() + integration.auth_id = "enc-auth-id" + integration.auth_token = "enc-auth-token" + + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.resolve_telephony_integration", + lambda provider, db, organization_id, **_kwargs: integration, + ) + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.decrypt_api_key", + lambda value: {"enc-auth-id": "plivo-auth-id", "enc-auth-token": "plivo-token"}[value], + ) + + serializer = build_carrier_frame_serializer( + provider_platform="plivo", + stream_id="stream-plivo", + call_id="call-plivo", + organization_id=org_id, + db=MagicMock(), + ) + + assert isinstance(serializer, PlivoFrameSerializer) + assert serializer._auth_id == "plivo-auth-id" + assert serializer._auth_token == "plivo-token" + assert serializer._call_id == "call-plivo" + + +def test_carrier_frame_serializer_falls_back_to_platform_plivo_credentials(monkeypatch): + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.resolve_telephony_integration", + lambda *_args, **_kwargs: None, + ) + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.settings.PLIVO_AUTH_ID", + "platform-plivo-id", + raising=False, + ) + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.settings.PLIVO_AUTH_TOKEN", + "platform-plivo-token", + raising=False, + ) + + serializer = build_carrier_frame_serializer( + provider_platform="plivo", + stream_id="stream-plivo", + call_id="call-plivo", + organization_id=uuid4(), + db=MagicMock(), + ) + + assert isinstance(serializer, PlivoFrameSerializer) + assert serializer._auth_id == "platform-plivo-id" + assert serializer._auth_token == "platform-plivo-token" + + +def test_carrier_frame_serializer_keeps_vobiz_credentials_for_vobiz_calls(monkeypatch): + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.settings.VOBIZ_AUTH_ID", + "vobiz-auth-id", + raising=False, + ) + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.settings.VOBIZ_AUTH_TOKEN", + "vobiz-token", + raising=False, + ) + monkeypatch.setattr( + "app.services.telephony.carrier_media_serializer.settings.VOBIZ_API_BASE", + "https://api.vobiz.ai", + raising=False, + ) + + serializer = build_carrier_frame_serializer( + provider_platform="vobiz", + stream_id="stream-vobiz", + call_id="call-vobiz", + organization_id=uuid4(), + db=MagicMock(), + ) + + assert isinstance(serializer, VobizFrameSerializer) + assert serializer._auth_id == "vobiz-auth-id" + assert serializer._auth_token == "vobiz-token" + assert serializer._call_id == "call-vobiz" diff --git a/tests/test_services/test_voice_agent/test_llm_voice_providers.py b/tests/test_services/test_voice_agent/test_llm_voice_providers.py index 69d7a345..999b95ab 100644 --- a/tests/test_services/test_voice_agent/test_llm_voice_providers.py +++ b/tests/test_services/test_voice_agent/test_llm_voice_providers.py @@ -44,4 +44,4 @@ def _factory(**kwargs): assert captured["service"] == "FireworksLLMService" assert captured["kwargs"]["model"] == "accounts/fireworks/models/llama-v3p1-8b-instruct" - assert captured["kwargs"]["api_key"] == "test-key" \ No newline at end of file + assert captured["kwargs"]["api_key"] == "test-key" From 8fcafc73d062657d39f3563f8d4c18d1a20741bd Mon Sep 17 00:00:00 2001 From: Tejas Narayan Date: Tue, 1 Sep 2026 18:03:59 +0000 Subject: [PATCH 3/5] fix: security path --- app/api/v1/routes/vobiz_telephony.py | 10 ++++-- .../telephony/call_recording_lifecycle.py | 3 ++ .../telephony/carrier_media_serializer.py | 32 +++++++++++++++++-- .../telephony/inbound_stream_answer.py | 5 ++- app/services/telephony/phone_routing.py | 12 +++---- app/services/telephony/recording_download.py | 24 ++++++++++++-- app/services/telephony/telephony_service.py | 4 ++- tests/test_api/test_vobiz_telephony.py | 4 ++- .../test_telephony/test_recording_download.py | 16 ++++++++++ .../test_telephony/test_vobiz.py | 20 ++++++++++-- 10 files changed, 111 insertions(+), 19 deletions(-) diff --git a/app/api/v1/routes/vobiz_telephony.py b/app/api/v1/routes/vobiz_telephony.py index a750152a..aff7b141 100644 --- a/app/api/v1/routes/vobiz_telephony.py +++ b/app/api/v1/routes/vobiz_telephony.py @@ -49,7 +49,10 @@ from app.services.telephony.vobiz_xml import reject_call, speak_and_hangup, stream_to_agent from app.services.voice_agent.bot_fast_api import run_bot from app.services.voice_agent.voice_bundle import run_voice_bundle_fastapi -from app.services.telephony.carrier_media_serializer import build_carrier_frame_serializer +from app.services.telephony.carrier_media_serializer import ( + build_carrier_frame_serializer, + telephony_integration_id_from_call_row, +) from efficientai.runner.utils import parse_telephony_websocket # Exposed at module scope so tests can patch `.delay` without importing Celery tasks. @@ -165,7 +168,9 @@ def _resolve_agent_for_answer( if session and session.agent_id and session.organization_id: return UUID(session.agent_id), UUID(session.organization_id), call_ref - agent_id, organization_id = resolve_inbound_agent_for_number(db, params.get("to")) + agent_id, organization_id, _telephony_integration_id = resolve_inbound_agent_for_number( + db, params.get("to") + ) if not agent_id or not organization_id: return None, None, None return agent_id, organization_id, None @@ -686,6 +691,7 @@ async def carrier_media_websocket(websocket: WebSocket): call_id=call_id, organization_id=UUID(session.organization_id), db=db, + telephony_integration_id=telephony_integration_id_from_call_row(call_row), ) if context.use_voice_bundle_pipeline: diff --git a/app/services/telephony/call_recording_lifecycle.py b/app/services/telephony/call_recording_lifecycle.py index a7c417a0..1abf4516 100644 --- a/app/services/telephony/call_recording_lifecycle.py +++ b/app/services/telephony/call_recording_lifecycle.py @@ -237,6 +237,7 @@ def create_inbound_call_recording( evaluator_id: Optional[UUID] = None, evaluator_result_id: Optional[UUID] = None, provider_platform: str = "vobiz", + telephony_integration_id: Optional[UUID] = None, ) -> CallRecording: existing = find_call_recording(db, call_ref=call_ref, provider_call_id=provider_call_id) if existing: @@ -254,6 +255,8 @@ def create_inbound_call_recording( } if evaluator_id is not None: call_data["evaluator_id"] = str(evaluator_id) + if telephony_integration_id is not None: + call_data["telephony_integration_id"] = str(telephony_integration_id) row = CallRecording( organization_id=organization_id, workspace_id=agent.workspace_id, diff --git a/app/services/telephony/carrier_media_serializer.py b/app/services/telephony/carrier_media_serializer.py index 039946ff..3585f36f 100644 --- a/app/services/telephony/carrier_media_serializer.py +++ b/app/services/telephony/carrier_media_serializer.py @@ -15,9 +15,17 @@ def _plivo_call_control_credentials( - db: Session, organization_id: UUID + db: Session, + organization_id: UUID, + *, + telephony_integration_id: Optional[UUID] = None, ) -> tuple[str, str]: - integration = resolve_telephony_integration("plivo", db, organization_id) + integration = resolve_telephony_integration( + "plivo", + db, + organization_id, + credential_id=telephony_integration_id, + ) if integration: return ( decrypt_api_key(integration.auth_id).strip(), @@ -26,6 +34,19 @@ def _plivo_call_control_credentials( return (settings.PLIVO_AUTH_ID or "").strip(), (settings.PLIVO_AUTH_TOKEN or "").strip() +def telephony_integration_id_from_call_row(call_row) -> Optional[UUID]: + if call_row is None: + return None + data = call_row.call_data if isinstance(call_row.call_data, dict) else {} + raw = data.get("telephony_integration_id") + if not raw: + return None + try: + return UUID(str(raw)) + except (TypeError, ValueError): + return None + + def build_carrier_frame_serializer( *, provider_platform: Optional[str], @@ -33,11 +54,16 @@ def build_carrier_frame_serializer( call_id: Optional[str], organization_id: UUID, db: Session, + telephony_integration_id: Optional[UUID] = None, ): """Return the media serializer whose hangup API matches the live carrier.""" platform = (provider_platform or "").strip().lower() if platform == "plivo": - auth_id, auth_token = _plivo_call_control_credentials(db, organization_id) + auth_id, auth_token = _plivo_call_control_credentials( + db, + organization_id, + telephony_integration_id=telephony_integration_id, + ) return PlivoFrameSerializer( stream_id=stream_id, call_id=call_id, diff --git a/app/services/telephony/inbound_stream_answer.py b/app/services/telephony/inbound_stream_answer.py index d5a19a8f..8f674b51 100644 --- a/app/services/telephony/inbound_stream_answer.py +++ b/app/services/telephony/inbound_stream_answer.py @@ -40,7 +40,9 @@ def build_inbound_stream_answer_xml( if not to_number: return speak_and_hangup("Call could not be routed.") - agent_id, organization_id = resolve_inbound_agent_for_number(db, to_number) + agent_id, organization_id, telephony_integration_id = resolve_inbound_agent_for_number( + db, to_number + ) if not agent_id or not organization_id: logger.warning( "Inbound stream answer miss: to={} from={} call_uuid={}", @@ -103,6 +105,7 @@ def build_inbound_stream_answer_xml( evaluator_id=inbound_evaluator_id, evaluator_result_id=inbound_evaluator_result_id, provider_platform=provider_platform, + telephony_integration_id=telephony_integration_id, ) if call_uuid: diff --git a/app/services/telephony/phone_routing.py b/app/services/telephony/phone_routing.py index b84446ca..8b011899 100644 --- a/app/services/telephony/phone_routing.py +++ b/app/services/telephony/phone_routing.py @@ -195,11 +195,11 @@ def _resolve_agent_for_org( def resolve_inbound_agent_for_number( db: Session, to_number_raw: Optional[str], -) -> Tuple[Optional[UUID], Optional[UUID]]: - """Resolve (agent_id, organization_id) for an inbound called number.""" +) -> Tuple[Optional[UUID], Optional[UUID], Optional[UUID]]: + """Resolve (agent_id, organization_id, telephony_integration_id) for inbound To.""" candidates = expand_phone_candidates(to_number_raw) if not candidates: - return None, None + return None, None, None number_row = _find_inbound_number_row(db, candidates) if not number_row: @@ -208,7 +208,7 @@ def resolve_inbound_agent_for_number( to_number_raw, candidates, ) - return None, None + return None, None, None agent_id, organization_id = _resolve_agent_for_org( db, @@ -217,7 +217,7 @@ def resolve_inbound_agent_for_number( number_row, ) if agent_id and organization_id: - return agent_id, organization_id + return agent_id, organization_id, number_row.telephony_integration_id logger.warning( "Inbound number {} owned by org {} but no agent linked (candidates={})", @@ -225,7 +225,7 @@ def resolve_inbound_agent_for_number( number_row.organization_id, candidates, ) - return None, None + return None, None, None def sync_agent_telephony_number_link(db: Session, agent: Agent) -> None: diff --git a/app/services/telephony/recording_download.py b/app/services/telephony/recording_download.py index 2340fc3f..be4e2107 100644 --- a/app/services/telephony/recording_download.py +++ b/app/services/telephony/recording_download.py @@ -29,6 +29,13 @@ "cloudfront.net", ) +# Credentialed telephony fetches must not send Basic auth to shared storage hosts. +_CREDENTIALED_ALLOWED_HOST_SUFFIXES = ( + "exotel.com", + "plivo.com", + "vobiz.ai", +) + _BLOCKED_NETWORKS = ( ipaddress.ip_network("0.0.0.0/8"), ipaddress.ip_network("10.0.0.0/8"), @@ -174,13 +181,26 @@ def download_recording_url( "User-supplied recording URLs must not be fetched with credentials" ) - assert_recording_url_safe(recording_url, user_supplied=user_supplied) + if auth is not None: + allowed_suffixes = list(_CREDENTIALED_ALLOWED_HOST_SUFFIXES) + else: + allowed_suffixes = _allowed_host_suffixes() + + assert_recording_url_safe( + recording_url, + user_supplied=user_supplied, + allowed_suffixes=allowed_suffixes, + ) request_hooks: Optional[dict[str, List[Callable[..., None]]]] = None if not user_supplied: def _validate_redirect(request: httpx.Request) -> None: - assert_recording_url_safe(str(request.url), user_supplied=False) + assert_recording_url_safe( + str(request.url), + user_supplied=False, + allowed_suffixes=allowed_suffixes, + ) request_hooks = {"request": [_validate_redirect]} diff --git a/app/services/telephony/telephony_service.py b/app/services/telephony/telephony_service.py index a8037a98..ef358c58 100644 --- a/app/services/telephony/telephony_service.py +++ b/app/services/telephony/telephony_service.py @@ -535,6 +535,8 @@ def initiate_outbound_call( raise ValueError("No default workspace found for organization") workspace_id = default_workspace.id + call_data = dict(response or {}) + call_data["telephony_integration_id"] = str(integration.id) db.add( CallRecording( organization_id=org_id, @@ -543,7 +545,7 @@ def initiate_outbound_call( status=CallRecordingStatus.PENDING, source=CallRecordingSource.WEBHOOK, call_event="outbound_initiated", - call_data=response, + call_data=call_data, provider_call_id=call_uuid, provider_platform=integration.provider, agent_id=agent_id, diff --git a/tests/test_api/test_vobiz_telephony.py b/tests/test_api/test_vobiz_telephony.py index 7376a560..9d975ca1 100644 --- a/tests/test_api/test_vobiz_telephony.py +++ b/tests/test_api/test_vobiz_telephony.py @@ -268,7 +268,9 @@ def test_vobiz_inbound_routing_is_org_scoped(db_session, org_id, seed_org, make_ agent = make_agent() _seed_vobiz_phone(db_session, org_id, phone_number="+919876543210", agent_id=agent.id) - resolved_agent_id, resolved_org_id = resolve_inbound_agent_for_number(db_session, "+919876543210") + resolved_agent_id, resolved_org_id, _resolved_integration_id = resolve_inbound_agent_for_number( + db_session, "+919876543210" + ) assert resolved_agent_id == agent.id assert resolved_org_id == org_id diff --git a/tests/test_services/test_telephony/test_recording_download.py b/tests/test_services/test_telephony/test_recording_download.py index 07068c17..ef9edd4e 100644 --- a/tests/test_services/test_telephony/test_recording_download.py +++ b/tests/test_services/test_telephony/test_recording_download.py @@ -69,6 +69,22 @@ def test_download_recording_url_rejects_credentials_for_user_supplied_urls(): ) +def test_download_recording_url_credentialed_rejects_shared_storage_host(): + mock_client = MagicMock() + mock_client.__enter__.return_value = mock_client + mock_client.__exit__.return_value = False + + with patch.object(module.httpx, "Client", return_value=mock_client): + with pytest.raises(ExotelInvalidContentError, match="not allowlisted"): + module.download_recording_url( + "https://evil-bucket.s3.amazonaws.com/recording.mp3", + auth=("plivo-auth-id", "plivo-auth-token"), + credential_fingerprint="fp-plivo", + ) + + mock_client.get.assert_not_called() + + def test_download_public_recording_fetches_allowlisted_host(monkeypatch): monkeypatch.setattr( module.settings, diff --git a/tests/test_services/test_telephony/test_vobiz.py b/tests/test_services/test_telephony/test_vobiz.py index 1e956546..b164cdd9 100644 --- a/tests/test_services/test_telephony/test_vobiz.py +++ b/tests/test_services/test_telephony/test_vobiz.py @@ -241,10 +241,17 @@ def test_carrier_frame_serializer_uses_plivo_credentials_for_plivo_calls(monkeyp integration = MagicMock() integration.auth_id = "enc-auth-id" integration.auth_token = "enc-auth-token" + pinned_id = uuid4() + + def _resolve(provider, db, organization_id, credential_id=None, **_kwargs): + assert provider == "plivo" + assert organization_id == org_id + assert credential_id == pinned_id + return integration monkeypatch.setattr( "app.services.telephony.carrier_media_serializer.resolve_telephony_integration", - lambda provider, db, organization_id, **_kwargs: integration, + _resolve, ) monkeypatch.setattr( "app.services.telephony.carrier_media_serializer.decrypt_api_key", @@ -257,6 +264,7 @@ def test_carrier_frame_serializer_uses_plivo_credentials_for_plivo_calls(monkeyp call_id="call-plivo", organization_id=org_id, db=MagicMock(), + telephony_integration_id=pinned_id, ) assert isinstance(serializer, PlivoFrameSerializer) @@ -266,9 +274,15 @@ def test_carrier_frame_serializer_uses_plivo_credentials_for_plivo_calls(monkeyp def test_carrier_frame_serializer_falls_back_to_platform_plivo_credentials(monkeypatch): + org_id = uuid4() + + def _resolve(*_args, credential_id=None, **_kwargs): + assert credential_id is None + return None + monkeypatch.setattr( "app.services.telephony.carrier_media_serializer.resolve_telephony_integration", - lambda *_args, **_kwargs: None, + _resolve, ) monkeypatch.setattr( "app.services.telephony.carrier_media_serializer.settings.PLIVO_AUTH_ID", @@ -285,7 +299,7 @@ def test_carrier_frame_serializer_falls_back_to_platform_plivo_credentials(monke provider_platform="plivo", stream_id="stream-plivo", call_id="call-plivo", - organization_id=uuid4(), + organization_id=org_id, db=MagicMock(), ) From 933312ba99560017899c27bf725a9e7140aa5699 Mon Sep 17 00:00:00 2001 From: Tejas Narayan Date: Fri, 4 Sep 2026 08:32:54 +0000 Subject: [PATCH 4/5] fix: frontend fixes --- frontend/src/pages/agents/AgentWorkspaceDetail.tsx | 1 + 1 file changed, 1 insertion(+) diff --git a/frontend/src/pages/agents/AgentWorkspaceDetail.tsx b/frontend/src/pages/agents/AgentWorkspaceDetail.tsx index 61183fc1..1b12af9b 100644 --- a/frontend/src/pages/agents/AgentWorkspaceDetail.tsx +++ b/frontend/src/pages/agents/AgentWorkspaceDetail.tsx @@ -40,6 +40,7 @@ function agentToFormData(agent: NonNullable Date: Fri, 4 Sep 2026 10:30:02 +0000 Subject: [PATCH 5/5] feat: updating telephony integration --- .../telephony/call_recording_lifecycle.py | 46 ++++++++++++------- .../test_call_recording_lifecycle.py | 42 +++++++++++++++++ 2 files changed, 71 insertions(+), 17 deletions(-) diff --git a/app/services/telephony/call_recording_lifecycle.py b/app/services/telephony/call_recording_lifecycle.py index 48250b17..ee6c150c 100644 --- a/app/services/telephony/call_recording_lifecycle.py +++ b/app/services/telephony/call_recording_lifecycle.py @@ -9,6 +9,7 @@ from typing import Any, Dict, Optional from uuid import UUID +from sqlalchemy import func, or_ from sqlalchemy.orm import Session from sqlalchemy.orm.attributes import flag_modified @@ -180,18 +181,27 @@ def _normalize_call_event(status: Optional[str]) -> str: return normalized +def _dialect_name(db: Session) -> str: + bind = db.get_bind() if hasattr(db, "get_bind") else getattr(db, "bind", None) + if bind is None: + return "postgresql" + return bind.dialect.name + + +def _call_data_text(db: Session, key: str): + """Return JSON text for call_data[key] on Postgres and SQLite.""" + if _dialect_name(db) == "sqlite": + return func.json_extract(CallRecording.call_data, f"$.{key}") + return func.json_extract_path_text(CallRecording.call_data, key) + + def _find_by_call_ref(db: Session, call_ref: str) -> Optional[CallRecording]: - rows = ( + return ( db.query(CallRecording) + .filter(_call_data_text(db, "call_ref") == call_ref) .order_by(CallRecording.created_at.desc()) - .limit(200) - .all() + .first() ) - for row in rows: - data = row.call_data if isinstance(row.call_data, dict) else {} - if data.get("call_ref") == call_ref: - return row - return None def find_call_recording( @@ -208,18 +218,20 @@ def find_call_recording( ) if row: return row - rows = ( + json_id_match = or_( + *( + _call_data_text(db, key) == provider_call_id + for key in ("request_uuid", "message_uuid", "api_id", "call_uuid") + ) + ) + row = ( db.query(CallRecording) - .filter(CallRecording.provider_platform == "vobiz") + .filter(json_id_match) .order_by(CallRecording.created_at.desc()) - .limit(200) - .all() + .first() ) - for candidate in rows: - data = candidate.call_data if isinstance(candidate.call_data, dict) else {} - for key in ("request_uuid", "message_uuid", "api_id", "call_uuid"): - if data.get(key) == provider_call_id: - return candidate + if row: + return row if call_ref: return _find_by_call_ref(db, call_ref) return None diff --git a/tests/test_services/test_telephony/test_call_recording_lifecycle.py b/tests/test_services/test_telephony/test_call_recording_lifecycle.py index 9e36f7df..b5da7238 100644 --- a/tests/test_services/test_telephony/test_call_recording_lifecycle.py +++ b/tests/test_services/test_telephony/test_call_recording_lifecycle.py @@ -1,5 +1,6 @@ """Tests for Vobiz call recording lifecycle helpers.""" +from datetime import datetime, timedelta, timezone from unittest.mock import patch from uuid import uuid4 @@ -61,6 +62,47 @@ def test_find_call_recording_matches_request_uuid_in_call_data(db_session, org_i assert found.id == row.id +def test_find_call_recording_by_call_ref_outside_recent_window( + db_session, org_id, seed_org, default_workspace +): + """Live Plivo hangups need the recording even when 200 newer rows exist.""" + now = datetime.now(timezone.utc) + target_ref = "plivo-live-session" + target = CallRecording( + organization_id=org_id, + workspace_id=default_workspace.id, + call_short_id="plivo1", + status=CallRecordingStatus.PENDING, + source=CallRecordingSource.WEBHOOK, + call_event="call_in_progress", + call_data={"call_ref": target_ref, "live_transcript": []}, + provider_platform="plivo", + created_at=now - timedelta(hours=1), + ) + db_session.add(target) + newer = [ + CallRecording( + organization_id=org_id, + workspace_id=default_workspace.id, + call_short_id=f"{index:06d}", + status=CallRecordingStatus.PENDING, + source=CallRecordingSource.WEBHOOK, + call_event="call_ended", + call_data={"call_ref": f"newer-{index}", "live_transcript": []}, + provider_platform="vobiz", + created_at=now + timedelta(seconds=index), + ) + for index in range(200) + ] + db_session.add_all(newer) + db_session.commit() + + found = find_call_recording(db_session, call_ref=target_ref) + assert found is not None + assert found.id == target.id + assert found.provider_platform == "plivo" + + def test_link_provider_call_id_updates_row(db_session, org_id, seed_org, default_workspace): call_ref = "ref-2" row = _make_vobiz_recording(