r/hold-slayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(media): implement PJSUA2 audio capture port; document the media-plane refactor

Browse Source 
Implements the tap half of the media path and records why the other half
requires moving call placement into PJSUA2.

MediaPipeline.create_tap was a stub: it logged "🎤 Audio tap created" and
returned a tap that nothing ever fed, so the classifier received no audio on
a live call. It now builds a real pj.AudioMediaPort subclass whose
onFrameReceived converts the SWIG ByteVector to PCM bytes and fans it out to
every tap on the stream.

One capture port per stream, shared by all taps: a second port on the same
stream would be mixed back into the conference bridge and the call would echo.

Thread safety is the constraint here. onFrameReceived runs on a PJSUA2 worker
thread — a third execution context beside the asyncio loop and the Sippy ED
thread — and touches nothing but AudioTap.feed, which hops to the owning loop
via call_soon_threadsafe. An exception escaping into PJSUA2's C++ callback
would tear down the worker thread and silently kill media for every call, so
the handler catches and logs once per port rather than on every 20ms frame.

Also fixes a hard crash found while testing this against real PJSUA2: a media
port finalised after Endpoint.libDestroy() calls pjmedia_conf_remove_port
against a freed conference bridge and aborts the process on a native
assertion. Ports are now released in remove_stream while the bridge still
exists, and stop() forces a collection before libDestroy — dropping the last
Python reference is not sufficient on its own.

Verified against the real bindings: frames fan out to multiple taps, cross the
thread boundary intact, and shutdown is clean.

add_remote_stream remains a stub, and deliberately so. 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 via pj.Call.getAudioMedia() on a dialog PJSUA2 itself owns.
A design where Sippy owns the dialog can never obtain media from PJSUA2, so
that function cannot be written against this API. docs/architecture.md now
explains this and records the resolution: PJSUA2 places the trunk call while
Sippy keeps the SBC roles (device registration, routing, leg bridging), with
the emergency guard and concurrency cap staying first in gateway.make_call
regardless of which library dials.

The architecture doc also had drift unrelated to media: it described the
thread boundary as asyncio.run_in_executor() when the real mechanism is
run_coroutine_threadsafe / ED2.callFromThread, claimed two execution contexts
where there are three, and cited MediaPipeline.add_stream() and
SippyEngine.bridge() — neither of which exists. Corrected, with the data flow
now showing the emergency guard and concurrency cap in their real positions.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Robert Helewka
2026-07-29 06:56:50 -04:00
parent c00cf02676
commit 7979e70705
2 changed files with 222 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
core/media_pipeline.py
View File

@@ -20,6 +20,7 @@ PJSUA2 runs in its own thread with a dedicated Endpoint.
""" """
import asyncio import asyncio
import gc
import logging import logging
import threading import threading
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
@@ -98,6 +99,72 @@ class AudioTap:
self._active = False 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 # Stream Entry — tracks a single media stream in the pipeline
# ================================================================ # ================================================================
@@ -112,6 +179,8 @@ class MediaStream:
self.codec = codec self.codec = codec
self.conf_port: Optional[int] = None # PJSUA2 conference bridge port ID self.conf_port: Optional[int] = None # PJSUA2 conference bridge port ID
self.transport = None # PJSUA2 SipTransport 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.rtp_port: Optional[int] = None # Local RTP listen port
self.taps: list[AudioTap] = [] self.taps: list[AudioTap] = []
self.recorder = None # PJSUA2 AudioMediaRecorder self.recorder = None # PJSUA2 AudioMediaRecorder
@@ -173,6 +242,7 @@ class MediaPipeline:
self._next_rtp_port = rtp_start_port self._next_rtp_port = rtp_start_port
self._sample_rate = sample_rate self._sample_rate = sample_rate
self._channels = channels 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) self._null_audio = null_audio # Use null audio device (no sound card needed)
# State # State
@@ -255,11 +325,17 @@ class MediaPipeline:
tap.close() tap.close()
self._taps.clear() self._taps.clear()
# Remove all streams # Remove all streams (this releases their capture ports)
for stream_id in list(self._streams.keys()): for stream_id in list(self._streams.keys()):
self.remove_stream(stream_id) 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: if self._endpoint:
try: try:
self._endpoint.libDestroy() self._endpoint.libDestroy()
@@ -350,6 +426,19 @@ class MediaPipeline:
tap.close() tap.close()
self._taps.pop(stream_id, None) 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 # Stop recording
if stream.recorder: if stream.recorder:
try: try:
@@ -426,16 +515,28 @@ class MediaPipeline:
self._taps[stream_id] = [] self._taps[stream_id] = []
self._taps[stream_id].append(tap) 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: try:
import pjsua2 as pj # One capture port per stream, shared by every tap on it:
# Create an AudioMediaPort that captures frames # the bridge would otherwise mix each additional port back
# and feeds them to the tap # into the conference and the call would echo.
# In PJSUA2, we'd subclass AudioMediaPort and implement if stream.capture_port is None:
# onFrameReceived to call tap.feed(frame_data) port = make_capture_port(
logger.info(f" 🎤 Audio tap created for {stream_id} (PJSUA2)") 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: 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: else:
logger.info(f" 🎤 Audio tap created for {stream_id} (virtual)") logger.info(f" 🎤 Audio tap created for {stream_id} (virtual)")

