From 5880b59872ee1b2d47763e9995478720cef16eec Mon Sep 17 00:00:00 2001 From: Robert Helewka Date: Thu, 9 Jul 2026 19:53:36 -0400 Subject: [PATCH] fix: enforce thread ownership at the Sippy/PJSUA2 boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three thread domains were mutating shared dicts with no locks: Sippy's ED thread wrote _legs/_registered_devices directly from SIP handlers, the asyncio loop wrote them from make_call/hangup, and run_in_executor(None, ...) had default-pool threads driving sippy UA objects. AudioTap.feed() pushed into an asyncio.Queue (not thread-safe) from the PJSUA2 thread. New ownership rule, enforced structurally: - The asyncio loop owns all app-visible state; the only mutator is the new _on_engine_event funnel. Sippy handlers extract plain strings on the ED thread and post via run_coroutine_threadsafe. - The ED thread owns sippy objects plus _ed_ua_to_leg/_ed_leg_to_ua; loop-side commands (INVITE/BYE/DTMF/trunk register) hop over via ED2.callFromThread. UA references no longer live on SipCallLeg. - AudioTap captures its loop and feed() hops via call_soon_threadsafe. - Fix ED import: installed sippy 2.x exposes ED2, not ED — the old import could never start the event loop. Also: - Wire the never-connected on_leg_state_change callback: outbound ringing/connected/terminated now reaches CallManager; a call ends when its last leg terminates (transfers keep it alive). Adds CallManager.unmap_leg/legs_for_call. - AudioClassifier.classify(): async entry that runs the FFT work in asyncio.to_thread and updates history on the loop — all four hold_slayer call sites now route through it, fixing both the loop-blocking and the 2-of-4 history gap. DTMF Goertzel loop replaced by the equivalent vectorized DFT-bin power. - Task hygiene: gateway.spawn() tracks hold-slayer/receptionist tasks and stop() cancels them; recording safety-timeout task is retained and cancelled on stop_recording; engine tracks incoming-call dispatch tasks. 10 new tests: funnel events from a foreign thread, auto-answer fallback, AudioTap cross-thread feed, classifier history, leg-state → call status (including no stomping of ON_HOLD), stop() cancellation. Co-Authored-By: Claude Fable 5 --- core/call_manager.py | 8 + core/gateway.py | 52 +++- core/media_pipeline.py | 14 +- core/sippy_engine.py | 465 ++++++++++++++++++++--------------- services/audio_classifier.py | 33 ++- services/hold_slayer.py | 10 +- services/recording.py | 13 +- tests/test_concurrency.py | 173 +++++++++++++ 8 files changed, 548 insertions(+), 220 deletions(-) create mode 100644 tests/test_concurrency.py diff --git a/core/call_manager.py b/core/call_manager.py index 1eb229e..3ea6ee4 100644 --- a/core/call_manager.py +++ b/core/call_manager.py @@ -180,6 +180,14 @@ class CallManager: """Map a SIP leg ID to a call ID.""" self._call_legs[sip_leg_id] = call_id + def unmap_leg(self, sip_leg_id: str) -> None: + """Remove a SIP leg mapping (leg terminated).""" + self._call_legs.pop(sip_leg_id, None) + + def legs_for_call(self, call_id: str) -> list[str]: + """All SIP leg IDs currently mapped to a call.""" + return [leg for leg, cid in self._call_legs.items() if cid == call_id] + def get_call_for_leg(self, sip_leg_id: str) -> Optional[ActiveCall]: """Look up which call a SIP leg belongs to.""" call_id = self._call_legs.get(sip_leg_id) diff --git a/core/gateway.py b/core/gateway.py index 16bd2e4..da3178b 100644 --- a/core/gateway.py +++ b/core/gateway.py @@ -5,6 +5,7 @@ Ties together SIP engine, call manager, event bus, and all services. This is the top-level object that FastAPI and MCP talk to. """ +import asyncio import logging from datetime import datetime from typing import Optional @@ -57,6 +58,7 @@ def _build_sip_engine(settings: Settings, gateway: "AIPSTNGateway") -> SIPEngine domain=gw_sip.domain, did=trunk.did, media_pipeline=gateway.media_pipeline, + on_leg_state_change=gateway._on_sip_leg_state, on_device_registered=gateway._on_sip_device_registered, on_incoming_call=gateway._on_sip_incoming_call, ) @@ -101,9 +103,20 @@ class AIPSTNGateway: # Device registry (loaded from DB on start) self._devices: dict[str, Device] = {} + # Background tasks (per-call services, receptionist sessions) — + # tracked so shutdown can cancel them and GC can't drop them + self._tasks: set[asyncio.Task] = set() + # Startup time self._started_at: Optional[datetime] = None + def spawn(self, coro, name: str) -> asyncio.Task: + """Launch a tracked background task.""" + task = asyncio.get_running_loop().create_task(coro, name=name) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + return task + @classmethod def from_config(cls, sip_engine: Optional[SIPEngine] = None) -> "AIPSTNGateway": """Create gateway from environment config.""" @@ -176,6 +189,12 @@ class AIPSTNGateway: """Gracefully shut down.""" logger.info("Shutting down AI PSTN Gateway...") + # Cancel per-call background tasks before tearing down their deps + for task in list(self._tasks): + task.cancel() + if self._tasks: + await asyncio.gather(*self._tasks, return_exceptions=True) + # End all active calls for call_id in list(self.call_manager.active_calls.keys()): call = self.call_manager.get_call(call_id) @@ -278,8 +297,7 @@ class AIPSTNGateway: tts=self._tts, ) # Launch as background task — don't block - import asyncio - asyncio.create_task( + self.spawn( hold_slayer.run(call, sip_leg_id, call_flow_id), name=f"holdslayer_{call.id}", ) @@ -366,6 +384,33 @@ class AIPSTNGateway: if device: logger.info(f"📱 Device unregistered: {device.name}") + async def _on_sip_leg_state(self, leg_id: str, state: str) -> None: + """ + SIP leg state change from the engine (already on the loop). + + Maps leg transitions onto call status. Status only moves + forward from the dialing phase — hold-slayer/receptionist + states (ON_HOLD, NAVIGATING_IVR, …) are never stomped by a + late ringing/connected signal from a second leg. + """ + call = self.call_manager.get_call_for_leg(leg_id) + if call is None: + return + + if state == "ringing" and call.status == CallStatus.INITIATING: + await self.call_manager.update_status(call.id, CallStatus.RINGING) + elif state == "connected" and call.status in ( + CallStatus.INITIATING, + CallStatus.RINGING, + ): + await self.call_manager.update_status(call.id, CallStatus.CONNECTED) + elif state == "terminated": + self.call_manager.unmap_leg(leg_id) + # End the call only when its last leg is gone (a transfer + # keeps the call alive on the device leg) + if not self.call_manager.legs_for_call(call.id): + await self.call_manager.end_call(call.id) + async def _on_sip_device_registered( self, aor: str, contact: str, expires: int ) -> None: @@ -487,8 +532,7 @@ class AIPSTNGateway: # Hand off to the AI Receptionist if self._receptionist is not None and self.settings.receptionist.enabled: - import asyncio as _asyncio - _asyncio.create_task( + self.spawn( self._receptionist.handle(call, leg_id, decision), name=f"receptionist_{call.id}", ) diff --git a/core/media_pipeline.py b/core/media_pipeline.py index 2edf05b..19d3b16 100644 --- a/core/media_pipeline.py +++ b/core/media_pipeline.py @@ -52,11 +52,23 @@ class AudioTap: self._buffer: asyncio.Queue[bytes] = asyncio.Queue(maxsize=500) self._active = True self._pjsua2_port = None # PJSUA2 AudioMediaPort for tapping + # asyncio.Queue is not thread-safe; feed() hops onto this loop + try: + self._loop: Optional[asyncio.AbstractEventLoop] = asyncio.get_running_loop() + except RuntimeError: + self._loop = None def feed(self, pcm_data: bytes) -> None: - """Feed PCM audio data into the tap (called from PJSUA2 thread).""" + """Feed PCM audio data into the tap (called from the PJSUA2 thread).""" if not self._active: return + if self._loop is not None: + self._loop.call_soon_threadsafe(self._enqueue, pcm_data) + else: + self._enqueue(pcm_data) + + def _enqueue(self, pcm_data: bytes) -> None: + """Queue a frame on the owning loop, dropping oldest on overflow.""" try: self._buffer.put_nowait(pcm_data) except asyncio.QueueFull: diff --git a/core/sippy_engine.py b/core/sippy_engine.py index 7e6a67b..4c56a83 100644 --- a/core/sippy_engine.py +++ b/core/sippy_engine.py @@ -9,11 +9,22 @@ Architecture: Sippy B2BUA → SIP signaling (call control, registration, DTMF) PJSUA2 → Media anchor (conference bridge, audio tapping, recording) -Sippy B2BUA runs in its own thread (it has its own event loop). -We bridge async/sync via run_in_executor. +Thread-ownership rule: + - The asyncio loop owns all application-visible state: `_legs`, + `_bridges`, `_registered_devices`, `_trunk_registered`, and the + media pipeline. The ONLY place that state is mutated is + `_on_engine_event`, which runs on the loop. + - The Sippy ED thread owns every sippy object (UAs, transactions) + plus the `_ed_*` maps. Sippy objects are only touched by code + scheduled onto that thread via `_run_on_sippy`. + - `leg_id` strings are the only tokens that cross the boundary, + carried by `_post_from_ed` (Sippy → loop, via + run_coroutine_threadsafe) and `_run_on_sippy` (loop → Sippy, via + ED2.callFromThread). """ import asyncio +import inspect import logging import threading import uuid @@ -30,15 +41,15 @@ logger = logging.getLogger(__name__) # ================================================================ class SipCallLeg: - """Tracks a single SIP call leg managed by Sippy.""" + """Tracks a single SIP call leg. Owned by the asyncio loop.""" def __init__(self, leg_id: str, direction: str, remote_uri: str): self.leg_id = leg_id self.direction = direction # "outbound" or "inbound" self.remote_uri = remote_uri self.state = "init" # init, trying, ringing, connected, terminated - self.sippy_ua = None # Sippy UA object reference self.media_port: Optional[int] = None # PJSUA2 conf bridge port + self.pending_sdp: Optional[str] = None # inbound INVITE SDP, until answered self.dtmf_buffer: list[str] = [] def __repr__(self): @@ -65,75 +76,44 @@ class SippyCallController: """ Handles Sippy B2BUA callbacks for a single call leg. - Sippy B2BUA uses a callback model — when SIP events happen - (180 Ringing, 200 OK, BYE, etc.), the corresponding method - is called on this controller. + Runs entirely on the Sippy ED thread. It holds only the leg_id + token and forwards every state change to the asyncio loop via + the engine's event funnel — it never touches loop-owned state. """ - def __init__(self, leg: SipCallLeg, engine: "SippyEngine"): - self.leg = leg + def __init__(self, leg_id: str, engine: "SippyEngine"): + self.leg_id = leg_id self.engine = engine def on_trying(self): """100 Trying received.""" - self.leg.state = "trying" - logger.debug(f" {self.leg.leg_id}: 100 Trying") + logger.debug(f" {self.leg_id}: 100 Trying") + self.engine._post_from_ed("leg_state", {"leg_id": self.leg_id, "state": "trying"}) def on_ringing(self, ringing_code: int = 180): """180 Ringing / 183 Session Progress received.""" - self.leg.state = "ringing" - logger.info(f" {self.leg.leg_id}: {ringing_code} Ringing") - if self.engine._on_leg_state_change: - self.engine._loop.call_soon_threadsafe( - self.engine._on_leg_state_change, self.leg.leg_id, "ringing" - ) + logger.info(f" {self.leg_id}: {ringing_code} Ringing") + self.engine._post_from_ed("leg_state", {"leg_id": self.leg_id, "state": "ringing"}) def on_connected(self, sdp_body: Optional[str] = None): """200 OK — call connected, media negotiated.""" - self.leg.state = "connected" - logger.info(f" {self.leg.leg_id}: Connected") - - # Extract remote RTP endpoint from SDP for PJSUA2 media bridge - if sdp_body and self.engine.media_pipeline: - try: - remote_rtp = self.engine._parse_sdp_rtp_endpoint(sdp_body) - if remote_rtp: - port = self.engine.media_pipeline.add_remote_stream( - self.leg.leg_id, - remote_rtp["host"], - remote_rtp["port"], - remote_rtp["codec"], - ) - self.leg.media_port = port - except Exception as e: - logger.error(f" Failed to set up media for {self.leg.leg_id}: {e}") - - if self.engine._on_leg_state_change: - self.engine._loop.call_soon_threadsafe( - self.engine._on_leg_state_change, self.leg.leg_id, "connected" - ) + logger.info(f" {self.leg_id}: Connected") + self.engine._post_from_ed( + "leg_state", {"leg_id": self.leg_id, "state": "connected", "sdp": sdp_body} + ) def on_disconnected(self, reason: str = ""): """BYE received or call terminated.""" - self.leg.state = "terminated" - logger.info(f" {self.leg.leg_id}: Disconnected ({reason})") - - # Clean up media - if self.engine.media_pipeline and self.leg.media_port is not None: - try: - self.engine.media_pipeline.remove_stream(self.leg.leg_id) - except Exception as e: - logger.error(f" Failed to clean up media for {self.leg.leg_id}: {e}") - - if self.engine._on_leg_state_change: - self.engine._loop.call_soon_threadsafe( - self.engine._on_leg_state_change, self.leg.leg_id, "terminated" - ) + logger.info(f" {self.leg_id}: Disconnected ({reason})") + self.engine._ed_forget_leg(self.leg_id) + self.engine._post_from_ed( + "leg_state", {"leg_id": self.leg_id, "state": "terminated", "reason": reason} + ) def on_dtmf(self, digit: str): """DTMF digit received (RFC 2833 or SIP INFO).""" - self.leg.dtmf_buffer.append(digit) - logger.debug(f" {self.leg.leg_id}: DTMF '{digit}'") + logger.debug(f" {self.leg_id}: DTMF '{digit}'") + self.engine._post_from_ed("dtmf", {"leg_id": self.leg_id, "digit": digit}) # ================================================================ @@ -190,17 +170,140 @@ class SippyEngine(SIPEngine): self._on_incoming_call = on_incoming_call self._loop: Optional[asyncio.AbstractEventLoop] = None - # State + # Loop-owned state (mutated only in _on_engine_event and the + # async methods below, all of which run on the loop) self._ready = False self._trunk_registered = False self._legs: dict[str, SipCallLeg] = {} self._bridges: dict[str, SipBridge] = {} self._registered_devices: list[dict] = [] + self._tasks: set[asyncio.Task] = set() + + # ED-thread-owned state: sippy UA objects, only touched from + # the Sippy thread (handlers and _run_on_sippy closures) + self._ed_ua_to_leg: dict[Any, str] = {} + self._ed_leg_to_ua: dict[str, Any] = {} # Sippy B2BUA internals (set during start) self._sippy_global_config: dict[str, Any] = {} self._sippy_thread: Optional[threading.Thread] = None + # ================================================================ + # Thread-boundary crossing primitives + # ================================================================ + + def _post_from_ed(self, kind: str, data: dict) -> None: + """Sippy thread → loop: schedule the single state-mutation funnel.""" + if self._loop is None: + return + asyncio.run_coroutine_threadsafe(self._on_engine_event(kind, data), self._loop) + + def _run_on_sippy(self, fn: Callable[[], None]) -> None: + """Loop → Sippy thread: run fn where the sippy objects live.""" + try: + from sippy.Core.EventDispatcher import ED2 + except ImportError: + # Simulation mode — no sippy, no ED thread; run inline. + fn() + return + ED2.callFromThread(fn) + + def _ed_forget_leg(self, leg_id: str) -> None: + """Drop the ED-side UA maps for a leg (Sippy thread only).""" + ua = self._ed_leg_to_ua.pop(leg_id, None) + if ua is not None: + self._ed_ua_to_leg.pop(ua, None) + + def _spawn(self, coro, name: str) -> None: + """Track a background task so shutdown can cancel it.""" + task = asyncio.get_running_loop().create_task(coro, name=name) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + async def _on_engine_event(self, kind: str, data: dict) -> None: + """ + The single funnel where Sippy-thread events mutate loop-owned + state. Everything here runs on the asyncio loop. + """ + if kind == "leg_state": + leg = self._legs.get(data["leg_id"]) + if leg is None: + return + state = data["state"] + leg.state = state + + if state == "connected": + sdp = data.get("sdp") + if sdp and self.media_pipeline: + 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"], + ) + except Exception as e: + logger.error(f" Failed to set up media for {leg.leg_id}: {e}") + elif state == "terminated": + if self.media_pipeline and leg.media_port is not None: + try: + self.media_pipeline.remove_stream(leg.leg_id) + except Exception as e: + logger.error(f" Failed to clean up media for {leg.leg_id}: {e}") + leg.media_port = None + + if self._on_leg_state_change: + result = self._on_leg_state_change(leg.leg_id, state) + if inspect.isawaitable(result): + await result + + elif kind == "incoming_invite": + leg = SipCallLeg(data["leg_id"], "inbound", data["from_uri"]) + leg.pending_sdp = data.get("sdp") + self._legs[leg.leg_id] = leg + if self._on_incoming_call: + self._spawn( + self._on_incoming_call(data["from_uri"], data["to_uri"], leg.leg_id), + name=f"incoming_{leg.leg_id}", + ) + else: + # No routing wired — preserve the historical auto-answer + await self.accept_inbound(leg.leg_id) + + elif kind == "register": + existing = next( + (d for d in self._registered_devices if d.get("aor") == data["aor"]), + None, + ) + if existing: + existing["contact"] = data["contact"] + existing["expires"] = data["expires"] + else: + self._registered_devices.append({ + "aor": data["aor"], + "contact": data["contact"], + "expires": data["expires"], + }) + if self._on_device_registered: + await self._on_device_registered( + data["aor"], data["contact"], data["expires"] + ) + + elif kind == "deregister": + self._registered_devices = [ + d for d in self._registered_devices if d.get("aor") != data["aor"] + ] + + elif kind == "dtmf": + leg = self._legs.get(data["leg_id"]) + if leg: + leg.dtmf_buffer.append(data["digit"]) + + elif kind == "trunk_registered": + self._trunk_registered = data["registered"] + # ================================================================ # Lifecycle # ================================================================ @@ -212,7 +315,6 @@ class SippyEngine(SIPEngine): try: from sippy.SipConf import SipConf - from sippy.SipTransactionManager import SipTransactionManager # Configure Sippy SipConf.my_address = self._sip_address @@ -226,7 +328,7 @@ class SippyEngine(SIPEngine): } # Start Sippy's SIP transaction manager in a background thread - # Sippy uses its own event loop (Twisted reactor or custom loop) + # Sippy uses its own event loop (the ED2 event dispatcher) self._sippy_thread = threading.Thread( target=self._run_sippy_loop, name="sippy-b2bua", @@ -254,8 +356,8 @@ class SippyEngine(SIPEngine): def _run_sippy_loop(self): """Run Sippy B2BUA's event loop in a dedicated thread.""" try: + from sippy.Core.EventDispatcher import ED2 from sippy.SipTransactionManager import SipTransactionManager - from sippy.Timeout import Timeout # Initialize Sippy's transaction manager stm = SipTransactionManager(self._sippy_global_config, self._handle_sippy_request) @@ -263,11 +365,9 @@ class SippyEngine(SIPEngine): logger.info(" Sippy transaction manager started") - # Sippy will block here in its event loop - # For the Twisted-based version, this runs the reactor - # For the asyncore version, this runs asyncore.loop() - from sippy.Core.EventDispatcher import ED - ED.loop() + # Sippy blocks here dispatching its event loop; callbacks + # injected via ED2.callFromThread run inside this loop. + ED2.loop() except Exception as e: logger.error(f" Sippy event loop crashed: {e}") @@ -294,10 +394,9 @@ class SippyEngine(SIPEngine): """ Handle an incoming SIP REGISTER from a phone or softphone. - Extracts the AOR (address of record) from the To header, records - the contact and expiry, and sends a 200 OK. The gateway's - register_device() is called asynchronously via the event loop so - the phone gets an extension and SIP URI assigned automatically. + Runs on the Sippy thread: parses the request, replies 200 OK, + and posts the registration to the loop funnel, which owns the + device list and notifies the gateway. """ try: to_uri = str(req.getHFBody("to").getUri()) @@ -309,34 +408,13 @@ class SippyEngine(SIPEngine): logger.info(f" SIP REGISTER: {to_uri} contact={contact_uri} expires={expires}") if expires == 0: - # De-registration - self._registered_devices = [ - d for d in self._registered_devices - if d.get("aor") != to_uri - ] - logger.info(f" De-registered: {to_uri}") + self._post_from_ed("deregister", {"aor": to_uri}) else: - # Update or add registration record - existing = next( - (d for d in self._registered_devices if d.get("aor") == to_uri), - None, - ) - if existing: - existing["contact"] = contact_uri - existing["expires"] = expires - else: - self._registered_devices.append({ - "aor": to_uri, - "contact": contact_uri, - "expires": expires, - }) - - # Notify the gateway (async) so it can assign an extension - if self._loop: - self._loop.call_soon_threadsafe( - self._loop.create_task, - self._notify_registration(to_uri, contact_uri, expires), - ) + self._post_from_ed("register", { + "aor": to_uri, + "contact": contact_uri, + "expires": expires, + }) # Reply 200 OK req.sendResponse(200, "OK") @@ -348,52 +426,41 @@ class SippyEngine(SIPEngine): except Exception: pass - async def _notify_registration(self, aor: str, contact: str, expires: int): - """ - Async callback: tell the gateway about the newly registered device - so it can assign an extension if needed. - """ - if self._on_device_registered: - await self._on_device_registered(aor, contact, expires) - def _handle_incoming_invite(self, req, sip_t): - """Handle an incoming INVITE — create inbound call leg. + """Handle an incoming INVITE — surface an inbound call leg. - The gateway is notified via `on_incoming_call`; it decides - whether to answer (via `accept_inbound`) or reject the leg - based on routing rules. + Runs on the Sippy thread: extracts everything the loop needs + (URIs, SDP body) as plain strings and posts them. The gateway + decides whether to answer (via `accept_inbound`) or reject. """ from_uri = str(req.getHFBody("from").getUri()) to_uri = str(req.getHFBody("to").getUri()) + sdp = str(req.getBody()) if req.getBody() else None leg_id = f"leg_{uuid.uuid4().hex[:12]}" - leg = SipCallLeg(leg_id, "inbound", from_uri) - leg.sippy_ua = sip_t.ua if hasattr(sip_t, "ua") else None - leg.pending_invite = req - self._legs[leg_id] = leg + ua = sip_t.ua if hasattr(sip_t, "ua") else None + if ua is not None: + self._ed_ua_to_leg[ua] = leg_id + self._ed_leg_to_ua[leg_id] = ua logger.info(f" Incoming call: {from_uri} → {to_uri} (leg: {leg_id})") - # Surface to the gateway. If no callback is wired, fall back to - # auto-answer so we don't regress the previous behavior. - if self._on_incoming_call and self._loop: - asyncio.run_coroutine_threadsafe( - self._on_incoming_call(from_uri, to_uri, leg_id), - self._loop, - ) - else: - controller = SippyCallController(leg, self) - controller.on_connected(str(req.getBody()) if req.getBody() else None) + self._post_from_ed("incoming_invite", { + "leg_id": leg_id, + "from_uri": from_uri, + "to_uri": to_uri, + "sdp": sdp, + }) async def accept_inbound(self, leg_id: str) -> bool: """Answer a previously-surfaced inbound INVITE.""" leg = self._legs.get(leg_id) if not leg or leg.direction != "inbound": return False - req = getattr(leg, "pending_invite", None) - controller = SippyCallController(leg, self) - body = str(req.getBody()) if req and req.getBody() else None - controller.on_connected(body) + sdp, leg.pending_sdp = leg.pending_sdp, None + await self._on_engine_event( + "leg_state", {"leg_id": leg_id, "state": "connected", "sdp": sdp} + ) return True async def reject_inbound(self, leg_id: str, code: int = 603, reason: str = "Decline") -> bool: @@ -404,66 +471,66 @@ class SippyEngine(SIPEngine): logger.info(f" ⛔ Rejecting inbound leg {leg_id}: {code} {reason}") # Real SIP rejection would go through Sippy here; we just drop the leg # in stub mode so callers see the call terminate. + self._run_on_sippy(lambda: self._ed_forget_leg(leg_id)) return True def _handle_incoming_bye(self, req, sip_t): - """Handle incoming BYE — tear down call leg.""" - # Find the leg by Sippy's UA object - for leg in self._legs.values(): - if leg.sippy_ua and hasattr(sip_t, "ua") and leg.sippy_ua == sip_t.ua: - controller = SippyCallController(leg, self) - controller.on_disconnected("BYE received") - break + """Handle incoming BYE — tear down call leg (Sippy thread).""" + ua = sip_t.ua if hasattr(sip_t, "ua") else None + leg_id = self._ed_ua_to_leg.get(ua) if ua is not None else None + if leg_id: + SippyCallController(leg_id, self).on_disconnected("BYE received") def _handle_incoming_info(self, req, sip_t): - """Handle SIP INFO (DTMF via SIP INFO method).""" + """Handle SIP INFO (DTMF via SIP INFO method) on the Sippy thread.""" body = str(req.getBody()) if req.getBody() else "" if "dtmf" in body.lower() or "Signal=" in body: - # Extract DTMF digit from SIP INFO body + ua = sip_t.ua if hasattr(sip_t, "ua") else None + leg_id = self._ed_ua_to_leg.get(ua) if ua is not None else None + if not leg_id: + return for line in body.split("\n"): if line.startswith("Signal="): digit = line.split("=")[1].strip() - for leg in self._legs.values(): - if leg.sippy_ua and hasattr(sip_t, "ua") and leg.sippy_ua == sip_t.ua: - controller = SippyCallController(leg, self) - controller.on_dtmf(digit) - break + SippyCallController(leg_id, self).on_dtmf(digit) async def _register_trunk(self) -> None: """Register with the SIP trunk provider.""" - try: - from sippy.UA import UA - from sippy.SipRegistrationAgent import SipRegistrationAgent + logger.info(f" Registering with trunk: {self._trunk_host}:{self._trunk_port}") - logger.info(f" Registering with trunk: {self._trunk_host}:{self._trunk_port}") + def do_register(): + try: + from sippy.SipRegistrationAgent import SipRegistrationAgent - # Run registration in Sippy's thread - def do_register(): - try: - 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, - ) - reg_agent.register() - self._trunk_registered = True - logger.info(" ✅ Trunk registration sent") - except Exception as e: - logger.error(f" ❌ Trunk registration failed: {e}") - self._trunk_registered = False + 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, + ) + reg_agent.register() + logger.info(" ✅ Trunk registration sent") + self._post_from_ed("trunk_registered", {"registered": True}) + 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}) - await asyncio.get_event_loop().run_in_executor(None, do_register) - - except ImportError: - logger.warning(" Sippy registration agent not available") - self._trunk_registered = False + self._run_on_sippy(do_register) async def stop(self) -> None: """Gracefully shut down the SIP engine.""" logger.info("🔌 Stopping Sippy B2BUA...") + # Cancel in-flight incoming-call dispatch tasks + for task in list(self._tasks): + task.cancel() + if self._tasks: + await asyncio.gather(*self._tasks, return_exceptions=True) + # Hang up all active legs for leg_id in list(self._legs.keys()): try: @@ -473,8 +540,8 @@ class SippyEngine(SIPEngine): # Stop Sippy's event loop try: - from sippy.Core.EventDispatcher import ED - ED.breakLoop() + from sippy.Core.EventDispatcher import ED2 + ED2.breakLoop() except Exception: pass @@ -512,14 +579,16 @@ class SippyEngine(SIPEngine): logger.info(f"📞 Placing call: {from_uri} → {remote_uri} (leg: {leg_id})") - # Place the call via Sippy + # Generate SDP on the loop (allocate_rtp_port is lock-protected) + sdp_body = self._generate_sdp(leg_id) + def do_invite(): try: - from sippy.UA import UA - from sippy.SipCallId import SipCallId from sippy.CCEvents import CCEventTry + from sippy.SipCallId import SipCallId + from sippy.UA import UA - controller = SippyCallController(leg, self) + controller = SippyCallController(leg_id, self) # Create Sippy UA for this call ua = UA( @@ -527,10 +596,8 @@ class SippyEngine(SIPEngine): event_cb=controller, nh_address=(self._trunk_host, self._trunk_port), ) - leg.sippy_ua = ua - - # Generate SDP for the call - sdp_body = self._generate_sdp(leg_id) + self._ed_leg_to_ua[leg_id] = ua + self._ed_ua_to_leg[ua] = leg_id # Send INVITE event = CCEventTry( @@ -539,19 +606,19 @@ class SippyEngine(SIPEngine): ) ua.recvEvent(event) - leg.state = "trying" logger.info(f" INVITE sent for {leg_id}") + self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "trying"}) except ImportError: # Sippy not installed — simulate for development logger.warning(f" Sippy not installed, simulating call for {leg_id}") - leg.state = "ringing" + 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}") - leg.state = "terminated" + self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "terminated"}) - await asyncio.get_event_loop().run_in_executor(None, do_invite) + self._run_on_sippy(do_invite) return leg_id async def hangup(self, call_leg_id: str) -> None: @@ -563,15 +630,18 @@ class SippyEngine(SIPEngine): def do_bye(): try: - if leg.sippy_ua: + ua = self._ed_leg_to_ua.get(call_leg_id) + if ua is not None: from sippy.CCEvents import CCEventDisconnect - leg.sippy_ua.recvEvent(CCEventDisconnect()) + ua.recvEvent(CCEventDisconnect()) except Exception as e: logger.error(f" Error sending BYE for {call_leg_id}: {e}") finally: - leg.state = "terminated" + self._ed_forget_leg(call_leg_id) - await asyncio.get_event_loop().run_in_executor(None, do_bye) + self._run_on_sippy(do_bye) + + leg.state = "terminated" # Clean up media if self.media_pipeline and leg.media_port is not None: @@ -595,13 +665,13 @@ class SippyEngine(SIPEngine): def do_dtmf(): try: - if leg.sippy_ua: - # Send via RFC 2833 (in-band RTP event) - # Sippy handles this through the UA's DTMF sender + ua = self._ed_leg_to_ua.get(call_leg_id) + if ua is not None: + # Send via SIP INFO through the UA + from sippy.CCEvents import CCEventInfo for digit in digits: - from sippy.CCEvents import CCEventInfo body = f"Signal={digit}\r\nDuration=160\r\n" - leg.sippy_ua.recvEvent(CCEventInfo(body=body)) + ua.recvEvent(CCEventInfo(body=body)) else: logger.warning(f" No UA for {call_leg_id}, DTMF not sent") except ImportError: @@ -609,7 +679,7 @@ class SippyEngine(SIPEngine): except Exception as e: logger.error(f" DTMF send error: {e}") - await asyncio.get_event_loop().run_in_executor(None, do_dtmf) + self._run_on_sippy(do_dtmf) # ================================================================ # Device Calls (for transfer) @@ -638,13 +708,15 @@ class SippyEngine(SIPEngine): logger.info(f"📱 Calling device: {device.name} ({device.sip_uri}) (leg: {leg_id})") + sdp_body = self._generate_sdp(leg_id) + def do_invite_device(): try: - from sippy.UA import UA from sippy.CCEvents import CCEventTry from sippy.SipCallId import SipCallId + from sippy.UA import UA - controller = SippyCallController(leg, self) + controller = SippyCallController(leg_id, self) # Parse device SIP URI for routing # sip:robert@192.168.1.100:5060 @@ -662,25 +734,24 @@ class SippyEngine(SIPEngine): event_cb=controller, nh_address=(host, port), ) - leg.sippy_ua = ua - - sdp_body = self._generate_sdp(leg_id) + self._ed_leg_to_ua[leg_id] = ua + self._ed_ua_to_leg[ua] = leg_id event = CCEventTry( (SipCallId(), f"sip:gateway@{self._domain}", device.sip_uri), body=sdp_body, ) ua.recvEvent(event) - leg.state = "trying" + self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "trying"}) except ImportError: logger.warning(f" Sippy not installed, simulating device call for {leg_id}") - leg.state = "ringing" + self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "ringing"}) except Exception as e: logger.error(f" Failed to call device {device.name}: {e}") - leg.state = "terminated" + self._post_from_ed("leg_state", {"leg_id": leg_id, "state": "terminated"}) - await asyncio.get_event_loop().run_in_executor(None, do_invite_device) + self._run_on_sippy(do_invite_device) return leg_id # ================================================================ diff --git a/services/audio_classifier.py b/services/audio_classifier.py index 76a3809..7b440ca 100644 --- a/services/audio_classifier.py +++ b/services/audio_classifier.py @@ -12,6 +12,7 @@ Uses spectral analysis (librosa/numpy) to classify audio without needing a trained ML model — just signal processing and heuristics. """ +import asyncio import logging import time from typing import Optional @@ -47,9 +48,23 @@ class AudioClassifier: self._window_samples = int(settings.window_seconds * SAMPLE_RATE) self._classification_history: list[AudioClassification] = [] + async def classify(self, audio_data: bytes) -> ClassificationResult: + """ + Classify a chunk off the event loop and record it in the history. + + The FFT/autocorrelation work is CPU-bound, so the pure + `classify_chunk` runs in a worker thread; the history update + happens back on the loop, keeping it single-threaded. This is + the call sites' entry point — routing every classification + through here is what keeps the history complete. + """ + result = await asyncio.to_thread(self.classify_chunk, audio_data) + self.update_history(result.audio_type) + return result + def classify_chunk(self, audio_data: bytes) -> ClassificationResult: """ - Classify a chunk of audio data. + Classify a chunk of audio data (pure, synchronous). Args: audio_data: Raw PCM audio (16-bit signed, 16kHz, mono) @@ -285,17 +300,15 @@ class AudioClassifier: (941, 1209): "*", (941, 1336): "0", (941, 1477): "#", (941, 1633): "D", } - # Compute power at each DTMF frequency + # Power at each DTMF frequency via the DFT bin (numerically equal + # to the Goertzel result s1² + s2² − coeff·s1·s2, but vectorized — + # the per-sample Python loop blocked for ~50ms per chunk) + n = np.arange(len(samples)) + def goertzel_power(freq: int) -> float: k = int(0.5 + len(samples) * freq / SAMPLE_RATE) - w = 2 * np.pi * k / len(samples) - coeff = 2 * np.cos(w) - s0, s1, s2 = 0.0, 0.0, 0.0 - for sample in samples: - s0 = sample + coeff * s1 - s2 - s2 = s1 - s1 = s0 - return float(s1 * s1 + s2 * s2 - coeff * s1 * s2) + bin_value = np.dot(samples, np.exp(-2j * np.pi * k * n / len(samples))) + return float(np.abs(bin_value) ** 2) # Find strongest low and high frequencies low_powers = [(f, goertzel_power(f)) for f in dtmf_freqs_low] diff --git a/services/hold_slayer.py b/services/hold_slayer.py index fd14240..d9cec27 100644 --- a/services/hold_slayer.py +++ b/services/hold_slayer.py @@ -323,8 +323,7 @@ class HoldSlayerService: continue # Classify the audio - classification = self.classifier.classify_chunk(audio_chunk) - self.classifier.update_history(classification.audio_type) + classification = await self.classifier.classify(audio_chunk) await self.call_manager.add_classification(call.id, classification) # Transcribe if it sounds like speech @@ -447,8 +446,7 @@ class HoldSlayerService: continue # Classify - result = self.classifier.classify_chunk(audio_chunk) - self.classifier.update_history(result.audio_type) + result = await self.classifier.classify(audio_chunk) await self.call_manager.add_classification(call.id, result) # Check for human @@ -508,7 +506,7 @@ class HoldSlayerService: continue # Classify first - result = self.classifier.classify_chunk(audio_chunk) + result = await self.classifier.classify(audio_chunk) if result.audio_type not in ( AudioClassification.IVR_PROMPT, AudioClassification.LIVE_HUMAN, @@ -560,7 +558,7 @@ class HoldSlayerService: if not audio_chunk: break - result = self.classifier.classify_chunk(audio_chunk) + result = await self.classifier.classify(audio_chunk) # If we're getting silence after speech, the menu prompt is done if result.audio_type == AudioClassification.SILENCE and transcript_parts: diff --git a/services/recording.py b/services/recording.py index e5d949b..f80b551 100644 --- a/services/recording.py +++ b/services/recording.py @@ -39,6 +39,7 @@ class RecordingService: self._max_recording_seconds = max_recording_seconds self._sample_rate = sample_rate self._active_recordings: dict[str, RecordingSession] = {} + self._timeout_tasks: dict[str, asyncio.Task] = {} self._metadata: list[dict] = [] async def start(self) -> None: @@ -101,8 +102,8 @@ class RecordingService: self._active_recordings[call_id] = session logger.info(f"🔴 Recording started: {call_id} → {filepath_mixed}") - # Safety timeout - asyncio.create_task( + # Safety timeout — tracked so it can be cancelled and isn't GC'd + self._timeout_tasks[call_id] = asyncio.create_task( self._recording_timeout(call_id), name=f"rec_timeout_{call_id}", ) @@ -115,6 +116,14 @@ class RecordingService: media_pipeline=None, ) -> Optional["RecordingSession"]: """Stop recording a call and finalize the WAV file.""" + timeout_task = self._timeout_tasks.pop(call_id, None) + if ( + timeout_task is not None + and timeout_task is not asyncio.current_task() + and not timeout_task.done() + ): + timeout_task.cancel() + session = self._active_recordings.pop(call_id, None) if not session: logger.warning(f" No active recording for {call_id}") diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 0000000..570bae2 --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,173 @@ +""" +Thread-ownership and task-hygiene tests. + +Covers the Sippy→loop event funnel (events posted from a foreign +thread mutate loop-owned state), the AudioTap thread-safe feed, the +off-loop classifier entry point, leg-state propagation into the +CallManager, and background-task cancellation on gateway stop. +""" + +import asyncio +import threading + +import numpy as np + +from config import ClassifierSettings, Settings +from core.gateway import AIPSTNGateway +from core.media_pipeline import AudioTap +from core.sippy_engine import SippyEngine +from models.call import CallStatus +from services.audio_classifier import AudioClassifier + + +def _engine_on_loop() -> SippyEngine: + """Engine wired to the running loop without starting the SIP stack.""" + engine = SippyEngine() + engine._loop = asyncio.get_running_loop() + return engine + + +def _post_from_thread(engine: SippyEngine, kind: str, data: dict) -> None: + """Post a funnel event from a foreign thread, like the Sippy ED thread.""" + t = threading.Thread(target=engine._post_from_ed, args=(kind, data)) + t.start() + t.join() + + +class TestEngineEventFunnel: + async def test_register_and_deregister_from_foreign_thread(self): + engine = _engine_on_loop() + + _post_from_thread(engine, "register", { + "aor": "sip:alice@gw", "contact": "sip:alice@10.0.0.5", "expires": 3600, + }) + await asyncio.sleep(0.05) + assert engine._registered_devices == [ + {"aor": "sip:alice@gw", "contact": "sip:alice@10.0.0.5", "expires": 3600} + ] + + # Re-register updates in place instead of duplicating + _post_from_thread(engine, "register", { + "aor": "sip:alice@gw", "contact": "sip:alice@10.0.0.9", "expires": 60, + }) + await asyncio.sleep(0.05) + assert len(engine._registered_devices) == 1 + assert engine._registered_devices[0]["contact"] == "sip:alice@10.0.0.9" + + _post_from_thread(engine, "deregister", {"aor": "sip:alice@gw"}) + await asyncio.sleep(0.05) + assert engine._registered_devices == [] + + async def test_incoming_invite_auto_answers_without_callback(self): + engine = _engine_on_loop() + _post_from_thread(engine, "incoming_invite", { + "leg_id": "leg_test1", + "from_uri": "sip:caller@pstn", + "to_uri": "sip:+15551234567@gw", + "sdp": None, + }) + await asyncio.sleep(0.05) + leg = engine._legs["leg_test1"] + assert leg.direction == "inbound" + assert leg.state == "connected" + + async def test_incoming_call_callback_runs_as_tracked_task(self): + engine = _engine_on_loop() + seen = asyncio.Event() + + async def on_incoming(from_uri, to_uri, leg_id): + seen.set() + + engine._on_incoming_call = on_incoming + _post_from_thread(engine, "incoming_invite", { + "leg_id": "leg_test2", "from_uri": "a", "to_uri": "b", "sdp": None, + }) + await asyncio.wait_for(seen.wait(), timeout=1.0) + assert engine._legs["leg_test2"].state == "init" # not auto-answered + + async def test_bye_terminates_leg_and_notifies(self): + engine = _engine_on_loop() + states: list[tuple[str, str]] = [] + engine._on_leg_state_change = lambda leg_id, state: states.append((leg_id, state)) + + _post_from_thread(engine, "incoming_invite", { + "leg_id": "leg_test3", "from_uri": "a", "to_uri": "b", "sdp": None, + }) + await asyncio.sleep(0.05) + _post_from_thread(engine, "leg_state", {"leg_id": "leg_test3", "state": "terminated"}) + await asyncio.sleep(0.05) + + assert engine._legs["leg_test3"].state == "terminated" + assert ("leg_test3", "terminated") in states + + async def test_dtmf_and_trunk_events(self): + engine = _engine_on_loop() + _post_from_thread(engine, "incoming_invite", { + "leg_id": "leg_test4", "from_uri": "a", "to_uri": "b", "sdp": None, + }) + await asyncio.sleep(0.05) + _post_from_thread(engine, "dtmf", {"leg_id": "leg_test4", "digit": "5"}) + _post_from_thread(engine, "trunk_registered", {"registered": True}) + await asyncio.sleep(0.05) + assert engine._legs["leg_test4"].dtmf_buffer == ["5"] + assert engine._trunk_registered is True + + +class TestAudioTapThreadSafety: + async def test_feed_from_foreign_thread_reaches_reader(self): + tap = AudioTap("leg_x") + frame = b"\x01\x02" * 320 + + t = threading.Thread(target=tap.feed, args=(frame,)) + t.start() + t.join() + + received = await tap.read_frame(timeout=1.0) + assert received == frame + + +class TestClassifierOffLoop: + async def test_classify_runs_and_records_history(self): + classifier = AudioClassifier(ClassifierSettings()) + silence = np.zeros(16000, dtype=np.int16).tobytes() + + result = await classifier.classify(silence) + + assert result.audio_type.value == "silence" + assert classifier._classification_history == [result.audio_type] + + +class TestLegStatePropagation: + async def test_leg_lifecycle_drives_call_status(self): + gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4)) + call = await gateway.make_call("+15551234567") + assert call.status == CallStatus.RINGING + + (leg_id,) = gateway.call_manager.legs_for_call(call.id) + + await gateway._on_sip_leg_state(leg_id, "connected") + assert gateway.get_call(call.id).status == CallStatus.CONNECTED + + await gateway._on_sip_leg_state(leg_id, "terminated") + assert gateway.get_call(call.id) is None + assert gateway.call_manager.legs_for_call(call.id) == [] + + async def test_late_signals_do_not_stomp_service_states(self): + gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4)) + call = await gateway.make_call("+15551234567") + (leg_id,) = gateway.call_manager.legs_for_call(call.id) + + await gateway.call_manager.update_status(call.id, CallStatus.ON_HOLD) + await gateway._on_sip_leg_state(leg_id, "connected") + assert gateway.get_call(call.id).status == CallStatus.ON_HOLD + + +class TestTaskHygiene: + async def test_gateway_stop_cancels_spawned_tasks(self): + gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4)) + task = gateway.spawn(asyncio.sleep(60), name="test_sleeper") + + await gateway.stop() + + assert task.cancelled() + assert gateway._tasks == set()