Stage 4: honest health, explicit error policy, event-bus integrity

Engine mode is now explicit: USE_MOCK_SIP=true is the only way to get
the mock engine; an unconfigured trunk fails startup with guidance
instead of silently degrading. Root-caused why the engine always ran
mock: nested pydantic-settings never read .env (no env_file on the
sub-settings classes) — all 8 now declare it.

/health stops lying: reports engine mode (sippy|mock), a live DB
SELECT 1, trunk registration state with reason, and TTS/STT
availability from their last real request; "healthy" now requires
ready + db + sippy + registered trunk.

Error policy: leaf services (tts/transcription/llm_client) raise and
track availability; call-loop callers catch, publish EventType.ERROR
naming the failed service, and apply an explicit fallback. Persistence
writes get one bounded 3x exponential retry, then an ERROR log — no
more silent data loss.

Event bus: a full subscriber queue drops its oldest event (counted)
instead of silently evicting the subscription; subscribe(replay_last=N)
delivers the advertised history replay, used by /ws/events (25).

Receptionist correctness: a matched TAKE_MESSAGE rule beats the LLM;
voicemail polls for early hangup and stops/transcribes/hangs up in
finally; RecordingSession finally keeps its leg_ids so taps detach.

Dead code removed: models/contact.py + Contact table, dtmf_buffer,
transcribe_stream stub, SMS stub in notification.py.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

core/event_bus.py

@@ -20,63 +20,70 @@ class EventBus:
 
     Features:
     - Non-blocking publish (put_nowait)
-    - Automatic dead-subscriber cleanup (full queues are removed)
-    - Event history (last N events for late joiners)
+    - Slow subscribers lose their oldest event, never their subscription
+    - Event history (last N events, replayable to late joiners)
     - Typed event filtering on subscriptions
     - Async iteration via EventSubscription
     """
 
     def __init__(self, max_history: int = 1000):
-        self._subscribers: list[tuple[asyncio.Queue[GatewayEvent], Optional[set[EventType]]]] = []
+        self._subscribers: list[EventSubscription] = []
         self._history: list[GatewayEvent] = []
         self._max_history = max_history
 
     async def publish(self, event: GatewayEvent) -> None:
-        """Publish an event to all subscribers."""
+        """Publish an event to all subscribers.
+
+        A full subscriber queue drops its oldest event (counted on the
+        subscription) — a slow dashboard must never be silently
+        unsubscribed while its socket stays open.
+        """
         self._history.append(event)
         if len(self._history) > self._max_history:
             self._history = self._history[-self._max_history :]
 
         logger.info(f"📡 Event: {event.type.value} | {event.message or ''}")
 
-        dead_queues = []
-        for queue, type_filter in self._subscribers:
-            # Skip if subscriber has a type filter and this event doesn't match
-            if type_filter and event.type not in type_filter:
+        for sub in self._subscribers:
+            if sub.type_filter and event.type not in sub.type_filter:
                 continue
-            try:
-                queue.put_nowait(event)
-            except asyncio.QueueFull:
-                dead_queues.append((queue, type_filter))
-
-        for entry in dead_queues:
-            self._subscribers.remove(entry)
+            sub.deliver(event)
 
     def subscribe(
         self,
         max_size: int = 100,
         event_types: Optional[set[EventType]] = None,
+        replay_last: int = 0,
     ) -> "EventSubscription":
         """
         Create a new subscription.
 
         Args:
-            max_size: Queue depth before subscriber is considered dead.
+            max_size: Queue depth; overflow drops the oldest event.
             event_types: Optional filter — only receive these event types.
                          None means receive everything.
