fix: clean up ARI transferred call legs on participant hangup (#498)

api/services/telephony/ari_manager.py

@@ -119,11 +119,20 @@ class ARIConnection:
         r = await self._get_redis()
         return await r.exists(f"{_EXT_CHANNEL_KEY_PREFIX}{channel_id}") > 0
 
-    async def _delete_ext_channel(self, channel_id: str):
+    async def _delete_ext_channel(self, channel_id: Optional[str]):
         """Remove the external media channel marker."""
+        if not channel_id:
+            return
         r = await self._get_redis()
         await r.delete(f"{_EXT_CHANNEL_KEY_PREFIX}{channel_id}")
 
+    async def _delete_transfer_channel_mapping(self, channel_id: Optional[str]):
+        """Remove transfer destination channel correlation marker."""
+        if not channel_id:
+            return
+        r = await self._get_redis()
+        await r.delete(f"ari:transfer_channel:{channel_id}")
+
     async def _set_pending_bridge(
         self,
         ext_channel_id: str,
@@ -766,6 +775,9 @@ class ARIConnection:
             ext_channel_id = ctx.get("ext_channel_id")
             bridge_id = ctx.get("bridge_id")
             transfer_state = ctx.get("transfer_state")
+            transfer_bridge_id = ctx.get("transfer_bridge_id") or bridge_id
+            transfer_caller_channel_id = ctx.get("transfer_caller_channel_id") or call_id
+            transfer_destination_channel_id = ctx.get("transfer_destination_channel_id")
 
             # Check if this is a call transfer scenario external channel. Skip full teardown if
             # transfer is in progress and this is the external media channel
@@ -796,6 +808,63 @@ class ARIConnection:
                 )
                 return
 
+            if (
+                transfer_state == "complete"
+                and transfer_bridge_id
+                and transfer_caller_channel_id
+                and transfer_destination_channel_id
+                and channel_id in (
+                    transfer_caller_channel_id,
+                    transfer_destination_channel_id,
+                )
+            ):
+                peer_channel_id = (
+                    transfer_destination_channel_id
+                    if channel_id == transfer_caller_channel_id
+                    else transfer_caller_channel_id
+                )
+                logger.info(
+                    f"[ARI org={self.organization_id}] Completed transfer participant "
+                    f"{channel_id} left Stasis; tearing down peer {peer_channel_id} "
+                    f"and bridge {transfer_bridge_id}"
+                )
+
+                # Mark terminal state before issuing ARI deletes so duplicate
+                # StasisEnd events run through the idempotent normal cleanup path.
+                ctx["transfer_state"] = "terminated"
+                await db_client.update_workflow_run(
+                    run_id=int(workflow_run_id), gathered_context=ctx
+                )
+
+                await self._delete_bridge(transfer_bridge_id)
+                if peer_channel_id and peer_channel_id != channel_id:
+                    await self._delete_channel(peer_channel_id)
+
+                keys_to_delete = [
+                    cid
+                    for cid in (
+                        transfer_caller_channel_id,
+                        transfer_destination_channel_id,
+                        ext_channel_id,
+                        channel_id,
+                    )
+                    if cid
+                ]
+                if keys_to_delete:
+                    await self._delete_channel_run(*keys_to_delete)
+
+                await self._delete_ext_channel(ext_channel_id)
+                await self._delete_transfer_channel_mapping(
+                    transfer_destination_channel_id
+                )
+
+                logger.info(
+                    f"[ARI org={self.organization_id}] Completed transfer teardown "
+                    f"finished for channel={channel_id}, peer={peer_channel_id}, "
+                    f"bridge={transfer_bridge_id}"
+                )
+                return
+
             # Normal full teardown for non-transfer scenarios (transfer_state is None or not in-progress)
             # Delete the bridge first (removes channels from it)
             if bridge_id:
@@ -815,6 +884,9 @@ class ARIConnection:
 
             # Clean up the Redis marker for external channel
             await self._delete_ext_channel(ext_channel_id)
+            await self._delete_transfer_channel_mapping(
+                transfer_destination_channel_id
+            )
 
             logger.info(
                 f"[ARI org={self.organization_id}] StasisEnd full teardown for "

api/services/telephony/providers/ari/strategies.py

@@ -103,8 +103,16 @@ class ARIBridgeSwapStrategy(TransferStrategy):
                 f"destination={destination_channel_id}, ext_media={ext_channel_id}"
             )
 
-            # 3. Set transfer state to prevent StasisEnd auto-teardown
-            workflow_run.gathered_context["transfer_state"] = "in-progress"
+            # 3. Set transfer state to prevent StasisEnd auto-teardown and
+            # persist the transferred pair for post-handoff participant cleanup.
+            workflow_run.gathered_context.update(
+                {
+                    "transfer_state": "in-progress",
+                    "transfer_bridge_id": bridge_id,
+                    "transfer_caller_channel_id": channel_id,
+                    "transfer_destination_channel_id": destination_channel_id,
+                }
+            )
             await db_client.update_workflow_run(
                 run_id=int(workflow_run_id),
                 gathered_context=workflow_run.gathered_context,
@@ -124,6 +132,13 @@ class ARIBridgeSwapStrategy(TransferStrategy):
                         logger.info(
                             f"[ARI Transfer] Added destination {destination_channel_id} to bridge {bridge_id}"
                         )
+                        # Let ari_manager route StasisEnd for the destination leg
+                        # back to this workflow after transfer context cleanup.
+                        await redis.setex(
+                            f"ari:channel:{destination_channel_id}",
+                            3600,
+                            workflow_run_id,
+                        )
                     else:
                         error_text = await response.text()
                         logger.error(
@@ -169,6 +184,7 @@ class ARIBridgeSwapStrategy(TransferStrategy):
             )
 
             # 5. Clean up transfer context after successful completion
+            await redis.delete(f"ari:transfer_channel:{destination_channel_id}")
 
             call_transfer_manager = await get_call_transfer_manager()
             await call_transfer_manager.remove_transfer_context(

api/tests/telephony/test_ari_manager_transfer_teardown.py

@@ -0,0 +1,118 @@
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from api.services.telephony import ari_manager
+from api.services.telephony.ari_manager import ARIConnection
+
+
+class _FakeDbClient:
+    def __init__(self, gathered_context):
+        self.workflow_run = SimpleNamespace(gathered_context=gathered_context)
+        self.updated_contexts = []
+
+    async def get_workflow_run_by_id(self, run_id: int):
+        return self.workflow_run
+
+    async def update_workflow_run(self, run_id: int, gathered_context: dict):
+        self.updated_contexts.append((run_id, dict(gathered_context)))
+        self.workflow_run.gathered_context = gathered_context
+
+
+class _RecordingARIConnection(ARIConnection):
+    def __init__(self):
+        super().__init__(
+            organization_id=1,
+            telephony_configuration_id=10,
+            ari_endpoint="http://asterisk.test:8088",
+            app_name="dograh",
+            app_password="secret",
+            ws_client_name="dograh_ws",
+        )
+        self.deleted_bridges = []
+        self.deleted_channels = []
+        self.deleted_channel_runs = []
+        self.deleted_ext_channels = []
+        self.deleted_transfer_channel_mappings = []
+
+    async def _delete_bridge(self, bridge_id: str):
+        self.deleted_bridges.append(bridge_id)
+
+    async def _delete_channel(self, channel_id: str):
+        self.deleted_channels.append(channel_id)
+
+    async def _delete_channel_run(self, *channel_ids: str):
+        self.deleted_channel_runs.extend(channel_ids)
+
+    async def _delete_ext_channel(self, channel_id: str | None):
+        self.deleted_ext_channels.append(channel_id)
+
+    async def _delete_transfer_channel_mapping(self, channel_id: str | None):
+        self.deleted_transfer_channel_mappings.append(channel_id)
+
+    async def _get_transfer_id_for_channel(self, channel_id: str):
+        return None
+
+    async def _handle_transfer_failed(
+        self, transfer_id: str, channel_id: str, reason: str
+    ):
+        raise AssertionError("completed transfer hangup should not publish transfer failure")
+
+
+def _completed_transfer_context():
+    return {
+        "call_id": "caller-chan",
+        "ext_channel_id": "ext-chan",
+        "bridge_id": "bridge-1",
+        "transfer_state": "complete",
+        "transfer_bridge_id": "bridge-1",
+        "transfer_caller_channel_id": "caller-chan",
+        "transfer_destination_channel_id": "dest-chan",
+    }
+
+
+@pytest.mark.asyncio
+async def test_completed_transfer_tears_down_destination_when_caller_leaves(monkeypatch):
+    fake_db = _FakeDbClient(_completed_transfer_context())
+    monkeypatch.setattr(ari_manager, "db_client", fake_db)
+    conn = _RecordingARIConnection()
+
+    await conn._handle_stasis_end("caller-chan", "123")
+
+    assert conn.deleted_bridges == ["bridge-1"]
+    assert conn.deleted_channels == ["dest-chan"]
+    assert "caller-chan" in conn.deleted_channel_runs
+    assert "dest-chan" in conn.deleted_channel_runs
+    assert conn.deleted_transfer_channel_mappings == ["dest-chan"]
+    assert fake_db.workflow_run.gathered_context["transfer_state"] == "terminated"
+
+
+@pytest.mark.asyncio
+async def test_completed_transfer_tears_down_caller_when_destination_leaves(monkeypatch):
+    fake_db = _FakeDbClient(_completed_transfer_context())
+    monkeypatch.setattr(ari_manager, "db_client", fake_db)
+    conn = _RecordingARIConnection()
+
+    await conn._handle_stasis_end("dest-chan", "123")
+
+    assert conn.deleted_bridges == ["bridge-1"]
+    assert conn.deleted_channels == ["caller-chan"]
+    assert "caller-chan" in conn.deleted_channel_runs
+    assert "dest-chan" in conn.deleted_channel_runs
+    assert conn.deleted_transfer_channel_mappings == ["dest-chan"]
+    assert fake_db.workflow_run.gathered_context["transfer_state"] == "terminated"
+
+
+@pytest.mark.asyncio
+async def test_completed_transfer_destination_destroyed_without_transfer_mapping_is_ignored():
+    conn = _RecordingARIConnection()
+    event = {
+        "type": "ChannelDestroyed",
+        "channel": {"id": "dest-chan", "state": "Up"},
+        "cause": 16,
+        "cause_txt": "Normal Clearing",
+        "tech_cause": "unknown",
+    }
+
+    await conn._handle_event(json.dumps(event))