diff --git a/.gitignore b/.gitignore index e4ccaf3f..4fa3938e 100644 --- a/.gitignore +++ b/.gitignore @@ -19,3 +19,7 @@ coturn/ *.wav dograh_pcm_cache/ node_modules/ + +# Local self-signed TLS for the remote stack (regenerate with ./generate_certificate.sh). +# NEVER commit the private key. +certs/ diff --git a/api/services/telephony/ari_manager.py b/api/services/telephony/ari_manager.py index 2648affd..7d8ff40d 100644 --- a/api/services/telephony/ari_manager.py +++ b/api/services/telephony/ari_manager.py @@ -432,6 +432,46 @@ class ARIConnection: logger.info(f"[ARI org={self.organization_id}] Answered channel {channel_id}") return True + async def _get_channel_var(self, channel_id: str, variable: str) -> str: + """Read a channel variable/function via ARI. Returns '' if unset.""" + result = await self._ari_request( + "GET", f"/channels/{channel_id}/variable", params={"variable": variable} + ) + return (result or {}).get("value", "") or "" + + async def _capture_upstream_pbx(self, channel_id: str) -> Optional[dict]: + """Capture upstream-PBX (VICIdial) identity from the inbound SIP headers. + + When the call is patched in from VICIdial, the trunk INVITE carries + ``X-VICIDIAL-*`` headers. The callerid + remote-agent user are the handle + dograh uses to drive VICIdial's ``ra_call_control`` API for hangup and + transfer (the customer's real leg lives on VICIdial, not on dograh). + Returns None for non-upstream calls (no callerid header present). + """ + callerid = await self._get_channel_var( + channel_id, "PJSIP_HEADER(read,X-VICIDIAL-callerid)" + ) + if not callerid: + return None + upstream = { + "provider": "vicidial", + "callerid": callerid, + "agent_user": await self._get_channel_var( + channel_id, "PJSIP_HEADER(read,X-VICIDIAL-user)" + ), + "lead_id": await self._get_channel_var( + channel_id, "PJSIP_HEADER(read,X-VICIDIAL-lead_id)" + ), + "campaign_id": await self._get_channel_var( + channel_id, "PJSIP_HEADER(read,X-VICIDIAL-campaign_id)" + ), + } + logger.info( + f"[ARI org={self.organization_id}] Captured upstream_pbx for channel " + f"{channel_id}: {upstream}" + ) + return upstream + async def _create_external_media( self, workflow_id: str, @@ -566,6 +606,10 @@ class ARIConnection: # 3. Create workflow run call_id = channel_id + # Capture the upstream-PBX (VICIdial) identity off the SIP headers so + # the hangup/transfer strategies can drive VICIdial's API. The + # customer's real leg lives on VICIdial; this is the handle for it. + upstream_pbx = await self._capture_upstream_pbx(channel_id) workflow_run = await db_client.create_workflow_run( name=f"ARI Inbound {caller_number}", workflow_id=inbound_workflow_id, @@ -578,6 +622,7 @@ class ARIConnection: "direction": "inbound", "provider": "ari", "telephony_configuration_id": self.telephony_configuration_id, + "upstream_pbx": upstream_pbx, }, gathered_context={ "call_id": call_id, diff --git a/api/services/telephony/providers/ari/strategies.py b/api/services/telephony/providers/ari/strategies.py index 288110f3..6ac00182 100644 --- a/api/services/telephony/providers/ari/strategies.py +++ b/api/services/telephony/providers/ari/strategies.py @@ -201,6 +201,12 @@ class ARIHangupStrategy(HangupStrategy): ) return False + # If this call came from an upstream PBX (VICIdial), hang up its + # customer leg via its API FIRST -- so the upstream manages its own + # conference/remote-agent teardown cleanly instead of racing dograh's + # SIP BYE -- THEN drop dograh's own leg below. + await self._terminate_upstream_if_any(channel_id) + endpoint = f"{ari_endpoint}/ari/channels/{channel_id}" auth = BasicAuth(app_name, app_password) @@ -227,3 +233,38 @@ class ARIHangupStrategy(HangupStrategy): except Exception as e: logger.exception(f"Failed to hang up Asterisk channel: {e}") return False + + async def _terminate_upstream_if_any(self, channel_id: str) -> None: + """If this run came from an upstream PBX, hang up its customer leg via API. + + Reuses the same channel->run lookup the transfer strategy uses + (Redis ``ari:channel:{id}`` -> run_id -> ``initial_context``). Best-effort: + never blocks dograh's own hangup if the upstream call fails. + """ + try: + import redis.asyncio as aioredis + + from api.constants import REDIS_URL + from api.db import db_client + from api.services.telephony.upstream_pbx import terminate_upstream_call + + redis = aioredis.from_url(REDIS_URL, decode_responses=True) + run_id = await redis.get(f"ari:channel:{channel_id}") + if not run_id: + return + run = await db_client.get_workflow_run_by_id(int(run_id)) + if not run: + return + upstream = (run.initial_context or {}).get("upstream_pbx") + # If the call was already transferred to the upstream PBX, the customer + # leg has moved on -- do NOT hang it up (that would drop the transferred + # customer); just let dograh's own legs tear down below. + transferred = (run.gathered_context or {}).get("upstream_transferred") + if upstream and not transferred: + logger.info( + f"[ARI Hangup] Upstream PBX call ({upstream.get('provider')}); " + f"terminating customer leg via API before dropping dograh leg" + ) + await terminate_upstream_call(upstream) + except Exception as e: + logger.error(f"[ARI Hangup] upstream terminate check failed: {e}") diff --git a/api/services/telephony/upstream_pbx.py b/api/services/telephony/upstream_pbx.py new file mode 100644 index 00000000..5ff5f2be --- /dev/null +++ b/api/services/telephony/upstream_pbx.py @@ -0,0 +1,87 @@ +"""Upstream-PBX call control (POC). + +When a call originates on an upstream PBX (e.g. VICIdial) and is patched into +dograh over a SIP trunk, the *customer's* real call leg lives on the upstream +PBX, not on dograh's Asterisk -- dograh only owns its agent leg (the SIP leg +into Stasis + the externalMedia WebSocket). So when the AI decides to hang up or +transfer, dograh must tell the upstream PBX what to do via its API, and must do +so BEFORE tearing down its own SIP leg -- otherwise the SIP BYE races the +upstream PBX's own conference/remote-agent teardown and can leave it in an +inconsistent state. + +The upstream identity (the handle for these API calls) is captured off the +inbound SIP headers in ari_manager and stored on the workflow run's +``initial_context["upstream_pbx"]``. + +POC scope: the VICIdial agent-API connection is hardcoded below. The productized +version moves these into the ARI telephony-configuration credentials and selects +the adapter by ``upstream["provider"]`` (see the upstream-PBX seam design doc). +""" + +import aiohttp +from loguru import logger + +# --- POC hardcoded VICIdial agent-API connection (refine: telephony config creds) --- +_VICIDIAL_API_URL = "http://10.10.10.15/agc/api.php" +_VICIDIAL_API_USER = "6666" +_VICIDIAL_API_PASS = "1234" +_VICIDIAL_API_SOURCE = "dograh" + +_HTTP_TIMEOUT = aiohttp.ClientTimeout(total=8) + + +async def _ra_call_control(upstream: dict, stage: str, **extra) -> bool: + """Invoke VICIdial's agent API ``ra_call_control`` for the captured RA call. + + The call is identified by ``value`` (the VICIdial callerid captured from the + ``X-VICIDIAL-callerid`` header) plus the remote-agent ``agent_user``. + """ + params = { + "source": _VICIDIAL_API_SOURCE, + "user": _VICIDIAL_API_USER, + "pass": _VICIDIAL_API_PASS, + "agent_user": upstream.get("agent_user", ""), + "function": "ra_call_control", + "stage": stage, + "value": upstream.get("callerid", ""), + **extra, + } + try: + async with aiohttp.ClientSession() as session: + async with session.get( + _VICIDIAL_API_URL, params=params, timeout=_HTTP_TIMEOUT + ) as resp: + text = (await resp.text()).strip() + ok = text.startswith("SUCCESS") + logger.info( + f"[upstream_pbx] VICIdial ra_call_control {stage} " + f"(agent_user={params['agent_user']}, value={params['value']}) -> {text}" + ) + return ok + except Exception as e: + logger.error(f"[upstream_pbx] VICIdial ra_call_control {stage} failed: {e}") + return False + + +async def terminate_upstream_call(upstream: dict) -> bool: + """Hang up the upstream PBX's customer leg. Call this BEFORE dropping dograh's leg.""" + if not upstream or upstream.get("provider") != "vicidial": + return False + return await _ra_call_control(upstream, "HANGUP") + + +async def transfer_upstream_call(upstream: dict, destination: str) -> bool: + """Transfer the upstream PBX's customer leg (scaffolding for the transfer seam). + + ``ingroup:`` -> VICIdial INGROUPTRANSFER (to a queue/agent group); + anything else -> EXTENSIONTRANSFER to that number/extension (tolerates a + leading ``PJSIP/`` so dograh's existing transfer destination strings work). + """ + if not upstream or upstream.get("provider") != "vicidial": + return False + if destination.startswith("ingroup:"): + return await _ra_call_control( + upstream, "INGROUPTRANSFER", ingroup_choices=destination.split(":", 1)[1] + ) + number = destination.split("/")[-1] + return await _ra_call_control(upstream, "EXTENSIONTRANSFER", phone_number=number) diff --git a/api/services/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index 25298d73..6a75ac88 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/api/services/workflow/pipecat_engine_custom_tools.py @@ -554,6 +554,51 @@ class CustomToolManager: ) return + # Upstream-PBX (VICIdial) call: the customer's real leg lives on + # the upstream PBX, so transfer it via the upstream API + # (ra_call_control EXTENSIONTRANSFER / INGROUPTRANSFER) instead of + # dograh's ARI bridge-swap, which assumes dograh owns both legs. + upstream = (workflow_run.initial_context or {}).get("upstream_pbx") + if upstream: + from api.services.telephony.upstream_pbx import ( + transfer_upstream_call, + ) + + await self._play_config_message(config) + ok = await transfer_upstream_call(upstream, destination) + # Mark the run so the end-of-call hangup does NOT also hang up + # the now-transferred customer (see ARIHangupStrategy). + await db_client.update_workflow_run( + run_id=self._engine._workflow_run_id, + gathered_context={"upstream_transferred": True}, + ) + await function_call_params.result_callback( + { + "status": "success" if ok else "failed", + "action": "upstream_transfer", + "message": ( + "Transferring your call now." + if ok + else "I'm sorry, I couldn't complete the transfer." + ), + }, + properties=properties, + ) + if ok: + # Give the upstream PBX time to redirect the customer out + # of the conference toward the destination BEFORE we drop + # dograh's own leg -- otherwise the teardown races the + # redirect and cancels the customer's call to the + # destination (it never rings). + await asyncio.sleep(4) + # Tear down dograh's own legs; the customer has moved to the + # upstream PBX side. + await self._engine.end_call_with_reason( + EndTaskReason.END_CALL_TOOL_REASON.value, + abort_immediately=True, + ) + return + # Validate destination format based on workflow run mode if workflow_run.mode == WorkflowRunMode.ARI.value: # For ARI provider, also accept SIP endpoints diff --git a/docker-compose.override.yaml b/docker-compose.override.yaml new file mode 100644 index 00000000..44ee2b33 --- /dev/null +++ b/docker-compose.override.yaml @@ -0,0 +1,34 @@ +# Auto-generated by setup_remote.sh (build mode). +# Overrides docker-compose.yaml to build api and ui images from local source +# instead of pulling them from a registry. Remove this file to revert to +# pulling prebuilt images. +# +# NOTE: also carries the POC pbx-net attachment (see api.networks below). If +# setup_remote.sh regenerates this file, re-add the api networks block and the +# external pbx-net declaration, otherwise dograh->VICIdial API calls will time out. +services: + api: + build: + context: . + dockerfile: api/Dockerfile + image: dograh-local/dograh-api:local + pull_policy: never + # POC: attach api to the VICIdial PBX LAN so its Python (upstream_pbx.py) + # can reach VICIdial's agent API at 10.10.10.15. Without this the + # ra_call_control HANGUP/TRANSFER calls time out. Survives recreate. + networks: + - app-network + - pbx-net + + ui: + build: + context: . + dockerfile: ui/Dockerfile + image: dograh-local/dograh-ui:local + pull_policy: never + +# pbx-net is created/managed outside this compose project (the LXD/PBX bridge), +# so reference it as external — compose attaches to it but won't create/destroy it. +networks: + pbx-net: + external: true diff --git a/docker-compose.yaml b/docker-compose.yaml index 0bd27178..40c65c7d 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -133,7 +133,7 @@ services: environment: # Core application config ENVIRONMENT: "${ENVIRONMENT:-local}" - LOG_LEVEL: "INFO" + LOG_LEVEL: "DEBUG" # Replace this environment variable if you are using a custom # domain to host the stack diff --git a/generate_certificate.sh b/generate_certificate.sh new file mode 100755 index 00000000..62e1c778 --- /dev/null +++ b/generate_certificate.sh @@ -0,0 +1,7 @@ +#!/bin/bash +mkdir -p certs +openssl req -x509 -nodes -newkey rsa:2048 \ + -keyout certs/local.key \ + -out certs/local.crt \ + -days 365 \ + -subj "/CN=3.150.115.69"