+            replay_last: Seed the queue with up to N most recent
+                         history events (post-filter) before live ones.
 
         Returns:
             An async iterator of GatewayEvents.
         """
         queue: asyncio.Queue[GatewayEvent] = asyncio.Queue(maxsize=max_size)
-        entry = (queue, event_types)
-        self._subscribers.append(entry)
-        return EventSubscription(queue, self, entry)
+        sub = EventSubscription(queue, self, event_types)
+        if replay_last > 0:
+            replayable = [
+                e for e in self._history
+                if not event_types or e.type in event_types
+            ]
+            for event in replayable[-replay_last:]:
+                sub.deliver(event)
+        self._subscribers.append(sub)
+        return sub
 
-    def unsubscribe(self, entry: tuple) -> None:
+    def unsubscribe(self, sub: "EventSubscription") -> None:
         """Remove a subscriber."""
-        if entry in self._subscribers:
-            self._subscribers.remove(entry)
+        if sub in self._subscribers:
+            self._subscribers.remove(sub)
 
     @property
     def recent_events(self) -> list[GatewayEvent]:
@@ -95,11 +102,28 @@ class EventSubscription:
         self,
         queue: asyncio.Queue[GatewayEvent],
         bus: EventBus,
-        entry: tuple,
+        type_filter: Optional[set[EventType]] = None,
     ):
         self._queue = queue
         self._bus = bus
-        self._entry = entry
+        self.type_filter = type_filter
+        self.dropped = 0  # events lost to queue overflow
+
+    def deliver(self, event: GatewayEvent) -> None:
+        """Enqueue an event, dropping the oldest on overflow."""
+        try:
+            self._queue.put_nowait(event)
+        except asyncio.QueueFull:
+            try:
+                self._queue.get_nowait()
+                self._queue.put_nowait(event)
+            except (asyncio.QueueEmpty, asyncio.QueueFull):
+                pass
+            self.dropped += 1
+            if self.dropped in (1, 10, 100) or self.dropped % 1000 == 0:
+                logger.warning(
+                    f"📡 Slow subscriber: {self.dropped} events dropped"
+                )
 
     def __aiter__(self):
         return self
@@ -108,7 +132,7 @@ class EventSubscription:
         try:
             return await self._queue.get()
         except asyncio.CancelledError:
-            self._bus.unsubscribe(self._entry)
+            self._bus.unsubscribe(self)
             raise
 
     async def get(self, timeout: Optional[float] = None) -> GatewayEvent:
@@ -117,4 +141,4 @@ class EventSubscription:
 
     def close(self):
         """Unsubscribe from the event bus."""
-        self._bus.unsubscribe(self._entry)
+        self._bus.unsubscribe(self)

core/gateway.py

@@ -33,32 +33,43 @@ def build_sip_engine(
     on_device_registered: Callable,
     on_incoming_call: Callable,
 ) -> SIPEngine:
-    """Build the appropriate SIP engine from config."""
+    """
+    Build the SIP engine from config.
+
+    The mock engine must be requested explicitly (USE_MOCK_SIP=true).
+    An unconfigured trunk or a failed SippyEngine construction raises —
+    the caller fails startup rather than running a gateway that can't
+    place real calls while reporting healthy.
+    """
+    if settings.use_mock_sip:
+        logger.warning("🧪 USE_MOCK_SIP=true — SIP engine is a mock, no real calls")
+        return MockSIPEngine()
+
     trunk = settings.sip_trunk
     gw_sip = settings.gateway_sip
 
-    if trunk.host and trunk.host != "sip.provider.com":
-        # Real trunk configured — use Sippy B2BUA
-        try:
-            return SippyEngine(
-                sip_address=gw_sip.host,
-                sip_port=gw_sip.port,
-                trunk_host=trunk.host,
-                trunk_port=trunk.port,
-                trunk_username=trunk.username,
-                trunk_password=trunk.password.get_secret_value(),
-                trunk_transport=trunk.transport,
-                domain=gw_sip.domain,
-                did=trunk.did,
-                media_pipeline=media_pipeline,
-                on_leg_state_change=on_leg_state_change,
-                on_device_registered=on_device_registered,
-                on_incoming_call=on_incoming_call,
-            )
-        except Exception as e:
-            logger.warning(f"Could not create SippyEngine: {e} — using mock")
+    if not trunk.host or trunk.host in ("sip.provider.com", "sip.yourprovider.com"):
+        raise RuntimeError(
+            "SIP trunk is not configured (SIP_TRUNK_HOST is unset or a "
+            "placeholder). Set SIP_TRUNK_* in .env, or set USE_MOCK_SIP=true "
+            "for development without a trunk."
+        )
 
-    return MockSIPEngine()
+    return SippyEngine(
+        sip_address=gw_sip.host,
+        sip_port=gw_sip.port,
+        trunk_host=trunk.host,
+        trunk_port=trunk.port,
+        trunk_username=trunk.username,
+        trunk_password=trunk.password.get_secret_value(),
+        trunk_transport=trunk.transport,
+        domain=gw_sip.domain,
+        did=trunk.did,
+        media_pipeline=media_pipeline,
+        on_leg_state_change=on_leg_state_change,
+        on_device_registered=on_device_registered,
+        on_incoming_call=on_incoming_call,
+    )
 
 
 class AIPSTNGateway:

core/sippy_engine.py

@@ -50,7 +50,6 @@ class SipCallLeg:
         self.state = "init"  # init, trying, ringing, connected, terminated
         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):
         return f"<SipCallLeg {self.leg_id} {self.direction} {self.state} → {self.remote_uri}>"
@@ -297,9 +296,8 @@ class SippyEngine(SIPEngine):
             ]
 
         elif kind == "dtmf":
-            leg = self._legs.get(data["leg_id"])
-            if leg:
-                leg.dtmf_buffer.append(data["digit"])
+            # Received DTMF has no consumer yet; log until one exists
+            logger.info(f"  DTMF '{data['digit']}' received on {data['leg_id']}")
 
         elif kind == "trunk_registered":
             self._trunk_registered = data["registered"]