diff --git a/api/services/telephony/ari_manager.py b/api/services/telephony/ari_manager.py index 2648affd..58920d9c 100644 --- a/api/services/telephony/ari_manager.py +++ b/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 " diff --git a/api/services/telephony/providers/ari/strategies.py b/api/services/telephony/providers/ari/strategies.py index 288110f3..2d7e1cb3 100644 --- a/api/services/telephony/providers/ari/strategies.py +++ b/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( diff --git a/api/tests/telephony/test_ari_manager_transfer_teardown.py b/api/tests/telephony/test_ari_manager_transfer_teardown.py new file mode 100644 index 00000000..6490d650 --- /dev/null +++ b/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))