fix(lab): silence Asterisk noise at the source, keep notice logging

Follow-up to 3150f78, which over-corrected. Dropping `notice` from logger.conf
made the log quiet by making the gateway undebuggable: Asterisk reports
rejected SIP requests at notice level via log_failed_request ("No matching
endpoint found", "Failed to authenticate"), and on a box whose entire job is
answering SIP those are the most useful lines it produces. A call was refused
and nothing said so.

`notice` is restored. The noise is dealt with where it originates instead:

- modules.conf, new. The image ships `autoload=yes` and loads every module it
  was built with, including chan_alsa on a container with no sound card:
  ~74 ALSA config errors per restart, plus module-load ERRORs for CDR/CEL
  backends, LDAP/ODBC realtime config, and format_ogg_vorbis — none of which
  can work here. Explicit noload for those; autoload stays on, because an
  allow-list would break quietly the first time a scenario needs a module
  nobody remembered to add.

- The healthcheck reduction from 3150f78 (one CLI connection on a 60s
  interval, rather than the image's ~7 every 30s) does the rest.

Verified on galatea: ALSA lines 74 → 0, all startup ERRORs gone (only one-off
benign WARNINGs remain), steady-state 4 lines per 2 minutes against ~840 per
30 minutes originally — and a refused call still logs 6 diagnostic lines.
Transport bound, both endpoints and dialplan loaded, container healthy.

`verbose` stays excluded: dialplan execution is worth having when tracing a
specific call, not worth shipping to Loki continuously. Raise it at runtime
with `asterisk -rx "core set verbose 3"`.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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 = ""
 

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,

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
@@ -143,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)
@@ -173,6 +243,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 +326,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()
@@ -270,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
@@ -291,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:
@@ -350,6 +437,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 +526,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)")
 

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()

core/sippy_engine.py

@@ -84,6 +84,48 @@ class SippyCallController:
         self.leg_id = leg_id
         self.engine = engine
 
+    def __call__(self, event, ua) -> None:
+        """Sippy's ``event_cb`` — invoked as ``event_cb(event, ua)``.
+
+        Sippy delivers call progress as CCEvent objects through this one
+        entry point; it never calls the ``on_*`` methods directly. This
+        dispatches to them so each SIP fact still has a named handler.
+        """
+        from sippy.CCEvents import (
+            CCEventConnect,
+            CCEventDisconnect,
+            CCEventFail,
+            CCEventPreConnect,
+            CCEventRing,
+        )
+
+        try:
+            if isinstance(event, CCEventRing):
+                self.on_ringing()
+            elif isinstance(event, (CCEventConnect, CCEventPreConnect)):
+                # data is (code, reason, body) — the body carries the
+                # negotiated SDP that tells the media pipeline where to
+                # send RTP.
+                data = event.getData()
+                body = data[2] if isinstance(data, tuple) and len(data) > 2 else None
+                self.on_connected(str(body) if body is not None else None)
+            elif isinstance(event, CCEventDisconnect):
+                self.on_disconnected("remote hangup")
+            elif isinstance(event, CCEventFail):
+                data = event.getData()
+                reason = " ".join(str(d) for d in data[:2]) if data else "call failed"
+                self.on_disconnected(reason)
+            # DTMF is not handled here: SIP INFO arrives as a request and is
+            # picked up by _handle_incoming_info, and RFC 2833 DTMF rides in
+            # the RTP stream, which is the media pipeline's business.
+        except Exception as e:
+            # This runs on the Sippy ED thread: an escaping exception is
+            # swallowed by the dispatcher and the leg would hang silently.
+            logger.error(
+                f"  {self.leg_id}: error handling {type(event).__name__}: {e}",
+                exc_info=True,
+            )
+
     def on_trying(self):
         """100 Trying received."""
         logger.debug(f"  {self.leg_id}: 100 Trying")
@@ -234,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:
@@ -319,10 +366,17 @@ class SippyEngine(SIPEngine):
             SipConf.my_port = self._sip_port
             SipConf.my_uaname = "Hold Slayer Gateway"
 
+            # SipTransactionManager dereferences _sip_logger unconditionally on
+            # every message, so it must exist before any SIP traffic. It
+            # defaults to the stderr backend (SIPLOG_BEND), not the
+            # /var/log/sip.log path in its signature — nothing to create.
+            from sippy.SipLogger import SipLogger
+
             self._sippy_global_config = {
                 "_sip_address": self._sip_address,
                 "_sip_port": self._sip_port,
                 "_sip_tm": None,  # Transaction manager set after start
+                "_sip_logger": SipLogger("hold-slayer"),
             }
 
             # Start Sippy's SIP transaction manager in a background thread
@@ -499,23 +553,53 @@ class SippyEngine(SIPEngine):
         def do_register():
             try:
                 from sippy.SipRegistrationAgent import SipRegistrationAgent
+                from sippy.SipURL import SipURL
 
+                def on_registered(_rtime, _contact, _cb_arg):
+                    logger.info("  ✅ Trunk registration accepted")
+                    self._post_from_ed("trunk_registered", {"registered": True})
+
+                def on_register_failed(status_line, _cb_arg):
+                    # status_line is the response's status line (e.g. "403
+                    # Forbidden") — surface it; a bad trunk password is the
+                    # most common cause and is otherwise invisible.
+                    logger.error(f"  ❌ Trunk registration rejected: {status_line}")
+                    self._post_from_ed(
+                        "trunk_registered",
+                        {"registered": False, "reason": str(status_line)},
+                    )
+
+                # A wildcard bind is not a routable Contact — the trunk would
+                # have nowhere to send the inbound INVITE. Fall back to
+                # loopback, matching _generate_sdp's handling.
+                contact_host = (
+                    self._sip_address if self._sip_address != "0.0.0.0" else "127.0.0.1"
+                )
+
+                # aor/contact must be SipURL objects: the agent calls
+                # .getCopy() and mutates .username/.port on them.
                 reg_agent = SipRegistrationAgent(
                     self._sippy_global_config,
-                    f"sip:{self._trunk_username}@{self._trunk_host}",
-                    f"sip:{self._trunk_host}:{self._trunk_port}",
-                    auth_name=self._trunk_username,
-                    auth_password=self._trunk_password,
+                    SipURL(f"sip:{self._trunk_username}@{self._trunk_host}"),
+                    SipURL(f"sip:{self._trunk_username}@{contact_host}:{self._sip_port}"),
+                    user=self._trunk_username,
+                    passw=self._trunk_password,
+                    rok_cb=on_registered,
+                    rfail_cb=on_register_failed,
                 )
-                reg_agent.register()
-                logger.info("  ✅ Trunk registration sent")
-                self._post_from_ed("trunk_registered", {"registered": True})
+                # Registration is asynchronous: success is reported by the
+                # callbacks above, not here. Reporting "registered" at send
+                # time would let /health go green on a rejected REGISTER.
+                reg_agent.doregister()
+                logger.info("  Trunk REGISTER sent, awaiting response")
             except ImportError:
                 logger.warning("  Sippy registration agent not available")
                 self._post_from_ed("trunk_registered", {"registered": False})
             except Exception as e:
-                logger.error(f"  ❌ Trunk registration failed: {e}")
-                self._post_from_ed("trunk_registered", {"registered": False})
+                logger.error(f"  ❌ Trunk registration failed: {e}", exc_info=True)
+                self._post_from_ed(
+                    "trunk_registered", {"registered": False, "reason": str(e)}
+                )
 
         self._run_on_sippy(do_register)
 
@@ -570,7 +654,8 @@ class SippyEngine(SIPEngine):
         else:
             remote_uri = f"sip:{number}@{self._domain}"
 
-        from_uri = f"sip:{caller_id or self._did}@{self._domain}"
+        caller_number = caller_id or self._did
+        from_uri = f"sip:{caller_number}@{self._domain}"
 
         leg = SipCallLeg(leg_id, "outbound", remote_uri)
         self._legs[leg_id] = leg
@@ -583,24 +668,38 @@ class SippyEngine(SIPEngine):
         def do_invite():
             try:
                 from sippy.CCEvents import CCEventTry
+                from sippy.MsgBody import MsgBody
                 from sippy.SipCallId import SipCallId
                 from sippy.UA import UA
 
                 controller = SippyCallController(leg_id, self)
 
-                # Create Sippy UA for this call
+                # Create Sippy UA for this call. The credentials are required:
+                # a trunk answers the first INVITE with 401/407, and sippy
+                # only retries with a digest response when they are set —
+                # without them every outbound call dies on the challenge.
                 ua = UA(
                     self._sippy_global_config,
                     event_cb=controller,
+                    username=self._trunk_username or None,
+                    password=self._trunk_password or None,
                     nh_address=(self._trunk_host, self._trunk_port),
                 )
                 self._ed_leg_to_ua[leg_id] = ua
                 self._ed_ua_to_leg[ua] = leg_id
 
-                # Send INVITE
+                # SDP travels inside the event's data tuple as a MsgBody, not
+                # as a kwarg. needs_update=False marks it final: with it set,
+                # sippy would call ua.on_local_sdp_change (unset here) before
+                # sending, and the INVITE would never go out.
+                body = MsgBody(sdp_body, mtype="application/sdp")
+                body.needs_update = False
+
+                # UacStateIdle unpacks exactly six fields and builds the SIP
+                # URIs itself from nh_address — callingID/calledID are bare
+                # usernames, not full URIs.
                 event = CCEventTry(
-                    (SipCallId(), from_uri, remote_uri),
-                    body=sdp_body,
+                    (SipCallId(), caller_number, number, body, None, None)
                 )
                 ua.recvEvent(event)
 
@@ -613,8 +712,11 @@ class SippyEngine(SIPEngine):
                 self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "ringing"})
 
             except Exception as e:
-                logger.error(f"  Failed to send INVITE for {leg_id}: {e}")
-                self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "terminated"})
+                logger.error(f"  Failed to send INVITE for {leg_id}: {e}", exc_info=True)
+                self._post_from_ed(
+                    "leg_state",
+                    {"leg_id": leg_id, "state": "terminated", "error": str(e)},
+                )
 
         self._run_on_sippy(do_invite)
         return leg_id

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).

docs/asterisk-lab-design.md

@@ -0,0 +1,293 @@
+# Asterisk Lab — design
+
+A **fake PSTN** for Hold Slayer: an Asterisk instance in Virgo Dev that answers
+calls, plays an IVR, holds you in a queue with music, and eventually connects a
+"human". It gives the gateway something real to dial that is not the PSTN — no
+charges, no strangers, no E911 exposure, and a *deterministic* script that makes
+classifier regressions reproducible.
+
+Status: **design, not built.** Nothing here has been deployed.
+
+---
+
+## Why Asterisk and not Kamailio
+
+Kamailio is a SIP **proxy** — it routes signaling and does not answer calls or
+handle media. The risks that remain unproven in Hold Slayer are mostly *media*
+risks: `send_dtmf` has only ever run against `MockSIPEngine` (a no-op), the
+audio classifier has never seen real RTP, and the PJSUA2 pipeline built in Phase
+1b has never carried a packet. A proxy forwards the INVITE and finds nobody
+home, so it exercises none of that.
+
+Asterisk is a B2BUA: it answers, plays prompts, collects DTMF, and can hold a
+call in a queue with music. That is precisely the hold-slayer scenario, so it is
+the right primary target.
+
+Kamailio still has a place — **later, and narrowly**. It models a real ITSP's
+registration/digest-auth behaviour better than Asterisk does, so it is the right
+tool for exercising `_register_trunk()` ([core/sippy_engine.py:495](../core/sippy_engine.py#L495))
+in isolation. It is deliberately *not* in scope for this lab.
+
+```
+Phase 2/3 (this lab)          Phase 4a (optional)        Phase 4b
+Asterisk IVR + media    →     Kamailio registration  →   real PSTN, one call
+```
+
+---
+
+## The one thing that makes this work without code changes
+
+`SIP_TRUNK_HOST` is just an address. `SippyEngine.make_call()` builds
+`sip:{number}@{trunk_host}:{trunk_port}`
+([core/sippy_engine.py:568](../core/sippy_engine.py#L568)) and registers against
+whatever host it is given. Point it at Asterisk and Hold Slayer dials it exactly
+as it would dial a real provider.
+
+**There is no test-only branch, no mock, no `if lab:` anywhere.** The code path
+under test is the production code path. That is the entire value of this
+approach — a lab that requires special-casing the application proves less than
+it costs.
+
+It also means the safety story is structural: while `SIP_TRUNK_HOST` points at
+Asterisk on the Dev LAN, there is **no route to the PSTN at all**. Not a policy
+that could be misconfigured — an absence of route.
+
+---
+
+## Placement in Virgo
+
+| Decision | Value | Why |
+|---|---|---|
+| Host | **nereid** (`10.0.1.214`) | Terraform describes it as "Experimental Apps (POC, testing new technologies)" — [terraform/incus/containers.tf](../../virgo/terraform/incus/containers.tf). An unauthenticated SIP endpoint is exactly experimental. |
+| Deploy | Docker Compose via Ansible | Matches every other Virgo service. `nereid` already has `docker = true`. |
+| Hostname | `asterisk.helu.ca` (internal) | LAN only. **No `*.d.helu.ca` HAProxy entry** — HAProxy is HTTP; SIP/RTP would not traverse it, and this must not be publicly reachable. |
+| Database | none | Asterisk needs no DB, so the "no databases in Docker" rule is not engaged. |
+
+Hold Slayer itself stays on **triton** (`10.0.1.213`), where it is already
+deployed at port 21081. Splitting the two hosts is deliberate: SIP then crosses
+a real network with real latency, jitter and MTU, rather than a loopback that
+hides every transport problem.
+
+### Ports
+
+Project **210** is Hold Slayer's (existing: `hold_slayer_web_port: 21081`).
+Verified free: only `21081` and `29181` are allocated in that space today.
+
+| Var | Port | Purpose |
+|---|---|---|
+| `asterisk_sip_port` | **21061** | SIP signalling (UDP). `2-10-6-1`: project 210, service 6 (SIP), instance 1 |
+| `asterisk_rtp_start` | **21100** | RTP media range start (UDP) |
+| `asterisk_rtp_end` | **21149** | RTP range end — 50 ports ≈ 25 concurrent calls, well above `max_concurrent_calls: 4` |
+| `asterisk_ari_port` | **21071** | ARI/HTTP management (service 7 = management) |
+| `asterisk_syslog_port` | **51462** | Docker syslog, `514YZ` convention (daedalus uses 51461) |
+
+Service digit `6` for SIP is a **new allocation** — the existing scheme
+([docs/virgo.md](../../virgo/docs/virgo.md) §Port Numbering) defines 1/2/5/7/8/9
+and has no telephony digit. Worth confirming before it becomes precedent.
+
+> **Note:** 5060 is *not* used. The convention forbids random ports, and using
+> the well-known SIP port invites scanner traffic. Nothing requires 5060 — both
+> ends are configured.
+
+---
+
+## Call scenarios
+
+Each maps to a Hold Slayer behaviour that is currently unproven. Extensions are
+what Hold Slayer dials as `number`.
+
+| Ext | Scenario | Proves |
+|---|---|---|
+| `1001` | **Immediate answer**, plays speech, hangs up after 30s | Baseline: INVITE→200→ACK→RTP→BYE, audio flows both ways, classifier reports `LIVE_HUMAN` |
+| `1002` | **IVR menu** — "press 1 for accounts, 2 for cards", branches on DTMF | `send_dtmf` genuinely emits RFC 2833 and Asterisk receives it. This is the big one — currently a no-op in the mock |
+| `1003` | **Hold music, then human** — 60s MoH, then answers | The whole hold-slayer loop: classify music → stay on hold → detect human → ring the owner |
+| `1004` | **Long hold** — 10 min MoH | `MAX_HOLD_TIME` and the hold-check interval |
+| `1005` | **Immediate busy** (`BUSY()`) | Failure path: call marked `FAILED`, no stuck leg |
+| `1006` | **Ring, never answer** | Timeout path |
+| `1007` | **Answer then hang up after 5s** | Remote-BYE handling, DB persistence on hangup |
+| `1008` | **Silence after answer** | Classifier `SILENCE` vs the 30s no-audio case |
+
+`1002` and `1003` are the two that matter most; the rest are cheap to add once
+the dialplan exists.
+
+### Classifier determinism
+
+The reason for a scripted IVR rather than a real call: real hold music varies
+per call, so a classifier regression on the PSTN is indistinguishable from
+noise. Against a fixed prompt the answer is binary. Recommend a **fixed MoH
+file committed to the repo** rather than Asterisk's stock music, so the
+classifier's input is byte-identical on every run and across hosts.
+
+This is what makes the Phase 0 music-vs-speech precedence fix
+([services/audio_classifier.py](../services/audio_classifier.py)) testable
+against real audio for the first time.
+
+---
+
+## Security — the part that needs a decision
+
+Two unauthenticated SIP endpoints would exist on the Dev LAN.
+
+**1. Asterisk.** Default `pjsip.conf` examples accept anonymous calls. This lab
+must not: `allow_anonymous_inbound = no`, an explicit endpoint for Hold Slayer
+with a password, and `permit=` limited to triton's address. Asterisk's default
+config is a well-known toll-fraud target and must not be shipped as-is.
+
+**2. Hold Slayer's own SIP listener — pre-existing, flagged earlier.**
+`_handle_incoming_register()` replies `200 OK` to any REGISTER with **no digest
+challenge**. On loopback that was tolerable. The moment the gateway binds a LAN
+interface to talk to Asterisk, any host on the Dev LAN can register as a device
+and receive transferred calls.
+
+That is not caused by this lab, but this lab is what makes it reachable. Options,
+in order of preference:
+
+1. **Implement digest auth** on inbound REGISTER — the real fix
+2. **Bind the SIP listener to a specific interface** and firewall 5060 to triton
+   ↔ nereid only — mitigation, not a fix
+3. Accept it explicitly, in writing, as Dev-only
+
+I would not deploy this without at least (2), and (1) is required before
+anything resembling production. **This needs your decision before build.**
+
+Also note: `pjsua2` runs `--enable-shared` with TLS available, so SIP-TLS +
+SRTP is possible later. Not proposed for the lab — plain UDP keeps `sngrep`
+readable, which matters enormously when debugging signalling.
+
+---
+
+## Observability
+
+Match the estate rather than inventing:
+
+- **Logs** — default `json-file` driver, discovered by the host Alloy Docker
+  socket source and labelled `job=asterisk`. **No syslog listener and no Loki
+  URL env** — that would double-ship, the same note already carried in
+  [hold-slayer's compose template](../../virgo/ansible/hold-slayer/docker-compose.yml.j2).
+- **Health** — Asterisk has no HTTP health endpoint by default. Enable ARI on
+  `21071` and probe `/ari/asterisk/info`, which is a genuine liveness signal
+  (the SIP stack answers), unlike a bare TCP check.
+- **`sngrep`** on nereid for live SIP ladder inspection. Not currently installed
+  anywhere; it is the single most useful tool when signalling misbehaves.
+
+---
+
+## What this does *not* prove
+
+Stated plainly so the lab is not over-trusted:
+
+- **Not real PSTN audio.** No G.711 transcoding artefacts, no packet loss, no
+  jitter, no carrier-side DTMF mangling. Asterisk is clean; the PSTN is not.
+- **Not real IVR behaviour.** Our dialplan is what we imagine a bank sounds
+  like. Real trees are longer, noisier, and interrupt.
+- **Not trunk registration/auth** — that is Kamailio's job (Phase 4a), or the
+  real trunk's.
+- **Not carrier-specific quirks** — each ITSP has its own.
+
+It proves the gateway's *own* logic end to end. That is the majority of the
+risk, and it is the part that is currently entirely untested against real media.
+
+---
+
+## Blocking prerequisite — the container runs stub media
+
+**The Hold Slayer Docker image deliberately does not build PJSUA2**
+([docs/pjsua2-build.md](pjsua2-build.md)), so the deployed container on triton
+runs the media pipeline in **stub mode**. Stub mode's audio calls *return
+successfully while doing nothing*.
+
+If the lab runs against the current image, every media test passes while proving
+nothing. This is the single most dangerous failure mode in the plan, because it
+looks like success.
+
+Two options:
+
+| Option | Effort | Trade-off |
+|---|---|---|
+| **A. Add a pjproject build stage to the Dockerfile** | Higher — multi-stage build, ~10 min build, larger image | The deployed artefact gains real media. Needed eventually regardless |
+| **B. Run Hold Slayer from a venv on triton for the lab** | Lower — the Phase 1b build already exists on caliban | Tests the binaries actually built, but diverges from the deployed artefact |
+
+**Recommendation: A.** B tests something that is not what ships, and the
+Dockerfile needs this anyway before Hold Slayer can place a real call from a
+container. Doing it now means the lab validates the real artefact. B is a
+reasonable short-cut only if you want a fast first signal.
+
+The existing deploy also has a **stale-config finding** — see below — that
+touches the same file, so both are worth doing in one pass.
+
+---
+
+## Finding: the deployed compose template is stale
+
+Independent of this lab, [the deployed template](../../virgo/ansible/hold-slayer/docker-compose.yml.j2)
+sets `API_TOKEN`, which **no longer exists** — auth is now Casdoor SSO + PATs
+via one resolver. The comment "the app's single static bearer across REST/WS/MCP
+… required on 0.0.0.0" describes an auth model that was removed.
+
+Live state confirms the service is up and `degraded`/`engine: mock` (correct and
+honest). Given `_check_startup_config` refuses SSO-off on a non-loopback bind, it
+is worth establishing how it is currently booting — most likely `CASDOOR_ENABLED`
+defaults such that the unknown `API_TOKEN` is simply ignored.
+
+Flagging, not fixing — it is outside this design, but it lives in the file the
+lab will modify, and `hold_slayer_api_token` is still being pulled from the OCI
+vault for a variable the app no longer reads.
+
+---
+
+## Build order
+
+Each step is independently verifiable; none commits you to the next.
+
+1. **Decide** the two open questions: media (A or B), and SIP-listener security
+   (digest / firewall / accept)
+2. **Dialplan + compose**, developed on caliban against a local Asterisk
+   container — no Virgo changes yet, fastest iteration
+3. **Prove `1001`** locally: Hold Slayer places a call, audio flows, classifier
+   sees `LIVE_HUMAN`. This is the real Phase 2 gate
+4. **Prove `1002`/`1003`** locally: DTMF lands, hold→human transition fires
+5. **Promote to Virgo** — Ansible role on nereid, `SIP_TRUNK_HOST=nereid.helu.ca`
+   on triton, re-run 1–8 across the LAN
+6. **Only then** consider Kamailio (4a) or the PSTN (4b)
+
+Steps 2–4 need no Virgo changes at all, which is worth exploiting: the dialplan
+is where the fiddly work is, and iterating locally is far faster than through
+Ansible.
+
+---
+
+## Files this would add
+
+```
+hold-slayer/
+  tests/lab/
+    dialplan/extensions.conf     # the 8 scenarios
+    dialplan/pjsip.conf          # endpoint for Hold Slayer, anonymous denied
+    sounds/hold-music.wav        # fixed MoH — deterministic classifier input
+    docker-compose.lab.yml       # local Asterisk for steps 2–4
+    README.md                    # how to run the lab locally
+
+virgo/
+  ansible/asterisk/
+    deploy.yml                   # mirrors ansible/hold-slayer/deploy.yml
+    docker-compose.yml.j2
+    extensions.conf.j2
+    pjsip.conf.j2
+  ansible/inventory/host_vars/nereid.helu.ca.yml   # + asterisk_* vars, + service
+```
+
+Dialplan lives in **hold-slayer**, not virgo: it is test fixture data that
+belongs with the code it tests, and step 2–4 iteration needs it locally.
+Ansible templates it out to nereid for step 5.
+
+---
+
+## Open questions
+
+1. **Media: A or B?** Determines whether step 5 tests the real artefact.
+2. **SIP listener security** — digest auth, firewall, or documented acceptance?
+   Blocking for step 5, not for steps 2–4.
+3. **Is service digit `6` acceptable for SIP** in the 22XYZ scheme, or should
+   telephony get a different digit? Sets estate precedent.
+4. **`asterisk.helu.ca` DNS** — needs an entry, or is `nereid.helu.ca` on the
+   allocated port sufficient? (Simpler, and one less thing to maintain.)

tests/lab/README.md

@@ -0,0 +1,213 @@
+# Asterisk lab — a fake PSTN
+
+An Asterisk instance that answers calls, plays an IVR, holds you with music,
+and eventually connects a "human". It gives the gateway something real to dial
+that is **not** the PSTN: no charges, no strangers, no E911 exposure, and a
+deterministic script that makes classifier regressions reproducible.
+
+Design rationale and the Virgo deployment plan:
+[docs/asterisk-lab-design.md](../../docs/asterisk-lab-design.md).
+
+> This lab found five bugs in `SippyEngine` on its first call — the engine had
+> never successfully placed one. Everything below runs against the real
+> `SippyEngine`, never `MockSIPEngine`, which is the entire point.
+
+---
+
+## Run it
+
+```bash
+cd tests/lab
+
+# 1. Generate the audio fixtures (the image ships with NO sound files).
+python sounds/generate.py
+
+# 2. Render the local configs. They carry a host-specific IP and the lab
+#    password, so they are gitignored — regenerate them per machine.
+cd dialplan
+LOCALIP=$(ip route get 1.1.1.1 | grep -oP '(?<=src\s)\d+(\.\d+){3}')
+sed -e "s/{{ asterisk_sip_port }}/21061/" \
+    -e "s/{{ asterisk_external_ip }}/$LOCALIP/" \
+    -e "s#{{ asterisk_local_net }}#10.10.0.0/24#" \
+    -e "s/{{ asterisk_match_host }}/127.0.0.1/" \
+    -e "s/{{ asterisk_sip_username }}/holdslayer/" \
+    -e "s/{{ asterisk_sip_password }}/labpassword/" \
+    pjsip.conf > pjsip.local.conf
+sed -e "s/{{ asterisk_rtp_start }}/21100/" \
+    -e "s/{{ asterisk_rtp_end }}/21149/" \
+    rtp.conf > rtp.local.conf
+cd ..
+
+# 3. Start it.
+docker compose -f docker-compose.lab.yml up -d
+```
+
+Point Hold Slayer at it — no code changes, no test-only branch. `make_call`
+builds `sip:{number}@{trunk_host}:{trunk_port}`, so the lab is just an address:
+
+```bash
+USE_MOCK_SIP=false
+SIP_TRUNK_HOST=127.0.0.1
+SIP_TRUNK_PORT=21061
+SIP_TRUNK_USERNAME=holdslayer
+SIP_TRUNK_PASSWORD=labpassword
+SIP_TRUNK_DID=+15550000000
+GATEWAY_SIP_PORT=21062     # must differ from the Asterisk port
+```
+
+> **The repo's own `.env` sets `USE_MOCK_SIP=true`** and a placeholder trunk
+> host, and pydantic-settings lets `.env` win over the process environment. If
+> the engine reports `MockSIPEngine` despite the above, that is why.
+>
+> `AIPSTNGateway(settings=...)` also defaults to `MockSIPEngine` unless an
+> engine is assigned — `main.py`'s lifespan calls `build_sip_engine()` after
+> construction. A harness that skips that step silently tests the mock.
+
+## The softphone (transfer target)
+
+**Asterisk is the registrar for devices, not Hold Slayer.** The gateway reaches
+a desk phone by dialling extension `2001`, which Asterisk routes to whatever
+has registered as `softphone`. This deliberately avoids Hold Slayer's own SIP
+listener, which answers `200 OK` to any REGISTER with no digest challenge.
+
+The `pjsua` CLI built alongside the Python bindings is the test device — same
+library stack as the gateway, no extra dependency. It needs an RPATH patch like
+the bindings did:
+
+```bash
+cp ~/src/pjproject/pjsip-apps/bin/pjsua-x86_64-pc-linux-gnu ~/.local/bin/pjsua
+patchelf --set-rpath $HOME/.local/lib ~/.local/bin/pjsua
+```
+
+Register it (config file avoids shell-quoting pain):
+
+```bash
+cat > softphone.cfg <<'EOF'
+--null-audio
+--auto-answer=200
+--max-calls=4
+--local-port=21070
+--id=sip:softphone@127.0.0.1
+--registrar=sip:127.0.0.1:21061
+--realm=asterisk
+--username=softphone
+--password=labphone
+--log-level=3
+EOF
+
+# pjsua is an interactive console app: it exits ~8s after start if stdin is
+# closed or /dev/null. Hold a fifo open on stdin — `script -qfc` and
+# `setsid </dev/null` both look like they work (registration succeeds) and
+# then the process dies, leaving a stale contact in Asterisk that routes
+# INVITEs to a port nobody is listening on.
+mkfifo sp.fifo
+setsid sh -c 'exec 3<>sp.fifo; pjsua --config-file softphone.cfg <&3 >softphone.log 2>&1' &
+```
+
+Verify — **check the port is actually bound**, not just that Asterisk holds a
+contact, since a stale registration outlives the process:
+
+```bash
+ss -lnup | grep 21070                                     # must be listening
+docker compose -f docker-compose.lab.yml exec asterisk \
+    asterisk -rx "pjsip show contacts"                    # must show softphone
+```
+
+Then place a call to `2001`. Both legs should show `Up` under one bridge id:
+
+```bash
+docker compose -f docker-compose.lab.yml exec asterisk \
+    asterisk -rx "core show channels concise"
+```
+
+> **`--realm=asterisk`, not `--realm='*'`** — the wildcard fails with
+> `PJSIP_EFAILEDCREDENTIAL` against Asterisk's digest challenge.
+
+> **Qualify is off** for this AOR (`qualify_frequency = 0`): the pjsua console
+> does not answer `OPTIONS`, so polling marks a working softphone `Unavail` and
+> the dialplan refuses to ring it. The `2001` guard therefore tests
+> `PJSIP_AOR(softphone,contact)` rather than `DEVICE_STATE`. A real hardphone
+> answers OPTIONS and can have qualify re-enabled.
+
+## Useful commands
+
+```bash
+docker compose -f docker-compose.lab.yml exec asterisk asterisk -rvvv   # CLI
+docker compose -f docker-compose.lab.yml logs -f asterisk               # logs
+docker compose -f docker-compose.lab.yml exec asterisk \
+    asterisk -rx "pjsip set logger on"                                  # SIP trace
+```
+
+---
+
+## Scenarios
+
+Hold Slayer dials these as `number`.
+
+| Ext | Scenario | Proves |
+|---|---|---|
+| `1001` | Answers, speech, hangs up | Baseline: INVITE→200→ACK→RTP→BYE, audio both ways |
+| `1002` | IVR menu, branches on DTMF | `send_dtmf` really emits RFC 2833 and Asterisk receives it |
+| `1003` | Hold music, then a human | The hold-slayer loop: music → wait → human → ring owner |
+| `1004` | Long hold (~10 min) | `MAX_HOLD_TIME`, `HOLD_CHECK_INTERVAL` |
+| `1005` | Busy | Failure path: call marked `FAILED`, no stuck leg |
+| `1006` | Rings, never answers | Timeout path |
+| `1007` | Answers, hangs up after 5s | Remote BYE, DB persistence on hangup |
+| `1008` | Answers, then silence | Classifier `SILENCE` vs. no-audio |
+| `1099` | Echo test | Debugging aid — confirm bidirectional RTP by ear |
+
+## Audio fixtures
+
+The Asterisk image ships **no sound files**, and the design calls for
+deterministic audio: real hold music varies per call, so a classifier
+regression on the PSTN is indistinguishable from noise. `sounds/generate.py`
+synthesises three fixtures from fixed seeds — byte-identical every run.
+
+Verified against `AudioClassifier` (16 kHz):
+
+| Fixture | Classifies as | Confidence |
+|---|---|---|
+| `lab-music.sln` | `MUSIC` | 0.85 |
+| `lab-speech.sln` | `LIVE_HUMAN` | 0.75 |
+| `lab-silence.sln` | `SILENCE` | 1.00 |
+
+Format is 8 kHz 16-bit mono signed-linear (`.sln`) — Asterisk's native
+telephony rate, played without transcoding.
+
+> The speech fixture's formants deliberately avoid the DTMF bands (rows
+> 697–941 Hz, columns 1209–1633 Hz). The first version landed on a valid
+> DTMF pair and the whole utterance classified as a keypress.
+
+---
+
+## Known limits
+
+- **The classifier receives nothing on a live call.**
+  `MediaPipeline.create_tap` is a stub — it logs `🎤 Audio tap created` and
+  returns a tap that is never fed (`core/media_pipeline.py`, and the same at
+  stream creation). RTP flows and Asterisk plays audio, but nothing reaches
+  the classifier. The table above was measured by feeding the fixtures
+  directly. **This blocks the hold-slayer scenarios (1002/1003/1004).**
+- **Not real PSTN audio** — no transcoding artefacts, packet loss, jitter, or
+  carrier-side DTMF mangling. Asterisk is clean; the PSTN is not.
+- **Not real IVR behaviour** — this dialplan is what we imagine a bank sounds
+  like. Real trees are longer, noisier, and interrupt.
+- **Not trunk registration against a real ITSP** — `_register_trunk()` works
+  against Asterisk, but carrier quirks are their own phase.
+
+## Security
+
+`pjsip.conf` refuses anonymous inbound calls: every call must authenticate as
+the `hold-slayer` endpoint. Asterisk's stock examples allow anonymous calls and
+are a well-known toll-fraud target — there is no PSTN behind this instance, so
+an unauthorised call reaches only the dialplan, but the lock-down keeps this
+config safe to copy.
+
+Endpoint matching is by **source address** (`type=identify`). Asterisk's
+default matches the From-header domain, which Hold Slayer populates from its
+SIP bind address — `0.0.0.0` on a wildcard bind, which never matches.
+
+> **Separate, pre-existing:** Hold Slayer's own SIP listener answers `200 OK`
+> to any REGISTER with no digest challenge. Fine on loopback; it must be
+> resolved before the gateway binds a LAN interface, or any host on the
+> network can register as a device and receive transferred calls.

tests/lab/dialplan/.gitignore

@@ -0,0 +1,3 @@
+# Rendered from the .conf templates by the local-lab instructions in
+# README.md; contains a host-specific IP and the lab password.
+*.local.conf

tests/lab/dialplan/asterisk.conf

@@ -0,0 +1,17 @@
+; Minimal Asterisk core config for the lab.
+;
+; Deliberately does NOT set [directories] or runuser/rungroup: the image's
+; compiled-in defaults are correct, and it runs as the `asterisk` user via a
+; USER directive. Overriding either risks breaking the container for no gain
+; (an earlier version of this file did both).
+[options]
+; Log to stdout so Docker's json-file driver captures it and Alloy ships it
+; to Loki. A file-based log inside the container would be invisible.
+verbose = 3
+debug = 0
+; No ANSI colour. Asterisk colourises the console by default and the escape
+; codes travel through Docker into Loki, where every line arrives wrapped in
+; \x1b[0;30m — unreadable in Grafana and awkward to filter on. This setting
+; lives here, not in logger.conf, and only takes effect if this file is
+; actually mounted into the container.
+nocolor = yes

tests/lab/dialplan/extensions.conf

@@ -0,0 +1,161 @@
+; ---------------------------------------------------------------------------
+; Hold Slayer lab dialplan — a fake bank phone tree
+; ---------------------------------------------------------------------------
+; Each extension is one scenario Hold Slayer must handle. Everything here is
+; deterministic on purpose: real hold music varies per call, so a classifier
+; regression on the PSTN is indistinguishable from noise. Against a fixed
+; prompt the answer is binary.
+;
+; Hold Slayer dials these as `number` with SIP_TRUNK_HOST pointing here.
+; ---------------------------------------------------------------------------
+
+[globals]
+; Lab-generated audio (tests/lab/sounds/, built by generate.py). The Asterisk
+; image ships with no sounds at all, and these are synthesised from a fixed
+; seed so the classifier sees byte-identical input on every run.
+;   lab-speech  -> must classify LIVE_HUMAN
+;   lab-music   -> must classify MUSIC
+;   lab-silence -> must classify SILENCE
+GREETING=lab-speech
+INVALID=lab-speech
+
+[hold-slayer-lab]
+
+; --- 1001: immediate answer, speech, hangup -------------------------------
+; Baseline. Proves INVITE→200→ACK→RTP→BYE and that audio flows both ways.
+; The classifier should report LIVE_HUMAN throughout.
+exten => 1001,1,NoOp(LAB 1001: immediate answer)
+ same => n,Answer()
+ same => n,Wait(1)
+ same => n,Playback(${GREETING})
+ same => n,Playback(lab-speech)
+ same => n,Wait(20)
+ same => n,Playback(lab-speech)
+ same => n,Hangup()
+
+; --- 1002: IVR menu, branches on DTMF -------------------------------------
+; THE important one. Proves send_dtmf genuinely emits RFC 2833 and that
+; Asterisk receives the digits — currently a no-op in MockSIPEngine.
+; Press 1 → accounts (answers as human). Press 2 → cards (hold, then human).
+exten => 1002,1,NoOp(LAB 1002: IVR menu)
+ same => n,Answer()
+ same => n,Wait(1)
+ same => n,Set(TRIES=0)
+ same => n(menu),Background(lab-speech)
+ same => n,WaitExten(8)
+ same => n,Set(TRIES=$[${TRIES} + 1])
+ same => n,GotoIf($[${TRIES} < 3]?menu)
+ same => n,Playback(lab-speech)
+ same => n,Hangup()
+
+; Option 1 — straight to a "human"
+exten => 1,1,NoOp(LAB 1002: caller pressed 1 -> accounts)
+ same => n,Playback(lab-speech)
+ same => n,Playback(lab-speech)
+ same => n,Wait(15)
+ same => n,Hangup()
+
+; Option 2 — hold queue, then a "human"
+exten => 2,1,NoOp(LAB 1002: caller pressed 2 -> cards, hold)
+ same => n,Playback(lab-speech)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-speech)
+ same => n,Wait(15)
+ same => n,Hangup()
+
+exten => i,1,NoOp(LAB 1002: invalid entry)
+ same => n,Playback(${INVALID})
+ same => n,Goto(1002,menu)
+
+exten => t,1,NoOp(LAB 1002: entry timeout)
+ same => n,Goto(1002,menu)
+
+; --- 1003: hold music, then a human ---------------------------------------
+; The whole hold-slayer loop in one call: classify music → stay on hold →
+; detect the human → ring the owner. 60s of MoH is long enough for several
+; classifier windows (CLASSIFIER_WINDOW_SECONDS defaults to 3.0).
+exten => 1003,1,NoOp(LAB 1003: hold then human)
+ same => n,Answer()
+ same => n,Wait(1)
+ same => n,Playback(lab-speech)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-speech)
+ same => n,Playback(lab-speech)
+ same => n,Wait(30)
+ same => n,Hangup()
+
+; --- 1004: long hold ------------------------------------------------------
+; Exercises MAX_HOLD_TIME and HOLD_CHECK_INTERVAL. 10 minutes.
+exten => 1004,1,NoOp(LAB 1004: long hold)
+ same => n,Answer()
+ same => n,Wait(1)
+ same => n,Playback(lab-speech)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-music)
+ same => n,Playback(lab-speech)
+ same => n,Wait(15)
+ same => n,Hangup()
+
+; --- 1005: busy -----------------------------------------------------------
+; Failure path: the call must be marked FAILED with no stuck leg.
+exten => 1005,1,NoOp(LAB 1005: busy)
+ same => n,Busy(20)
+ same => n,Hangup()
+
+; --- 1006: ring, never answer ---------------------------------------------
+; Timeout path. Rings for 120s without answering.
+exten => 1006,1,NoOp(LAB 1006: ring no answer)
+ same => n,Progress()
+ same => n,Wait(120)
+ same => n,Hangup()
+
+; --- 1007: answer, then remote hangup after 5s ----------------------------
+; Proves remote-BYE handling and that the call persists to the DB on hangup.
+exten => 1007,1,NoOp(LAB 1007: quick remote hangup)
+ same => n,Answer()
+ same => n,Playback(${GREETING})
+ same => n,Wait(5)
+ same => n,Hangup()
+
+; --- 1008: answer, then silence -------------------------------------------
+; Classifier SILENCE vs the no-audio case. 45s of nothing.
+exten => 1008,1,NoOp(LAB 1008: silence)
+ same => n,Answer()
+ same => n,Playback(lab-silence)
+ same => n,Wait(40)
+ same => n,Hangup()
+
+; --- 2001: ring the registered softphone ----------------------------------
+; The transfer target. Asterisk is the registrar for devices, so the gateway
+; reaches a desk phone by dialling this rather than by registering it itself.
+; Fails fast when nothing is registered — a silent 30s ring would look like a
+; gateway bug rather than an absent softphone.
+exten => 2001,1,NoOp(LAB 2001: ring softphone)
+ ; Count registered contacts rather than DEVICE_STATE: device state follows
+ ; the OPTIONS qualify, which is off for this AOR (the pjsua CLI does not
+ ; answer OPTIONS), so a registered softphone would still read UNAVAILABLE.
+ same => n,GotoIf($[${PJSIP_AOR(softphone,contact)} = ""]?nodevice)
+ same => n,Dial(PJSIP/softphone,30)
+ same => n,Hangup()
+ same => n(nodevice),NoOp(LAB 2001: no softphone registered)
+ same => n,Answer()
+ same => n,Playback(lab-speech)
+ same => n,Hangup()
+
+; --- echo test ------------------------------------------------------------
+; Not a scenario — a debugging aid. Echoes audio back so you can confirm
+; bidirectional RTP by ear when something looks wrong.
+exten => 1099,1,NoOp(LAB 1099: echo test)
+ same => n,Answer()
+ same => n,Playback(lab-speech)
+ same => n,Echo()
+ same => n,Hangup()
+
+; Anything else: reject explicitly rather than failing obscurely.
+exten => _X.,1,NoOp(LAB: unknown extension ${EXTEN})
+ same => n,Answer()
+ same => n,Playback(${INVALID})
+ same => n,Hangup()

tests/lab/dialplan/logger.conf

@@ -0,0 +1,30 @@
+; Log to stdout only — Docker's json-file driver captures it and the host
+; Alloy ships it to Loki as job=<compose project>. Writing to a file inside
+; the container would put the logs where nothing can see them.
+[general]
+dateformat = %F %T
+; Colour is disabled in asterisk.conf (`nocolor = yes`), not here — Asterisk
+; colourises the console by default and the escape codes travel through Docker
+; into Loki, where every line arrives wrapped in \x1b[0;30m. That file must be
+; mounted for the setting to take effect.
+
+[logfiles]
+; `notice` is KEPT, deliberately. Asterisk logs rejected SIP requests at
+; notice level via log_failed_request ("No matching endpoint found",
+; "Failed to authenticate"), and on a box whose whole job is answering SIP
+; those are the single most useful diagnostic. Dropping notice made the log
+; quieter and the gateway undebuggable — a call was refused and nothing said
+; so.
+;
+; `verbose` is excluded: dialplan execution ("Executing [1001@...]") is
+; useful when tracing a specific call, but it is not worth shipping to Loki
+; continuously. Raise it at runtime instead:
+;     asterisk -rx "core set verbose 3"
+;
+; The healthcheck's "Remote UNIX connection" pairs also arrive on notice.
+; They are dealt with at the source rather than by silencing the channel:
+; docker-compose replaces the image's ~7-connections-per-30s healthcheck with
+; a single check on a 60s interval, and modules.conf stops the container
+; loading hardware and database modules it has no use for (~74 ALSA lines per
+; restart). Those two together cut the noise without costing visibility.
+console => notice,warning,error

tests/lab/dialplan/modules.conf

@@ -0,0 +1,46 @@
+; Module loading for the lab.
+;
+; The image ships `autoload=yes`, which loads every module Asterisk was built
+; with. On a headless container that produces a lot of startup noise for
+; hardware and backends that do not exist here — measured at ~74 ALSA lines
+; plus a dozen module-load ERRORs per restart, all of it in Loki. None of it
+; was harmful; all of it made the log harder to read.
+;
+; autoload stays on: this is a lab, and an explicit allow-list would break
+; quietly every time a scenario needs a module nobody remembered to add. The
+; noload lines below are only for modules that *cannot* work in this container.
+[modules]
+autoload=yes
+
+; --- Audio hardware -------------------------------------------------------
+; No sound card in a container. chan_alsa/chan_oss probe for one and emit
+; ~74 lines of ALSA config errors on every start. The media path is RTP via
+; PJSIP, never a local device.
+noload => chan_alsa.so
+noload => chan_oss.so
+noload => chan_console.so
+
+; --- CDR/CEL backends for databases we do not run -------------------------
+; Each logs "declined to load" or a config error at startup. Call records live
+; in Hold Slayer's own Postgres, written by the gateway, not by Asterisk.
+noload => cdr_pgsql.so
+noload => cdr_sqlite3_custom.so
+noload => cdr_custom.so
+noload => cdr_csv.so
+noload => cel_pgsql.so
+noload => cel_sqlite3_custom.so
+noload => cel_custom.so
+
+; --- Realtime config backends we do not use -------------------------------
+; The lab's configuration is the mounted .conf files. LDAP/ODBC/PgSQL realtime
+; each complain about missing connection details on every start.
+noload => res_config_ldap.so
+noload => res_config_odbc.so
+noload => res_config_pgsql.so
+noload => res_config_sqlite3.so
+
+; --- Codecs and formats that fail to initialise ----------------------------
+; format_ogg_vorbis errors on load. The lab plays .sln (signed linear) and
+; negotiates ulaw/alaw, so nothing here needs Vorbis.
+noload => format_ogg_vorbis.so
+noload => format_ogg_speex.so

tests/lab/dialplan/pjsip.conf

@@ -0,0 +1,132 @@
+; ---------------------------------------------------------------------------
+; Hold Slayer lab — PJSIP configuration
+; ---------------------------------------------------------------------------
+; SECURITY: this endpoint answers calls. Asterisk's stock examples allow
+; anonymous inbound, which is a well-known toll-fraud target. This config
+; refuses it: every call must authenticate as the `hold-slayer` endpoint.
+;
+; There is no PSTN behind this Asterisk — an unauthorised call reaches only
+; the lab dialplan and costs nothing. The lock-down is defence in depth and
+; so this config is never copied somewhere it would matter.
+; ---------------------------------------------------------------------------
+
+[global]
+type = global
+; Do not fall through to an `anonymous` endpoint for unmatched calls.
+; This is the single most important line in the file.
+unidentified_request_count = 5
+unidentified_request_period = 5
+unidentified_request_prune_interval = 30
+
+[transport-udp]
+type = transport
+protocol = udp
+bind = 0.0.0.0:{{ asterisk_sip_port }}
+; The address Asterisk advertises in SDP. Without this, containers advertise
+; their internal bridge IP and RTP arrives at an unroutable address — the
+; classic "call connects but there is no audio" failure.
+external_media_address = {{ asterisk_external_ip }}
+external_signaling_address = {{ asterisk_external_ip }}
+local_net = {{ asterisk_local_net }}
+
+; ---------------------------------------------------------------------------
+; Hold Slayer endpoint
+; ---------------------------------------------------------------------------
+; Hold Slayer authenticates as this endpoint to place calls into the lab.
+
+; Identify the endpoint by source address *and port*. Asterisk's default
+; matching uses the From-header domain, which Hold Slayer populates from its
+; SIP bind address (0.0.0.0 on a wildcard bind) — never a value Asterisk can
+; match. Matching on where the packet came from sidesteps that.
+;
+; The port is essential when the softphone runs on the same host: a
+; host-only match claims *every* packet from that address, so the
+; softphone's REGISTER would be attributed to this endpoint and checked
+; against the gateway's password ("Failed to authenticate", confusingly).
+; Endpoints that authenticate by username (the softphone) must not be
+; covered by an identify block.
+[hold-slayer]
+type = identify
+endpoint = hold-slayer
+match = {{ asterisk_match_host }}:{{ asterisk_gateway_port }}
+
+[hold-slayer]
+type = endpoint
+context = hold-slayer-lab
+disallow = all
+; ulaw first: it is what the PSTN uses, so the lab exercises the same codec
+; path a real trunk would. alaw as fallback.
+allow = ulaw
+allow = alaw
+auth = hold-slayer-auth
+aors = hold-slayer
+; RFC 2833 out-of-band DTMF — what send_dtmf must produce. Setting this
+; explicitly (rather than `auto`) means a DTMF failure is a real failure and
+; not a negotiation fallback quietly rescuing it.
+dtmf_mode = rfc4733
+direct_media = no
+force_rport = yes
+rewrite_contact = yes
+rtp_symmetric = yes
+
+[hold-slayer-auth]
+type = auth
+auth_type = userpass
+username = {{ asterisk_sip_username }}
+password = {{ asterisk_sip_password }}
+
+; ---------------------------------------------------------------------------
+; Softphone endpoint — the transfer target
+; ---------------------------------------------------------------------------
+; Asterisk is the registrar for devices, not Hold Slayer. A softphone REGISTERs
+; here and the gateway transfers a live call to it by dialling extension 2001.
+;
+; Deliberate: Hold Slayer's own SIP listener answers 200 OK to any REGISTER
+; with no digest challenge, so anything on the network could register as a
+; device and receive transferred calls. Keeping registration in Asterisk means
+; the lab does not depend on that path, and the softphone is authenticated.
+;
+; Test with the pjsua CLI built alongside the Python bindings:
+;   pjsua --null-audio --auto-answer=200 \
+;         --id=sip:softphone@<asterisk-host> \
+;         --registrar=sip:<asterisk-host>:21061 \
+;         --realm='*' --username=softphone --password=<pw> \
+;         --local-port=<free port>
+
+[softphone]
+type = endpoint
+context = hold-slayer-lab
+disallow = all
+allow = ulaw
+allow = alaw
+auth = softphone-auth
+aors = softphone
+dtmf_mode = rfc4733
+direct_media = no
+force_rport = yes
+rewrite_contact = yes
+rtp_symmetric = yes
+
+[softphone-auth]
+type = auth
+auth_type = userpass
+username = {{ asterisk_softphone_username }}
+password = {{ asterisk_softphone_password }}
+
+[softphone]
+type = aor
+; The device's contact is learned from its REGISTER rather than configured —
+; a softphone's port is not known in advance.
+max_contacts = 1
+remove_existing = yes
+; No qualify: the pjsua CLI does not answer OPTIONS while sitting at its
+; console prompt, so polling marks a perfectly working softphone Unavail and
+; the dialplan refuses to ring it. Registration itself is the liveness signal
+; here. A real hardphone answers OPTIONS and can have qualify re-enabled.
+qualify_frequency = 0
+
+[hold-slayer]
+type = aor
+max_contacts = 2
+remove_existing = yes
+qualify_frequency = 60

tests/lab/dialplan/rtp.conf

@@ -0,0 +1,9 @@
+; RTP media port range for the lab.
+;
+; 50 ports ≈ 25 concurrent calls — comfortably above Hold Slayer's
+; max_concurrent_calls (default 4). The range must match the ports published
+; in docker-compose, or media arrives at a port Docker isn't forwarding and
+; the call connects with no audio.
+[general]
+rtpstart = {{ asterisk_rtp_start }}
+rtpend = {{ asterisk_rtp_end }}

tests/lab/docker-compose.lab.yml

@@ -0,0 +1,45 @@
+# Local Asterisk lab — for iterating on caliban before promoting to Virgo.
+#
+# This is the LOCAL variant: ports and credentials are concrete, not Jinja.
+# Ansible templates the same dialplan out to galatea with the estate's
+# variables (see virgo/ansible/asterisk/).
+#
+# Run:  docker compose -f docker-compose.lab.yml up -d
+# CLI:  docker compose -f docker-compose.lab.yml exec asterisk asterisk -rvvv
+#
+# host networking: SIP/RTP carry IP addresses *inside* the payload, so a
+# bridged network needs external_media_address set correctly or the call
+# connects with no audio. Host networking sidesteps that entirely for local
+# work. The Virgo deploy uses the same approach for the same reason.
+services:
+  asterisk:
+    image: andrius/asterisk:22.10.1_debian-trixie
+    container_name: asterisk-lab
+    network_mode: host
+    volumes:
+      - ./dialplan/extensions.conf:/etc/asterisk/extensions.conf:ro
+      - ./dialplan/pjsip.local.conf:/etc/asterisk/pjsip.conf:ro
+      - ./dialplan/rtp.local.conf:/etc/asterisk/rtp.conf:ro
+      - ./dialplan/logger.conf:/etc/asterisk/logger.conf:ro
+      # Carries `nocolor = yes`: without it every log line reaches Loki
+      # wrapped in ANSI escape codes.
+      - ./dialplan/asterisk.conf:/etc/asterisk/asterisk.conf:ro
+      # noload for hardware/DB modules this container cannot use — the image
+      # autoloads everything, which costs ~74 ALSA lines per restart.
+      - ./dialplan/modules.conf:/etc/asterisk/modules.conf:ro
+      # The image ships no sound files at all. These are generated by
+      # sounds/generate.py; Asterisk resolves Playback(lab-music) to
+      # lab-music.sln here (8kHz signed-linear, no transcoding).
+      - ./sounds:/var/lib/asterisk/sounds/en:ro
+    # The image's default command is `-vvvdddf` — verbosity 3 and debug 3
+    # forced on the command line, which overrides both asterisk.conf and
+    # logger.conf. That makes every healthcheck CLI connection log a
+    # "Remote UNIX connection" pair: ~2900 lines/day of pure noise that
+    # completely buried the real SIP events in Loki.
+    #
+    # -f foreground (required: Docker needs PID 1 to stay), -T timestamps,
+    # -W colour off, -U run as asterisk, -p realtime priority. No -v, no -d:
+    # warnings and errors still log, and verbosity can be raised at runtime
+    # with `asterisk -rx "core set verbose 3"` when tracing a call.
+    command: ["/usr/sbin/asterisk", "-f", "-T", "-W", "-U", "asterisk", "-p"]
+    restart: unless-stopped

tests/lab/sounds/.gitignore

@@ -0,0 +1,4 @@
+# Generated by generate.py — deterministic from fixed seeds, so the bytes are
+# reproducible and there is no reason to carry ~680K of binary in the repo.
+# Run `python generate.py` before starting the lab.
+*.sln

tests/lab/sounds/generate.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""Generate the lab's audio fixtures.
+
+The Asterisk container ships with no sound files, and the design calls for
+*deterministic* audio: real hold music varies per call, so a classifier
+regression on the PSTN is indistinguishable from noise. These are synthesised
+from a fixed seed, so every run classifies identical input.
+
+Output is 8 kHz 16-bit mono signed-linear (.sln), which Asterisk plays without
+transcoding — the format is implied by the extension, so `Playback(lab-music)`
+finds `lab-music.sln`.
+
+    python generate.py [outdir]
+"""
+import sys
+from pathlib import Path
+
+import numpy as np
+
+RATE = 8000  # Asterisk's native rate for ulaw/alaw telephony
+
+
+def _write_sln(path: Path, samples: np.ndarray) -> None:
+    """Write float samples in [-1, 1] as 16-bit signed little-endian PCM."""
+    clipped = np.clip(samples, -1.0, 1.0)
+    pcm = (clipped * 32767).astype("<i2")
+    path.write_bytes(pcm.tobytes())
+    print(f"  {path.name}: {len(pcm) / RATE:.1f}s ({path.stat().st_size} bytes)")
+
+
+def make_music(seconds: float = 30.0, seed: int = 7) -> np.ndarray:
+    """Sustained multi-harmonic tones — what the classifier must call MUSIC.
+
+    A chord progression with stable pitch and strong harmonic structure. The
+    steady spectrum across a long window is what distinguishes music from
+    speech; this deliberately has no pauses.
+    """
+    rng = np.random.default_rng(seed)
+    t = np.linspace(0, seconds, int(RATE * seconds), endpoint=False)
+    # A-minor-ish progression, one chord per 2s bar.
+    chords = [(220.0, 261.6, 329.6), (196.0, 246.9, 293.7),
+              (174.6, 220.0, 261.6), (196.0, 246.9, 329.6)]
+    out = np.zeros_like(t)
+    bar = 2.0
+    for i, chord in enumerate(chords * int(np.ceil(seconds / (bar * len(chords))))):
+        start, end = i * bar, (i + 1) * bar
+        if start >= seconds:
+            break
+        mask = (t >= start) & (t < end)
+        for j, freq in enumerate(chord):
+            # Fundamental plus four harmonics, decaying — a plucked-string
+            # feel. Enough harmonics to keep spectral flatness inside the
+            # music score's 0.05-0.4 band: with only three, some windows fall
+            # *below* 0.05 (too pure to read as music) and score as speech.
+            for h, amp in ((1, 0.30), (2, 0.12), (3, 0.05), (4, 0.03), (5, 0.02)):
+                out[mask] += amp / (j + 1) * np.sin(2 * np.pi * freq * h * t[mask])
+        # Gentle per-bar envelope so bars are distinguishable but never silent.
+        env = 0.8 + 0.2 * np.sin(2 * np.pi * (t[mask] - start) / bar)
+        out[mask] *= env
+
+    # Recording-style noise floor. Windows straddling a chord change have a
+    # momentarily sparse spectrum and land just *under* the music score's
+    # 0.05 flatness floor, scoring as speech. This is well below the level
+    # that would disturb tonality — every real recording has one.
+    out += rng.normal(0, 0.004, len(out))
+    return out * 0.45
+
+
+def make_speech(seconds: float = 8.0, seed: int = 1337) -> np.ndarray:
+    """Formant-like bursts with pauses — what the classifier must call SPEECH.
+
+    Not real speech, but it carries the features the classifier keys on: a
+    fundamental in the human range, shifting formants, and syllable-rate
+    amplitude modulation with genuine silence between utterances.
+    """
+    rng = np.random.default_rng(seed)
+    t = np.linspace(0, seconds, int(RATE * seconds), endpoint=False)
+    out = np.zeros_like(t)
+
+    pos = 0.3  # leading pause
+    while pos < seconds - 0.4:
+        syl = rng.uniform(0.12, 0.28)  # syllable length
+        mask = (t >= pos) & (t < pos + syl)
+        if mask.any():
+            local = t[mask] - pos
+            frac = local / syl
+
+            # Pitch CONTOUR, not a constant. This is the single feature that
+            # separates this fixture from music. `_detect_tonality` looks for
+            # an autocorrelation peak > 0.5 in the 50-1000 Hz lag range; a
+            # fixed f0 is perfectly periodic there, scores is_tonal=True, and
+            # hands the music score a free 0.3 that speech cannot outrun.
+            # Real voices glide and jitter, so the periodicity never locks.
+            f0_start = rng.uniform(95, 165)
+            f0_end = f0_start * rng.uniform(0.72, 1.38)  # rise or fall
+            f0 = f0_start + (f0_end - f0_start) * frac
+            # Cycle-to-cycle jitter on top of the glide (~2% is human).
+            f0 *= 1.0 + 0.02 * rng.standard_normal(len(local))
+            # Integrate frequency to phase — with a varying f0, `2*pi*f*t`
+            # would be wrong (that is a chirp only if f is the *instantaneous*
+            # rate, which it is not once f0 itself moves).
+            ph0 = 2 * np.pi * np.cumsum(f0) / RATE
+
+            # Two formants, swept across the syllable. The ranges deliberately
+            # avoid the DTMF bands (rows 697-941, columns 1209-1633): a formant
+            # pair landing on both trips the Goertzel detector and the whole
+            # utterance is classified as a keypress.
+            f1 = rng.uniform(300, 620) + rng.uniform(-40, 40) * frac
+            f2 = rng.uniform(1750, 2600) + rng.uniform(-120, 120) * frac
+            sig = (0.50 * np.sin(ph0)
+                   + 0.30 * np.sin(2 * np.pi * f1 * local)
+                   + 0.18 * np.sin(2 * np.pi * f2 * local))
+            # Aspiration noise — HIGH-PASSED, not broadband. Real speech noise
+            # sits above the formants; flat noise puts energy in every
+            # Goertzel bin, so the strongest DTMF row and column both clear
+            # the detector's `total_power * 0.1` threshold and every syllable
+            # reads as a keypress. A first-difference filter (y[n]-y[n-1]) is
+            # a cheap +6dB/octave tilt that leaves the 697-1633 Hz DTMF bands
+            # comparatively empty. The 0.09 level is chosen for margin: it puts
+            # spectral flatness at ~0.46, mid-way through the 0.1-0.5 band the
+            # speech score rewards, rather than on either edge.
+            noise = rng.standard_normal(len(local) + 1)
+            sig += 0.09 * np.diff(noise)
+            # Raised-cosine envelope: no clicks at syllable edges.
+            sig *= np.sin(np.pi * local / syl) ** 0.6
+            out[mask] += sig
+        # Inter-syllable gap; occasionally a longer between-word pause.
+        pos += syl + (rng.uniform(0.25, 0.5) if rng.random() < 0.25
+                      else rng.uniform(0.04, 0.12))
+    return out * 0.55
+
+
+def make_silence(seconds: float = 5.0) -> np.ndarray:
+    """Near-silence with a trace of noise — real lines are never digitally flat."""
+    rng = np.random.default_rng(4242)
+    return rng.normal(0, 0.0006, int(RATE * seconds))
+
+
+def main() -> None:
+    outdir = Path(sys.argv[1] if len(sys.argv) > 1 else Path(__file__).parent)
+    outdir.mkdir(parents=True, exist_ok=True)
+    print(f"Generating lab audio into {outdir}/")
+    _write_sln(outdir / "lab-music.sln", make_music())
+    _write_sln(outdir / "lab-speech.sln", make_speech())
+    _write_sln(outdir / "lab-silence.sln", make_silence())
+    print("Done. Deterministic: same bytes on every run.")
+
+
+if __name__ == "__main__":
+    main()

tests/test_lab_fixtures.py

@@ -0,0 +1,109 @@
+"""
+Lab audio fixtures — the classifier must agree with what each one claims to be.
+
+These guard the *fixtures*, not the classifier. `tests/lab/sounds/generate.py`
+synthesises music/speech/silence that the Asterisk lab plays down a real call;
+if a fixture drifts into the wrong class, every lab result built on it is
+quietly meaningless — a hold-music scenario that never classifies as music
+proves nothing about the hold slayer.
+
+The first version of these fixtures passed on the opening 3s window and drifted
+to MUSIC after, which a single-window check would not have caught. Hence the
+sweep across every window.
+
+Skipped when the fixtures have not been generated: they are gitignored (~680K,
+reproducible from a fixed seed), so a fresh checkout has none until
+`python tests/lab/sounds/generate.py` runs.
+"""
+
+import subprocess
+import sys
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from config import Settings
+from models.call import AudioClassification
+from services.audio_classifier import SAMPLE_RATE, AudioClassifier
+
+SOUNDS_DIR = Path(__file__).parent / "lab" / "sounds"
+GENERATOR = SOUNDS_DIR / "generate.py"
+
+# The lab writes 8 kHz .sln; the classifier works at 16 kHz.
+LAB_RATE = 8000
+WINDOW_SAMPLES = SAMPLE_RATE * 3  # classifier's 3s analysis window
+
+FIXTURES = [
+    ("lab-music.sln", AudioClassification.MUSIC),
+    ("lab-speech.sln", AudioClassification.LIVE_HUMAN),
+    ("lab-silence.sln", AudioClassification.SILENCE),
+]
+
+
+def _load_16k(path: Path) -> np.ndarray:
+    """Load an 8 kHz .sln and upsample to the classifier's 16 kHz."""
+    return np.repeat(np.fromfile(path, dtype="<i2"), SAMPLE_RATE // LAB_RATE)
+
+
+def _windows(samples: np.ndarray, step: int):
+    """Yield successive analysis windows; at least one, even for short files."""
+    end = max(1, len(samples) - WINDOW_SAMPLES)
+    for offset in range(0, end, step):
+        yield samples[offset : offset + WINDOW_SAMPLES].astype("<i2").tobytes()
+
+
+@pytest.fixture(scope="module")
+def classifier():
+    return AudioClassifier(settings=Settings().classifier)
+
+
+@pytest.mark.parametrize("filename,expected", FIXTURES)
+def test_fixture_classifies_correctly_in_every_window(filename, expected, classifier):
+    """Every window must classify correctly — not just the first.
+
+    Stepped at half the window length so windows overlap: a fixture that only
+    works on aligned boundaries would still be a trap in a live call, where
+    the window has no relationship to where the audio started.
+    """
+    path = SOUNDS_DIR / filename
+    if not path.exists():
+        pytest.skip(f"{filename} not generated — run {GENERATOR}")
+
+    samples = _load_16k(path)
+    results = [
+        classifier.classify_chunk(w).audio_type
+        for w in _windows(samples, step=WINDOW_SAMPLES // 2)
+    ]
+
+    wrong = [(i, r.value) for i, r in enumerate(results) if r is not expected]
+    assert not wrong, (
+        f"{filename} must classify as {expected.value} in all "
+        f"{len(results)} windows; wrong: {wrong}"
+    )
+
+
+def test_generator_is_deterministic(tmp_path):
+    """Same bytes on every run — the whole point of synthesising them.
+
+    Real hold music varies per call, so a classifier regression on the PSTN is
+    indistinguishable from noise. Fixed-seed audio makes the answer binary.
+    """
+    if not GENERATOR.exists():
+        pytest.skip("generator not present")
+
+    def run(target: Path) -> dict[str, bytes]:
+        subprocess.run(
+            [sys.executable, str(GENERATOR), str(target)],
+            check=True,
+            capture_output=True,
+        )
+        return {p.name: p.read_bytes() for p in sorted(target.glob("*.sln"))}
+
+    first = run(tmp_path / "a")
+    second = run(tmp_path / "b")
+
+    assert first, "generator produced no .sln files"
+    assert first.keys() == second.keys()
+    for name in first:
+        assert first[name] == second[name], f"{name} differs between runs"