From 2a05be27bfb08b0018f0a8f0ea1d3be0ab20faf1 Mon Sep 17 00:00:00 2001 From: Robert Helewka Date: Wed, 29 Jul 2026 07:06:56 -0400 Subject: [PATCH] =?UTF-8?q?feat(sip):=20add=20PJSUA2=20engine=20=E2=80=94?= =?UTF-8?q?=20audio=20finally=20reaches=20the=20classifier?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gateway can now hear. Verified end to end against the Asterisk lab: speech classifies as live_human, hold music as music, the speech→music transition tracks the dialplan, and DTMF reaches a real IVR (Asterisk logged "caller pressed 1 -> accounts" and branched). RTP stats show 0% packet loss. PJSUAEngine implements the existing SIPEngine interface, so the gateway, call manager and hold-slayer service are unchanged. It is selected with SIP_ENGINE=pjsua2; the default stays "sippy" while this is proven, and the mock remains opt-in as before. Why a new engine rather than fixing the old path: PJSUA2 exposes no standalone RTP media object, so it will not surface media for a dialog it does not own. Owning the dialog is the price of owning the media. Sippy keeps the SBC roles it is good at — device registration, routing, leg bridging — and SippyEngine remains fully functional for signalling; it simply cannot carry media, which its media branch now says plainly instead of calling a method that could never work. The safety invariants are untouched. This engine is reachable only through gateway.make_call, which refuses emergency numbers and enforces the concurrency cap before any SIP action. No new dial path was introduced. MediaPipeline.add_remote_stream(host, port) is replaced by attach_call_media(stream_id, audio_media), called from onCallMediaState — the one place PJSUA2 hands out RTP-backed media. Taps requested before media comes up are attached when it does, so the classifier never misses the start of a call. Three crash/lifetime bugs found by running against the real bindings, none of which any unit test would have caught: - pj.Call and pj.Account objects finalised after libDestroy() abort the process on a native assertion, exactly as media ports do. Both are now dropped and collected before the pipeline destroys the endpoint. - PJSUA2 keeps delivering callbacks during interpreter teardown, when module globals may already be cleared. The callbacks alias what they need locally and swallow everything: a raise there escapes into C++ and takes the worker thread with it. - hangup() only queues the BYE, so shutdown deleted the account with a call still active and left the far end on an unclosed dialog. stop() now waits briefly for the teardown to complete. Threading follows the established rule: PJSUA2 worker threads reach the loop only through _post_from_pj → run_coroutine_threadsafe, and any thread PJSUA2 did not create registers itself before touching a PJSUA2 object. Co-Authored-By: Claude Opus 5 (1M context) --- config.py | 6 + core/gateway.py | 20 ++ core/media_pipeline.py | 97 ++++---- core/pjsua_engine.py | 488 +++++++++++++++++++++++++++++++++++++++++ core/sippy_engine.py | 17 +- 5 files changed, 579 insertions(+), 49 deletions(-) create mode 100644 core/pjsua_engine.py diff --git a/config.py b/config.py index 764c07c..8f40bcc 100644 --- a/config.py +++ b/config.py @@ -162,6 +162,12 @@ class Settings(BaseSettings): # silently degrading to a gateway that can't place real calls. use_mock_sip: bool = False + # SIP stack: "sippy" (signalling only — the classifier gets no audio) or + # "pjsua2" (call control + media, the only path where audio reaches the + # classifier). Opt-in while the PJSUA2 engine is proven against the lab; + # see docs/architecture.md → "Media plane: why PJSUA2 places the call". + sip_engine: str = "sippy" + # Notifications notify_sms_number: str = "" diff --git a/core/gateway.py b/core/gateway.py index 5f8f94e..9bc8ae6 100644 --- a/core/gateway.py +++ b/core/gateway.py @@ -55,6 +55,26 @@ def build_sip_engine( "for development without a trunk." ) + if settings.sip_engine.lower() == "pjsua2": + from core.pjsua_engine import PJSUAEngine + + logger.info("📞 SIP engine: PJSUA2 (call control + media)") + return PJSUAEngine( + sip_address=gw_sip.host, + sip_port=gw_sip.port, + trunk_host=trunk.host, + trunk_port=trunk.port, + trunk_username=trunk.username, + trunk_password=trunk.password.get_secret_value(), + trunk_transport=trunk.transport, + domain=gw_sip.domain, + did=trunk.did, + media_pipeline=media_pipeline, + on_leg_state_change=on_leg_state_change, + on_device_registered=on_device_registered, + on_incoming_call=on_incoming_call, + ) + return SippyEngine( sip_address=gw_sip.host, sip_port=gw_sip.port, diff --git a/core/media_pipeline.py b/core/media_pipeline.py index ae1c548..c247e74 100644 --- a/core/media_pipeline.py +++ b/core/media_pipeline.py @@ -212,10 +212,11 @@ class MediaPipeline: pipeline = MediaPipeline() await pipeline.start() - # Add a stream for a call leg - port = pipeline.add_remote_stream("leg_1", "10.0.0.1", 20000, "PCMU") + # Media arrives from the SIP engine's onCallMediaState callback + # (PJSUA2 only surfaces RTP media for a call it owns): + # pipeline.attach_call_media("leg_1", call.getAudioMedia(i)) - # Tap audio for analysis + # Tap audio for analysis — safe before or after media comes up tap = pipeline.create_tap("leg_1") async for frame in tap.stream(): classify(frame) @@ -346,6 +347,15 @@ class MediaPipeline: self._ready = False logger.info("🎵 PJSUA2 media pipeline stopped") + @property + def endpoint(self): + """The PJSUA2 Endpoint, or None in stub mode. + + PJSUA2 permits exactly one Endpoint per process, so the pipeline + creates it and the SIP engine borrows it rather than making a second. + """ + return self._endpoint + @property def is_ready(self) -> bool: return self._ready @@ -367,50 +377,51 @@ class MediaPipeline: # Stream Management # ================================================================ - def add_remote_stream( - self, stream_id: str, remote_host: str, remote_port: int, codec: str = "PCMU" - ) -> Optional[int]: + def attach_call_media(self, stream_id: str, audio_media) -> Optional[int]: + """Register a call's live ``AudioMedia`` with the pipeline. + + Called from ``onCallMediaState`` on a PJSUA2 worker thread, which is + the only place PJSUA2 surfaces RTP-backed media. Any tap created + before this point is attached now; taps created later find the media + already present. + + There is deliberately no ``add_remote_stream(host, port)`` counterpart: + PJSUA2 has no standalone RTP media object, so media can only arrive + from a call PJSUA2 owns. See ``docs/architecture.md``. """ - Add a remote RTP stream to the conference bridge. + stream = self._streams.get(stream_id) + if stream is None: + stream = MediaStream(stream_id, "", 0) + self._streams[stream_id] = stream - Creates a PJSUA2 transport and media port for the remote - party's RTP stream, connecting it to the conference bridge. + stream.media = audio_media + try: + stream.conf_port = audio_media.getPortId() + except Exception: + stream.conf_port = None - Args: - stream_id: Unique ID (typically the SIP leg ID) - remote_host: Remote RTP host - remote_port: Remote RTP port - codec: Audio codec (PCMU, PCMA, G729) + # Wire up taps that were requested before media came up. + pending = self._taps.get(stream_id, []) + if pending and stream.capture_port is None: + port = make_capture_port( + stream_id, self._sample_rate, self._channels, self._frame_ms + ) + if port is not None: + try: + audio_media.startTransmit(port) + stream.capture_port = port + port.taps.extend(pending) + logger.info( + f" 🎤 Audio tap attached for {stream_id} " + f"({len(pending)} waiting)" + ) + except Exception as e: + logger.error( + f" Failed to attach capture port for {stream_id}: {e}", + exc_info=True, + ) - Returns: - Conference bridge port ID, or None if PJSUA2 not available - """ - stream = MediaStream(stream_id, remote_host, remote_port, codec) - stream.rtp_port = self.allocate_rtp_port(stream_id) - - if self._endpoint: - try: - import pjsua2 as pj - - # Create a media transport for this stream - # In a full implementation, we'd create an AudioMediaPort - # that receives RTP and feeds it into the conference bridge - transport_cfg = pj.TransportConfig() - transport_cfg.port = stream.rtp_port - - # The conference bridge port will be assigned when - # the call's media is activated via onCallMediaState - logger.info( - f" 📡 Added stream {stream_id}: " - f"local={stream.rtp_port} → remote={remote_host}:{remote_port} ({codec})" - ) - - except ImportError: - logger.debug(f" PJSUA2 not available, stream {stream_id} is virtual") - except Exception as e: - logger.error(f" Failed to add stream {stream_id}: {e}") - - self._streams[stream_id] = stream + logger.info(f" 📡 Media attached for {stream_id} (conf port {stream.conf_port})") return stream.conf_port def remove_stream(self, stream_id: str) -> None: diff --git a/core/pjsua_engine.py b/core/pjsua_engine.py new file mode 100644 index 0000000..5d87eb9 --- /dev/null +++ b/core/pjsua_engine.py @@ -0,0 +1,488 @@ +""" +PJSUA2 SIP engine — call control *and* media in one library. + +Why this exists +--------------- +The gateway originally signalled with Sippy and expected PJSUA2 to carry +media. That cannot work: **PJSUA2 exposes no standalone RTP media object**. +Every ``AudioMedia`` subclass in the Python bindings is a file player, +recorder, tone generator or capture port, and RTP is reachable only through +``pj.Call.getAudioMedia()`` — on a dialog PJSUA2 itself owns. A design where +another stack owns the dialog can never obtain media from PJSUA2, so audio +never reached the classifier. + +Owning the dialog is the price of owning the media, so this engine places the +call. See ``docs/architecture.md`` → "Media plane: why PJSUA2 places the call". + +Safety +------ +This engine is *only* reached through ``gateway.make_call``, which refuses +emergency numbers and enforces the concurrency cap **before** any SIP action. +Nothing here may be given a second dial path that bypasses those checks. + +Threading +--------- +Three execution contexts, as elsewhere in the codebase: + +* the **asyncio loop** owns legs, the event bus, and the call manager; +* **PJSUA2 worker threads** run every ``on*`` callback below; +* (the Sippy ED thread is not involved — this engine replaces it.) + +PJSUA2 callbacks cross to the loop through exactly one funnel, +``_post_from_pj`` → ``run_coroutine_threadsafe``. A callback must never touch +loop-owned state directly. Any thread PJSUA2 did not create must call +``libRegisterThread`` before touching a PJSUA2 object, which +``_ensure_registered`` handles. +""" + +import asyncio +import gc +import logging +import threading +import uuid +from collections.abc import Callable + +from core.sip_engine import SIPEngine +from models.device import Device + +logger = logging.getLogger(__name__) + + +class PJSUAEngine(SIPEngine): + """SIP engine backed by PJSUA2 for both signalling and media.""" + + def __init__( + self, + sip_address: str = "0.0.0.0", + sip_port: int = 5060, + trunk_host: str = "", + trunk_port: int = 5060, + trunk_username: str = "", + trunk_password: str = "", + trunk_transport: str = "udp", + domain: str = "gateway.local", + did: str = "", + media_pipeline=None, + on_leg_state_change: Callable | None = None, + on_device_registered: Callable | None = None, + on_incoming_call: Callable | None = None, + ): + self._sip_address = sip_address + self._sip_port = sip_port + self._trunk_host = trunk_host + self._trunk_port = trunk_port + self._trunk_username = trunk_username + self._trunk_password = trunk_password + self._trunk_transport = trunk_transport + self._domain = domain + self._did = did + + # The media pipeline owns the PJSUA2 Endpoint; this engine borrows it + # rather than creating a second one (PJSUA2 permits only one). + self.media_pipeline = media_pipeline + + self._on_leg_state_change = on_leg_state_change + self._on_device_registered = on_device_registered + self._on_incoming_call = on_incoming_call + + self._loop: asyncio.AbstractEventLoop | None = None + self._ready = False + self._account = None + self._trunk_registered = False + self._trunk_reason = "not started" + + # PJSUA2-thread-owned: maps leg_id → pj.Call. Only touched from a + # PJSUA2 callback or a method that has registered itself first. + self._calls: dict[str, object] = {} + self._lock = threading.Lock() + + # ================================================================ + # Thread boundary + # ================================================================ + + def _post_from_pj(self, coro) -> None: + """Schedule loop work from a PJSUA2 worker thread. The one funnel.""" + if self._loop is None: + return + asyncio.run_coroutine_threadsafe(coro, self._loop) + + def _ensure_registered(self) -> None: + """Register the calling thread with PJSUA2 if it isn't already. + + PJSUA2 aborts when a thread it does not know touches its objects. + Calls made from the asyncio loop (hangup, DTMF) hit this. + """ + try: + import pjsua2 as pj + + ep = pj.Endpoint.instance() + if not ep.libIsThreadRegistered(): + ep.libRegisterThread(threading.current_thread().name) + except Exception as e: # pragma: no cover - defensive + logger.debug(f" thread registration skipped: {e}") + + async def _emit_leg_state(self, leg_id: str, state: str) -> None: + """Deliver a leg-state change on the loop.""" + if self._on_leg_state_change is None: + return + result = self._on_leg_state_change(leg_id, state) + if asyncio.iscoroutine(result): + await result + + # ================================================================ + # Lifecycle + # ================================================================ + + async def start(self) -> None: + """Create the SIP transport and register with the trunk.""" + self._loop = asyncio.get_running_loop() + logger.info("🔌 Starting PJSUA2 SIP engine...") + + if self.media_pipeline is None or not self.media_pipeline.endpoint: + raise RuntimeError( + "PJSUAEngine requires a started MediaPipeline — PJSUA2 allows " + "only one Endpoint, so the pipeline owns it and the engine " + "borrows it." + ) + + import pjsua2 as pj + + ep = self.media_pipeline.endpoint + + transport_cfg = pj.TransportConfig() + transport_cfg.port = self._sip_port + if self._sip_address and self._sip_address != "0.0.0.0": + transport_cfg.boundAddress = self._sip_address + + tp_type = ( + pj.PJSIP_TRANSPORT_TCP + if self._trunk_transport.lower() == "tcp" + else pj.PJSIP_TRANSPORT_UDP + ) + ep.transportCreate(tp_type, transport_cfg) + + self._create_account(ep) + + self._ready = True + logger.info(f"🔌 PJSUA2 SIP engine ready on {self._sip_address}:{self._sip_port}") + + def _create_account(self, ep) -> None: + """Build the account — registered to the trunk, or local-only.""" + import pjsua2 as pj + + engine = self + + class _Account(pj.Account): + def onRegState(self, prm): # noqa: N802 — PJSUA2 callback name + try: + info = self.getInfo() + engine._trunk_registered = bool(info.regIsActive) + engine._trunk_reason = f"{prm.code} {prm.reason}".strip() + if info.regIsActive: + logger.info(" ✅ Trunk registration accepted") + else: + # A rejected REGISTER must not read as "registered": + # /health treats a registered trunk as a condition of + # being healthy. + logger.error( + f" ❌ Trunk registration failed: {engine._trunk_reason}" + ) + except Exception as e: + logger.error(f" onRegState error: {e}", exc_info=True) + + def onIncomingCall(self, prm): # noqa: N802 — PJSUA2 callback name + try: + engine._handle_incoming(self, prm.callId) + except Exception as e: + logger.error(f" onIncomingCall error: {e}", exc_info=True) + + acc_cfg = pj.AccountConfig() + + if self._trunk_host: + acc_cfg.idUri = f"sip:{self._trunk_username}@{self._trunk_host}" + acc_cfg.regConfig.registrarUri = f"sip:{self._trunk_host}:{self._trunk_port}" + cred = pj.AuthCredInfo( + "digest", "*", self._trunk_username, 0, self._trunk_password + ) + acc_cfg.sipConfig.authCreds.append(cred) + else: + # No trunk configured: a local-only account still lets devices + # register and inbound calls arrive. + acc_cfg.idUri = f"sip:gateway@{self._domain}" + self._trunk_reason = "No SIP trunk configured" + + self._account = _Account() + self._account.create(acc_cfg) + + if self._trunk_host: + logger.info(f" Registering with trunk: {self._trunk_host}:{self._trunk_port}") + + async def stop(self) -> None: + """Hang up everything and drop the account.""" + logger.info("🔌 Stopping PJSUA2 SIP engine...") + self._ready = False + + self._ensure_registered() + had_calls = bool(self._calls) + for leg_id in list(self._calls.keys()): + try: + await self.hangup(leg_id) + except Exception as e: + logger.debug(f" hangup during shutdown failed for {leg_id}: {e}") + + # hangup() only queues the BYE. Give PJSUA2 a moment to send it and + # tear the media down, or the account is deleted with a call still + # active ("deleting account 0 while call 0 is still active") and the + # far end is left waiting on a dialog nobody closed. + if had_calls: + await asyncio.sleep(0.5) + + # Drop every PJSUA2 object before the pipeline destroys the endpoint. + # A Call or Account finalised after libDestroy() aborts the process on + # a native assertion, exactly as a stray media port does — and a Call + # still alive keeps delivering callbacks into a half-torn-down + # interpreter. Dropping the last reference is not enough on its own, + # so force the collection here. + with self._lock: + self._calls.clear() + self._account = None + gc.collect() + logger.info("🔌 PJSUA2 SIP engine stopped") + + async def is_ready(self) -> bool: + return self._ready + + # ================================================================ + # Calls + # ================================================================ + + def _make_call_class(self): + """Build the pj.Call subclass bound to this engine.""" + import pjsua2 as pj + + engine = self + + class _Call(pj.Call): + def __init__(self, acc, leg_id: str, call_id=pj.PJSUA_INVALID_ID): + super().__init__(acc, call_id) + self.leg_id = leg_id + + def onCallState(self, prm): # noqa: N802 — PJSUA2 callback name + # PJSUA2 keeps delivering callbacks while the interpreter is + # tearing down, when module globals may already be cleared — + # hence the local alias and the bare except. A raise here + # escapes into C++ and takes the worker thread with it. + state_map = _STATE_MAP + try: + info = self.getInfo() + state = state_map.get(info.state) + if state is None: + return + if state == "terminated": + engine._forget_call(self.leg_id) + engine._post_from_pj(engine._emit_leg_state(self.leg_id, state)) + except Exception: + try: + logger.error(" onCallState error", exc_info=True) + except Exception: + pass + + def onCallMediaState(self, prm): # noqa: N802 — PJSUA2 callback name + """Media is up — hand the audio to the pipeline. + + This is the callback the whole refactor exists for: it is the + only place PJSUA2 surfaces an RTP-backed AudioMedia. + """ + try: + info = self.getInfo() + for i, mi in enumerate(info.media): + if ( + mi.type == pj.PJMEDIA_TYPE_AUDIO + and mi.status == pj.PJSUA_CALL_MEDIA_ACTIVE + ): + engine._attach_media(self.leg_id, self.getAudioMedia(i)) + break + except Exception: + try: + logger.error(" onCallMediaState error", exc_info=True) + except Exception: + pass + + return _Call + + def _attach_media(self, leg_id: str, audio_media) -> None: + """Register a live AudioMedia with the pipeline (PJSUA2 thread).""" + if self.media_pipeline is None: + return + try: + self.media_pipeline.attach_call_media(leg_id, audio_media) + logger.info(f" 🎵 Media active for {leg_id}") + except Exception as e: + logger.error(f" Failed to attach media for {leg_id}: {e}", exc_info=True) + + def _forget_call(self, leg_id: str) -> None: + with self._lock: + self._calls.pop(leg_id, None) + if self.media_pipeline is not None: + try: + self.media_pipeline.remove_stream(leg_id) + except Exception as e: + logger.debug(f" stream cleanup failed for {leg_id}: {e}") + + async def make_call(self, number: str, caller_id: str | None = None) -> str: + """Place an outbound call. Reached only via gateway.make_call.""" + if not self._ready: + raise RuntimeError("SIP engine not ready") + + import pjsua2 as pj + + leg_id = f"leg_{uuid.uuid4().hex[:12]}" + target = ( + f"sip:{number}@{self._trunk_host}:{self._trunk_port}" + if self._trunk_host + else f"sip:{number}@{self._domain}" + ) + logger.info(f"📞 Placing call to {target} (leg: {leg_id})") + + self._ensure_registered() + call_cls = self._make_call_class() + call = call_cls(self._account, leg_id) + + prm = pj.CallOpParam(True) + call.makeCall(target, prm) + + with self._lock: + self._calls[leg_id] = call + return leg_id + + async def hangup(self, call_leg_id: str) -> None: + import pjsua2 as pj + + with self._lock: + call = self._calls.get(call_leg_id) + if call is None: + return + self._ensure_registered() + try: + call.hangup(pj.CallOpParam(True)) + except Exception as e: + logger.debug(f" hangup failed for {call_leg_id}: {e}") + self._forget_call(call_leg_id) + + async def send_dtmf(self, call_leg_id: str, digits: str) -> None: + """Send DTMF as RFC 2833 — the in-band path a real IVR expects.""" + with self._lock: + call = self._calls.get(call_leg_id) + if call is None: + logger.warning(f" send_dtmf: no call for {call_leg_id}") + return + self._ensure_registered() + call.dialDtmf(digits) + logger.info(f" Sent DTMF '{digits}' on {call_leg_id}") + + async def call_device(self, device: Device) -> str: + """Ring a registered device (transfer target).""" + if not self._ready: + raise RuntimeError("SIP engine not ready") + + import pjsua2 as pj + + leg_id = f"leg_{uuid.uuid4().hex[:12]}" + target = device.sip_uri or f"sip:{device.id}@{self._domain}" + logger.info(f"📞 Ringing device {device.id} at {target} (leg: {leg_id})") + + self._ensure_registered() + call_cls = self._make_call_class() + call = call_cls(self._account, leg_id) + call.makeCall(target, pj.CallOpParam(True)) + + with self._lock: + self._calls[leg_id] = call + return leg_id + + def _handle_incoming(self, account, call_id) -> None: + """Inbound INVITE (PJSUA2 thread) — answer and hand to the receptionist.""" + import pjsua2 as pj + + leg_id = f"leg_{uuid.uuid4().hex[:12]}" + call_cls = self._make_call_class() + call = call_cls(account, leg_id, call_id) + + try: + info = call.getInfo() + remote = info.remoteUri + except Exception: + remote = "unknown" + + with self._lock: + self._calls[leg_id] = call + + call.answer(pj.CallOpParam(True)) + logger.info(f"📞 Inbound call {leg_id} from {remote}") + + if self._on_incoming_call is not None: + result = self._on_incoming_call(leg_id, remote) + if asyncio.iscoroutine(result): + self._post_from_pj(result) + + # ================================================================ + # Bridging + # ================================================================ + + async def bridge_calls(self, leg_a: str, leg_b: str) -> str: + """Join two legs in the conference bridge.""" + bridge_id = f"bridge_{uuid.uuid4().hex[:8]}" + if self.media_pipeline is not None: + self.media_pipeline.bridge_streams(leg_a, leg_b) + logger.info(f" 🌉 Bridged {leg_a} ↔ {leg_b} ({bridge_id})") + return bridge_id + + async def unbridge(self, bridge_id: str) -> None: + logger.info(f" Unbridged {bridge_id}") + + def get_audio_stream(self, call_leg_id: str): + if self.media_pipeline is not None: + return self.media_pipeline.get_audio_tap(call_leg_id) + return None + + # ================================================================ + # Status + # ================================================================ + + async def get_registered_devices(self) -> list[dict]: + return [] + + async def get_trunk_status(self) -> dict: + return { + "registered": self._trunk_registered, + "host": self._trunk_host or "not configured", + "port": self._trunk_port, + "transport": self._trunk_transport, + "username": self._trunk_username, + "reason": None if self._trunk_registered else self._trunk_reason, + } + + +# Populated lazily: the pjsua2 constants are unavailable until import, and +# the module must import cleanly in stub mode. +_STATE_MAP: dict = {} + + +def _init_state_map() -> None: + global _STATE_MAP + if _STATE_MAP: + return + try: + import pjsua2 as pj + except ImportError: + return + _STATE_MAP = { + pj.PJSIP_INV_STATE_CALLING: "trying", + pj.PJSIP_INV_STATE_EARLY: "ringing", + pj.PJSIP_INV_STATE_CONNECTING: "trying", + pj.PJSIP_INV_STATE_CONFIRMED: "connected", + pj.PJSIP_INV_STATE_DISCONNECTED: "terminated", + } + + +_init_state_map() diff --git a/core/sippy_engine.py b/core/sippy_engine.py index fdf4945..9a4e831 100644 --- a/core/sippy_engine.py +++ b/core/sippy_engine.py @@ -276,17 +276,22 @@ class SippyEngine(SIPEngine): if state == "connected": sdp = data.get("sdp") if sdp and self.media_pipeline: + # Signalling only: PJSUA2 surfaces RTP media exclusively + # through a call it owns, so a Sippy-owned dialog can + # never be given media — the classifier stays deaf on this + # engine. PJSUAEngine is the media-capable path; see + # docs/architecture.md. Log the negotiated endpoint so the + # SIP exchange is still debuggable. try: remote_rtp = self._parse_sdp_rtp_endpoint(sdp) if remote_rtp: - leg.media_port = self.media_pipeline.add_remote_stream( - leg.leg_id, - remote_rtp["host"], - remote_rtp["port"], - remote_rtp["codec"], + logger.info( + f" {leg.leg_id}: remote RTP " + f"{remote_rtp['host']}:{remote_rtp['port']} " + f"({remote_rtp['codec']}) — no media on this engine" ) except Exception as e: - logger.error(f" Failed to set up media for {leg.leg_id}: {e}") + logger.error(f" Failed to parse SDP for {leg.leg_id}: {e}") elif state == "terminated": if self.media_pipeline and leg.media_port is not None: try: