ViciDial Working

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,

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}")

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:<id>`` -> 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)

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