feat(sip): add PJSUA2 engine — audio finally reaches the classifier

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) <noreply@anthropic.com>

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: