fix: enforce thread ownership at the Sippy/PJSUA2 boundary

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 <noreply@anthropic.com>

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