diff --git a/core/media_pipeline.py b/core/media_pipeline.py index 19d3b16..ae1c548 100644 --- a/core/media_pipeline.py +++ b/core/media_pipeline.py @@ -20,6 +20,7 @@ PJSUA2 runs in its own thread with a dedicated Endpoint. """ import asyncio +import gc import logging import threading from collections.abc import AsyncIterator @@ -98,6 +99,72 @@ class AudioTap: self._active = False +def make_capture_port(stream_id: str, sample_rate: int, channels: int, frame_ms: int): + """Build a PJSUA2 media port that forks conference audio into taps. + + Defined as a factory rather than a module-level class because + ``pj.AudioMediaPort`` can only be subclassed once ``pjsua2`` imports — + and the whole pipeline degrades to stub mode when it doesn't. + + The returned port is a *sink*: the conference bridge transmits into it, + and every frame is copied to each registered tap. Returns ``None`` when + pjsua2 is unavailable. + """ + try: + import pjsua2 as pj + except ImportError: + return None + + class _CapturePort(pj.AudioMediaPort): + """Receives conference-bridge frames and fans them out to taps. + + ``onFrameReceived`` is called on a **PJSUA2 worker thread** — a third + execution context alongside the asyncio loop and the Sippy ED thread. + It must touch nothing but ``AudioTap.feed``, which is explicitly + thread-safe (it hops to the owning loop via ``call_soon_threadsafe``). + Reaching into pipeline state, the event bus, or a Sippy object from + here would be a data race. + """ + + def __init__(self, stream_id: str): + super().__init__() + self.stream_id = stream_id + self.taps: list[AudioTap] = [] + self._logged_error = False + + def onFrameReceived(self, frame): # noqa: N802 — PJSUA2 C++ callback name + try: + if not self.taps or frame.size <= 0: + return + # frame.buf is a SWIG ByteVector of signed chars; the tap + # contract is raw little-endian 16-bit PCM. + pcm = bytes(bytearray(b & 0xFF for b in frame.buf)) + for tap in self.taps: + tap.feed(pcm) + except Exception as e: + # An exception escaping into PJSUA2's C++ callback would tear + # down the worker thread and silently kill media for every + # call. Log once per port rather than on every 20ms frame. + if not self._logged_error: + self._logged_error = True + logger.error( + f" Audio capture failed for {self.stream_id}: {e}", + exc_info=True, + ) + + fmt = pj.MediaFormatAudio() + fmt.init( + pj.PJMEDIA_FORMAT_L16, + sample_rate, + channels, + frame_ms * 1000, # frameTimeUsec + 16, # bitsPerSample + ) + port = _CapturePort(stream_id) + port.createPort(f"tap-{stream_id}", fmt) + return port + + # ================================================================ # Stream Entry — tracks a single media stream in the pipeline # ================================================================ @@ -112,6 +179,8 @@ class MediaStream: self.codec = codec self.conf_port: Optional[int] = None # PJSUA2 conference bridge port ID self.transport = None # PJSUA2 SipTransport + self.media = None # PJSUA2 AudioMedia for this stream + self.capture_port = None # Shared _CapturePort feeding this stream's taps self.rtp_port: Optional[int] = None # Local RTP listen port self.taps: list[AudioTap] = [] self.recorder = None # PJSUA2 AudioMediaRecorder @@ -173,6 +242,7 @@ class MediaPipeline: self._next_rtp_port = rtp_start_port self._sample_rate = sample_rate self._channels = channels + self._frame_ms = 20 # Must match medConfig.audioFramePtime below self._null_audio = null_audio # Use null audio device (no sound card needed) # State @@ -255,11 +325,17 @@ class MediaPipeline: tap.close() self._taps.clear() - # Remove all streams + # Remove all streams (this releases their capture ports) for stream_id in list(self._streams.keys()): self.remove_stream(stream_id) - # Destroy PJSUA2 endpoint + # Destroy PJSUA2 endpoint. Every media port must be collected first: + # a port finalised after libDestroy() runs pjmedia_conf_remove_port + # against a freed conference bridge and aborts the process. Dropping + # the last Python reference is not enough on its own — force the + # collection here rather than leaving it to interpreter exit. + gc.collect() + if self._endpoint: try: self._endpoint.libDestroy() @@ -350,6 +426,19 @@ class MediaPipeline: tap.close() self._taps.pop(stream_id, None) + # Release the capture port while the conference bridge still exists. + # A port garbage-collected after Endpoint.libDestroy() calls + # pjmedia_conf_remove_port against a freed bridge and aborts the + # process on a native assertion — a hard crash, not an exception. + if stream.capture_port is not None: + try: + if stream.media is not None: + stream.media.stopTransmit(stream.capture_port) + except Exception as e: + logger.debug(f" stopTransmit failed for {stream_id}: {e}") + stream.capture_port.taps.clear() + stream.capture_port = None + # Stop recording if stream.recorder: try: @@ -426,16 +515,28 @@ class MediaPipeline: self._taps[stream_id] = [] self._taps[stream_id].append(tap) - if self._endpoint and stream and stream.conf_port is not None: + if self._endpoint and stream and stream.media is not None: try: - import pjsua2 as pj - # Create an AudioMediaPort that captures frames - # and feeds them to the tap - # In PJSUA2, we'd subclass AudioMediaPort and implement - # onFrameReceived to call tap.feed(frame_data) - logger.info(f" 🎤 Audio tap created for {stream_id} (PJSUA2)") + # One capture port per stream, shared by every tap on it: + # the bridge would otherwise mix each additional port back + # into the conference and the call would echo. + if stream.capture_port is None: + port = make_capture_port( + stream_id, self._sample_rate, self._channels, self._frame_ms + ) + if port is not None: + # The stream's media transmits into the capture port, + # not the reverse — the port is a sink. + stream.media.startTransmit(port) + stream.capture_port = port + logger.info(f" 🎤 Audio tap created for {stream_id} (PJSUA2)") + + if stream.capture_port is not None: + stream.capture_port.taps.append(tap) except Exception as e: - logger.error(f" Failed to create PJSUA2 tap for {stream_id}: {e}") + logger.error( + f" Failed to create PJSUA2 tap for {stream_id}: {e}", exc_info=True + ) else: logger.info(f" 🎤 Audio tap created for {stream_id} (virtual)") diff --git a/docs/architecture.md b/docs/architecture.md index f8ef511..1d35c00 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,6 +1,15 @@ # Architecture -Hold Slayer is a single-process async Python application built on FastAPI. It acts as an intelligent B2BUA (Back-to-Back User Agent) sitting between your SIP trunk (PSTN access) and your desk phone/softphone. +Hold Slayer is a single-process async Python application built on FastAPI. It +acts as an intelligent B2BUA (Back-to-Back User Agent) sitting between your SIP +trunk (PSTN access) and your desk phone/softphone. + +> **Media plane in transition.** The gateway currently signals with Sippy and +> intends PJSUA2 to carry media, but PJSUA2 will not surface an RTP stream for +> a dialog it does not own — so no audio ever reaches the classifier. The fix +> moves *call placement* into PJSUA2 while Sippy keeps the SBC roles. See +> [Media plane: why PJSUA2 places the call](#media-plane-why-pjsua2-places-the-call) +> before changing anything in `core/`. ## System Diagram @@ -27,8 +36,13 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac │ └────┬─────┘ └─────┬─────┘ └──────────────┘ │ │ │ │ │ │ ┌────┴──────────────┴───────────────────┐ │ -│ │ Sippy B2BUA Engine │ │ -│ │ (SIP calls, DTMF, conference bridge) │ │ +│ │ SIP Engine │ │ +│ │ signalling + call control │ │ +│ └────┬──────────────────────────────────┘ │ +│ │ │ +│ ┌────┴──────────────────────────────────┐ │ +│ │ Media Pipeline (PJSUA2) │ │ +│ │ RTP, conference bridge, taps, record │ │ │ └────┬──────────────────────────────────┘ │ │ │ │ └───────┼─────────────────────────────────────────────────────────┘ @@ -70,8 +84,8 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac | Component | File | Purpose | |-----------|------|---------| -| Sippy Engine | `core/sippy_engine.py` | SIP signaling (INVITE, BYE, REGISTER, DTMF) | -| Media Pipeline | `core/media_pipeline.py` | PJSUA2 RTP media handling, conference bridge, recording | +| Sippy Engine | `core/sippy_engine.py` | SIP signalling (INVITE, BYE, REGISTER, DTMF) | +| Media Pipeline | `core/media_pipeline.py` | PJSUA2 RTP media, conference bridge, taps, recording | | Recording | `services/recording.py` | WAV file management and storage | | Analytics | `services/call_analytics.py` | Call metrics, hold time stats, trends | | Notifications | `services/notification.py` | WebSocket + SMS alerts | @@ -84,9 +98,10 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac POST /api/v1/calls/hold-slayer { number, intent, call_flow_id } │ 2. Gateway.make_call() - ├── CallManager.create_call() → track state - ├── SippyEngine.make_call() → SIP INVITE to trunk - └── MediaPipeline.add_stream() → RTP media setup + ├── is_emergency_number() → REFUSE 911/112 (before anything else) + ├── concurrency cap check → refuse past max_concurrent_calls + ├── CallManager.create_call() → track state + └── sip_engine.make_call() → place the call, media follows │ 3. HoldSlayer.run_with_flow() or run_exploration() ├── AudioClassifier.classify() → analyze 3s audio windows @@ -99,14 +114,14 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac ├── TranscriptionService.transcribe() → STT on speech audio │ ├── LLMClient.analyze_ivr_menu() → pick menu option (fallback) - │ └── SippyEngine.send_dtmf() → press the button + │ └── sip_engine.send_dtmf() → press the button │ └── detect_hold_to_human_transition() └── HUMAN_DETECTED! → transfer │ 4. Transfer - ├── SippyEngine.bridge() → connect call legs - ├── MediaPipeline.bridge_streams() → bridge RTP + ├── SippyEngine.bridge_calls() → join the two call legs + ├── MediaPipeline.bridge_streams() → bridge RTP in the conf bridge ├── EventBus.publish(TRANSFER_STARTED) └── NotificationService → "Pick up your phone!" │ @@ -117,44 +132,95 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac → Analytics tracking ``` +The emergency guard and the concurrency cap are the first two steps of +`make_call` for a reason, and their order is load-bearing — see +[.claude/rules/call-safety.md](../.claude/rules/call-safety.md). + ## Threading Model -Hold Slayer is primarily single-threaded async (asyncio), with one exception: - -- **Main thread**: FastAPI + all async services (event bus, hold slayer, classifier, etc.) -- **Sippy thread**: Sippy B2BUA runs its own event loop in a dedicated daemon thread. The `SippyEngine` bridges async↔sync via `asyncio.run_in_executor()`. -- **PJSUA2**: Runs in the main thread using null audio device (no sound card needed — headless server mode). +The README's "single-process async" is a simplification. There are **three** +execution contexts, and the boundaries between them are the highest-leverage +invariant in the codebase. ``` -Main Thread (asyncio) -├── FastAPI (uvicorn) -├── EventBus -├── CallManager -├── HoldSlayer +asyncio loop (main thread) Sippy ED thread PJSUA2 worker threads +├── FastAPI (uvicorn) └── ED2 dispatcher └── media / RTP +├── EventBus ├── SIP signalling └── onFrameReceived +├── CallManager ├── UA objects +├── HoldSlayer └── DTMF relay ├── AudioClassifier ├── TranscriptionService ├── LLMClient -├── MediaPipeline (PJSUA2) ├── NotificationService └── RecordingService - -Sippy Thread (daemon) -└── Sippy B2BUA event loop - ├── SIP signaling - ├── DTMF relay - └── Call leg management ``` +**Crossing the boundaries — one funnel each way:** + +| Direction | Mechanism | Notes | +|---|---|---| +| Sippy ED → loop | `_post_from_ed` → `asyncio.run_coroutine_threadsafe` → `_on_engine_event` | The single funnel where Sippy-thread events mutate loop state | +| loop → Sippy ED | `_run_on_sippy` → `ED2.callFromThread` | Anything touching a Sippy UA object | +| PJSUA2 worker → loop | `AudioTap.feed` → `loop.call_soon_threadsafe` | The **only** thing a PJSUA2 callback may touch | + +`onFrameReceived` runs on a PJSUA2 worker thread every 20 ms. It must call +nothing but `AudioTap.feed`; reaching into pipeline state, the event bus, or a +Sippy object from there is a data race. An exception escaping into PJSUA2's C++ +callback tears down the worker thread and silently kills media for every call, +which is why the capture port catches and logs once rather than per frame. + +Full detail: [.claude/rules/concurrency-threads.md](../.claude/rules/concurrency-threads.md). + ## Design Decisions -### Why Sippy B2BUA + PJSUA2? +### Media plane: why PJSUA2 places the call -We split SIP signaling and media handling into two separate libraries: +The original split was *Sippy signals, PJSUA2 carries media*. It does not work, +for a reason that is not obvious until you try it: -- **Sippy B2BUA** handles SIP signaling (INVITE, BYE, REGISTER, re-INVITE, DTMF relay). It's battle-tested for telephony and handles the complex SIP state machine. -- **PJSUA2** handles RTP media (audio streams, conference bridge, recording, tone generation). It provides a clean C++/Python API for media manipulation without needing to deal with raw RTP. +**PJSUA2 exposes no standalone RTP media object.** Every `AudioMedia` subclass +in the Python bindings is a file player, recorder, tone generator, or capture +port. RTP is reachable only through `pj.Call.getAudioMedia()`, after +`onCallMediaState` fires on a dialog **PJSUA2 itself owns**. There is no +"give me an AudioMedia for this remote host:port" API to call. -This split lets us tap into the audio stream (for classification and STT) without interfering with SIP signaling, and bridge calls through a conference bridge for clean transfer. +So a design where Sippy owns the dialog can never obtain a media stream from +PJSUA2. `MediaPipeline.add_remote_stream()` is not unfinished work — it is a +function that cannot be written against this API. The consequence is that audio +never reaches the classifier: `create_tap` builds a valid capture port with +nothing to attach it to. + +**The resolution: PJSUA2 places the call; Sippy keeps every other role.** + +| Concern | Owner | +|---|---| +| Emergency guard, concurrency cap | `gateway.make_call` — unchanged, still first | +| Trunk registration | PJSUA2 `Account` | +| Outbound INVITE / answer / hangup | PJSUA2 `Call` | +| RTP, conference bridge, taps, recording | PJSUA2 media | +| DTMF | PJSUA2 `Call.dialDtmf` (RFC 2833) | +| Device registration, routing, leg bridging | Sippy / gateway | +| Inbound call dispatch | PJSUA2 `Account.onIncomingCall` | + +Sippy remains the SBC-shaped layer — it is where device registrations, routing +decisions and B2BUA leg-joining live. What moves is the raw dialog for a trunk +call, because owning the dialog is the price of owning the media. + +Alternatives considered and rejected: + +- **Terminate RTP ourselves** (aiortc or raw sockets) and feed PCM into + `AudioTap` directly, keeping Sippy on the wire. Preserves the split, but + means owning jitter buffering, packet loss concealment and ulaw/alaw + transcoding — precisely the work PJSUA2 exists to do. +- **A loopback `pj.Call` mirroring each real leg**, so PJSUA2 has a dialog it + owns. Avoids touching call placement, but adds a phantom call per real call + and the SDP juggling is fragile. + +> **Safety note for this refactor:** `is_emergency_number()` stays the first +> check in `gateway.make_call`, above the concurrency cap and above any SIP +> action, regardless of which library dials. A new outbound path that reaches +> the SIP layer without passing that guard is a serious regression even if +> every test passes. ### Why asyncio Queue-based EventBus? @@ -164,11 +230,13 @@ This split lets us tap into the audio stream (for classification and STT) withou - **Dead subscriber cleanup** — full queues are automatically removed - **Event history** — late joiners can catch up on recent events -If scaling to multiple gateway processes becomes necessary, the EventBus interface can be backed by Redis pub/sub without changing consumers. +If scaling to multiple gateway processes becomes necessary, the EventBus +interface can be backed by Redis pub/sub without changing consumers. ### Why OpenAI-compatible LLM API? -The LLM client uses raw HTTP (httpx) against any OpenAI-compatible endpoint. This means: +The LLM client uses raw HTTP (httpx) against any OpenAI-compatible endpoint. +This means: - **Ollama** (local, free) — `http://localhost:11434/v1` - **LM Studio** (local, free) — `http://localhost:1234/v1` @@ -176,3 +244,11 @@ The LLM client uses raw HTTP (httpx) against any OpenAI-compatible endpoint. Thi - **OpenAI** (cloud) — `https://api.openai.com/v1` No SDK dependency. No vendor lock-in. Switch models by changing one env var. + +## Testing against a fake PSTN + +`tests/lab/` runs an Asterisk instance that answers calls, plays an IVR, holds +with music and connects a "human" — so the gateway has something real to dial +that is not the PSTN. `SIP_TRUNK_HOST` is just an address, so the production +code path runs unmodified; while it points at the lab there is no route to the +PSTN at all. See [tests/lab/README.md](../tests/lab/README.md).