Merge pull request 'Stage 2: thread ownership at the SIP boundary, leg-state wiring, task hygiene' (#2) from feature/stage2-concurrency into feature/stage1-agent-surface

Reviewed-on: #2

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)

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}",
             )

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:

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

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]

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:

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}")

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