146
docs/architecture.md
View File

@@ -1,6 +1,15 @@
# Architecture # 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 ## System Diagram
@@ -27,8 +36,13 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
│ └────┬─────┘ └─────┬─────┘ └──────────────┘ │ │ └────┬─────┘ └─────┬─────┘ └──────────────┘ │
│ │ │ │ │ │ │ │
│ ┌────┴──────────────┴───────────────────┐ │ │ ┌────┴──────────────┴───────────────────┐ │
│ │ Sippy B2BUA Engine │ │ │ │ SIP Engine │ │
│ │ (SIP calls, DTMF, conference bridge) │ │ │ │ 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 | | Component | File | Purpose |
|-----------|------|---------| |-----------|------|---------|
| Sippy Engine | `core/sippy_engine.py` | SIP signaling (INVITE, BYE, REGISTER, DTMF) | | Sippy Engine | `core/sippy_engine.py` | SIP signalling (INVITE, BYE, REGISTER, DTMF) |
| Media Pipeline | `core/media_pipeline.py` | PJSUA2 RTP media handling, conference bridge, recording | | Media Pipeline | `core/media_pipeline.py` | PJSUA2 RTP media, conference bridge, taps, recording |
| Recording | `services/recording.py` | WAV file management and storage | | Recording | `services/recording.py` | WAV file management and storage |
| Analytics | `services/call_analytics.py` | Call metrics, hold time stats, trends | | Analytics | `services/call_analytics.py` | Call metrics, hold time stats, trends |
| Notifications | `services/notification.py` | WebSocket + SMS alerts | | 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 } POST /api/v1/calls/hold-slayer { number, intent, call_flow_id }
│ │
2. Gateway.make_call() 2. Gateway.make_call()
├── CallManager.create_call() → track state ├── is_emergency_number() → REFUSE 911/112 (before anything else)
├── SippyEngine.make_call() → SIP INVITE to trunk ├── concurrency cap check → refuse past max_concurrent_calls
└── MediaPipeline.add_stream() → RTP media setup ├── CallManager.create_call() → track state
└── sip_engine.make_call() → place the call, media follows
│ │
3. HoldSlayer.run_with_flow() or run_exploration() 3. HoldSlayer.run_with_flow() or run_exploration()
├── AudioClassifier.classify() → analyze 3s audio windows ├── 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 ├── TranscriptionService.transcribe() → STT on speech audio
│ │
├── LLMClient.analyze_ivr_menu() → pick menu option (fallback) ├── 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() └── detect_hold_to_human_transition()
└── HUMAN_DETECTED! → transfer └── HUMAN_DETECTED! → transfer
│ │
4. Transfer 4. Transfer
├── SippyEngine.bridge() → connect call legs ├── SippyEngine.bridge_calls() → join the two call legs
├── MediaPipeline.bridge_streams() → bridge RTP ├── MediaPipeline.bridge_streams() → bridge RTP in the conf bridge
├── EventBus.publish(TRANSFER_STARTED) ├── EventBus.publish(TRANSFER_STARTED)
└── NotificationService → "Pick up your phone!" └── 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 → 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 ## Threading Model
Hold Slayer is primarily single-threaded async (asyncio), with one exception: The README's "single-process async" is a simplification. There are **three**
execution contexts, and the boundaries between them are the highest-leverage
- **Main thread**: FastAPI + all async services (event bus, hold slayer, classifier, etc.) invariant in the codebase.
- **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).
``` ```
Main Thread (asyncio) asyncio loop (main thread) Sippy ED thread PJSUA2 worker threads
├── FastAPI (uvicorn) ├── FastAPI (uvicorn) └── ED2 dispatcher └── media / RTP
├── EventBus ├── EventBus ├── SIP signalling └── onFrameReceived
├── CallManager ├── CallManager ├── UA objects
├── HoldSlayer ├── HoldSlayer └── DTMF relay
├── AudioClassifier ├── AudioClassifier
├── TranscriptionService ├── TranscriptionService
├── LLMClient ├── LLMClient
├── MediaPipeline (PJSUA2)
├── NotificationService ├── NotificationService
└── RecordingService └── 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 ## 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 exposes no standalone RTP media object.** Every `AudioMedia` subclass
- **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. 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? ### 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 - **Dead subscriber cleanup** — full queues are automatically removed
- **Event history** — late joiners can catch up on recent events - **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? ### 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` - **Ollama** (local, free) — `http://localhost:11434/v1`
- **LM Studio** (local, free) — `http://localhost:1234/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` - **OpenAI** (cloud) — `https://api.openai.com/v1`
No SDK dependency. No vendor lock-in. Switch models by changing one env var. 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).