diff --git a/.env.example b/.env.example index 25f0a69..3cb31b6 100644 --- a/.env.example +++ b/.env.example @@ -12,6 +12,9 @@ DATABASE_URL=postgresql+asyncpg://holdslayer:@localhost:5432/holdsl API_TOKEN= # --- SIP Trunk --- +# The mock engine must be requested explicitly; an unconfigured trunk +# without USE_MOCK_SIP=true refuses to start. +USE_MOCK_SIP=false SIP_TRUNK_HOST=sip.yourprovider.com SIP_TRUNK_PORT=5060 SIP_TRUNK_USERNAME=your_sip_username diff --git a/api/websocket.py b/api/websocket.py index 85bef08..9673dfd 100644 --- a/api/websocket.py +++ b/api/websocket.py @@ -94,7 +94,7 @@ async def event_stream(websocket: WebSocket): # Immediately push current trunk status so the dashboard doesn't start blank await _send_trunk_status(websocket, gateway) - subscription = gateway.event_bus.subscribe() + subscription = gateway.event_bus.subscribe(replay_last=25) try: async for event in subscription: diff --git a/config.py b/config.py index f637bca..eee0626 100644 --- a/config.py +++ b/config.py @@ -11,7 +11,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict class SIPTrunkSettings(BaseSettings): """SIP trunk provider configuration.""" - model_config = SettingsConfigDict(env_prefix="SIP_TRUNK_") + model_config = SettingsConfigDict(env_prefix="SIP_TRUNK_", env_file=".env", extra="ignore") host: str = "sip.provider.com" port: int = 5060 @@ -24,7 +24,7 @@ class SIPTrunkSettings(BaseSettings): class GatewaySIPSettings(BaseSettings): """Gateway SIP listener for device registration.""" - model_config = SettingsConfigDict(env_prefix="GATEWAY_SIP_") + model_config = SettingsConfigDict(env_prefix="GATEWAY_SIP_", env_file=".env", extra="ignore") host: str = "0.0.0.0" port: int = 5060 @@ -34,7 +34,7 @@ class GatewaySIPSettings(BaseSettings): class SpeachesSettings(BaseSettings): """Speaches STT service configuration.""" - model_config = SettingsConfigDict(env_prefix="SPEACHES_") + model_config = SettingsConfigDict(env_prefix="SPEACHES_", env_file=".env", extra="ignore") url: str = "http://localhost:22070" model: str = "whisper-large-v3" @@ -43,7 +43,7 @@ class SpeachesSettings(BaseSettings): class ClassifierSettings(BaseSettings): """Audio classifier thresholds.""" - model_config = SettingsConfigDict(env_prefix="CLASSIFIER_") + model_config = SettingsConfigDict(env_prefix="CLASSIFIER_", env_file=".env", extra="ignore") music_threshold: float = 0.7 speech_threshold: float = 0.6 @@ -54,7 +54,7 @@ class ClassifierSettings(BaseSettings): class LLMSettings(BaseSettings): """LLM service configuration (OpenAI-compatible API).""" - model_config = SettingsConfigDict(env_prefix="LLM_") + model_config = SettingsConfigDict(env_prefix="LLM_", env_file=".env", extra="ignore") base_url: str = "http://localhost:11434/v1" model: str = "llama3" @@ -67,7 +67,7 @@ class LLMSettings(BaseSettings): class HoldSlayerSettings(BaseSettings): """Hold Slayer behavior settings.""" - model_config = SettingsConfigDict(env_prefix="HOLD_SLAYER_", env_prefix_allow_empty=True) + model_config = SettingsConfigDict(env_prefix="HOLD_SLAYER_", env_prefix_allow_empty=True, env_file=".env", extra="ignore") default_transfer_device: str = Field( default="sip_phone", validation_alias="DEFAULT_TRANSFER_DEVICE" @@ -79,7 +79,7 @@ class HoldSlayerSettings(BaseSettings): class TTSSettings(BaseSettings): """Rhema TTS service configuration (OpenAI-compatible /v1/audio/speech).""" - model_config = SettingsConfigDict(env_prefix="TTS_") + model_config = SettingsConfigDict(env_prefix="TTS_", env_file=".env", extra="ignore") base_url: str = "http://localhost:8000" model: str = "speaches-ai/Kokoro-82M-v1.0-ONNX" @@ -92,7 +92,7 @@ class TTSSettings(BaseSettings): class ReceptionistSettings(BaseSettings): """AI Receptionist behavior settings.""" - model_config = SettingsConfigDict(env_prefix="RECEPTIONIST_") + model_config = SettingsConfigDict(env_prefix="RECEPTIONIST_", env_file=".env", extra="ignore") enabled: bool = True greeting_template: str = ( @@ -133,6 +133,11 @@ class Settings(BaseSettings): # Outbound-call safety cap (REST + MCP make_call) max_concurrent_calls: int = 4 + # Explicit engine mode โ€” the mock engine must be asked for. An + # unconfigured trunk without this flag fails startup instead of + # silently degrading to a gateway that can't place real calls. + use_mock_sip: bool = False + # Notifications notify_sms_number: str = "" diff --git a/core/event_bus.py b/core/event_bus.py index f6270a7..00715ce 100644 --- a/core/event_bus.py +++ b/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) diff --git a/core/gateway.py b/core/gateway.py index 327e6a1..608f058 100644 --- a/core/gateway.py +++ b/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: diff --git a/core/sippy_engine.py b/core/sippy_engine.py index 4c56a83..aeaf6b0 100644 --- a/core/sippy_engine.py +++ b/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"" @@ -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"] diff --git a/db/database.py b/db/database.py index c93d658..5ddd6cf 100644 --- a/db/database.py +++ b/db/database.py @@ -85,24 +85,6 @@ class StoredCallFlow(Base): return f"" -class Contact(Base): - __tablename__ = "contacts" - - id = Column(String, primary_key=True) - name = Column(String, nullable=False) - phone_numbers = Column(JSON, nullable=False) # [{number, label, primary}, ...] - category = Column(String) # personal / business / service - routing_preference = Column(String, nullable=True) # how to handle their calls - notes = Column(Text, nullable=True) - call_count = Column(Integer, default=0) - last_call = Column(DateTime, nullable=True) - created_at = Column(DateTime, default=func.now()) - updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) - - def __repr__(self) -> str: - return f"" - - class Device(Base): __tablename__ = "devices" diff --git a/main.py b/main.py index 5a6978d..dd11fc7 100644 --- a/main.py +++ b/main.py @@ -172,18 +172,23 @@ async def lifespan(app: FastAPI): gateway.register_mode_handler(CallMode.HOLD_SLAYER, launch_hold_slayer) - gateway.sip_engine = build_sip_engine( - settings, - gateway.media_pipeline, - on_leg_state_change=gateway._on_sip_leg_state, - on_device_registered=gateway._on_sip_device_registered, - on_incoming_call=receptionist.on_inbound_call, - ) + try: + gateway.sip_engine = build_sip_engine( + settings, + gateway.media_pipeline, + on_leg_state_change=gateway._on_sip_leg_state, + on_device_registered=gateway._on_sip_device_registered, + on_incoming_call=receptionist.on_inbound_call, + ) + except Exception as e: + logger.critical(f"\nโŒ SIP engine failed to initialize:\n {e}") + sys.exit(1) await routing_svc.start() await gateway.start() app.state.gateway = gateway app.state.routing_service = routing_svc + app.state.transcription_service = transcription notification_svc = NotificationService(gateway.event_bus, settings) await notification_svc.start() @@ -298,21 +303,67 @@ async def root(): @app.get("/health", tags=["System"]) async def health(): - """Health check endpoint.""" + """ + Health check. "healthy" means the gateway can actually do its job: + real engine, registered trunk, reachable database. A mock engine or + a failing dependency reports "degraded" with the reason visible. + """ + from core.sip_engine import MockSIPEngine + from db.database import session_scope + gateway = getattr(app.state, "gateway", None) ready = gateway is not None and await gateway.sip_engine.is_ready() trunk_status = await gateway.sip_engine.get_trunk_status() if gateway else {"registered": False} - return { - "status": "healthy" if ready else "degraded", + engine_mode = ( + "mock" if gateway is None or isinstance(gateway.sip_engine, MockSIPEngine) + else "sippy" + ) + + db_ok = False + db_error = None + try: + from sqlalchemy import text + async with session_scope() as session: + await session.execute(text("SELECT 1")) + db_ok = True + except Exception as e: + db_error = str(e)[:200] + + healthy = ( + ready + and db_ok + and engine_mode == "sippy" + and trunk_status.get("registered", False) + ) + + checks = { "gateway": "ready" if gateway else "not initialized", + "engine": engine_mode, "sip_engine": "ready" if ready else "not ready", + "database": "ok" if db_ok else f"error: {db_error}", "sip_trunk": { "registered": trunk_status.get("registered", False), "host": trunk_status.get("host"), - "mock": trunk_status.get("mock", False), "reason": trunk_status.get("reason"), }, } + if gateway is not None: + tts = getattr(gateway, "_tts", None) + checks["tts"] = _availability(tts) + transcription = getattr(app.state, "transcription_service", None) + checks["stt"] = _availability(transcription) + + return {"status": "healthy" if healthy else "degraded", **checks} + + +def _availability(service) -> str: + """Last-known reachability of an HTTP leaf service.""" + if service is None: + return "not attached" + available = getattr(service, "available", None) + if available is None: + return "unknown (no requests yet)" + return "ok" if available else "unreachable" if __name__ == "__main__": diff --git a/models/contact.py b/models/contact.py deleted file mode 100644 index 9a0e973..0000000 --- a/models/contact.py +++ /dev/null @@ -1,60 +0,0 @@ -""" -Contact models โ€” People and organizations you call. -""" - -from datetime import datetime -from typing import Optional - -from pydantic import BaseModel, Field - - -class PhoneNumber(BaseModel): - """A phone number associated with a contact.""" - - number: str # E.164 format - label: str = "main" # main, mobile, work, home, fax, etc. - primary: bool = False - - -class ContactBase(BaseModel): - """Shared contact fields.""" - - name: str - phone_numbers: list[PhoneNumber] - category: Optional[str] = None # personal / business / service - routing_preference: Optional[str] = None # how to handle their calls - notes: Optional[str] = None - - -class Contact(ContactBase): - """Full contact model.""" - - id: str - call_count: int = 0 - last_call: Optional[datetime] = None - created_at: Optional[datetime] = None - updated_at: Optional[datetime] = None - - @property - def primary_number(self) -> Optional[str]: - """Get the primary phone number.""" - for pn in self.phone_numbers: - if pn.primary: - return pn.number - return self.phone_numbers[0].number if self.phone_numbers else None - - -class ContactCreate(ContactBase): - """Request model for creating a contact.""" - - pass - - -class ContactUpdate(BaseModel): - """Request model for updating a contact.""" - - name: Optional[str] = None - phone_numbers: Optional[list[PhoneNumber]] = None - category: Optional[str] = None - routing_preference: Optional[str] = None - notes: Optional[str] = None diff --git a/services/call_persistence.py b/services/call_persistence.py index af57d77..d12c8c5 100644 --- a/services/call_persistence.py +++ b/services/call_persistence.py @@ -7,6 +7,7 @@ surfaces can't drift. Every function takes an AsyncSession; callers own the transaction (get_db for REST, session_scope for MCP/services). """ +import asyncio import logging import uuid from datetime import datetime @@ -159,53 +160,67 @@ async def persist_call_on_end(call: ActiveCall, final_status: CallStatus) -> Non """Insert a CallRecord and any transcript chunks for `call`. Wired into CallManager as its on_call_ended hook by the - composition root in main.py. + composition root in main.py. Retries briefly โ€” losing the row + means the call never happened as far as history is concerned, so + the final failure logs at ERROR with the payload identifiers. """ - try: - async with session_scope() as session: - record = CallRecord( - id=call.id, - direction=call.direction, - remote_number=call.remote_number, - status=final_status.value, - mode=call.mode.value, - intent=call.intent, - started_at=call.started_at, - ended_at=datetime.now(), - duration=int(call.duration), - hold_time=int(call.hold_time), - device_used=call.device, - call_flow_id=call.call_flow_id, - classification_timeline=[ - { - "timestamp": c.timestamp, - "audio_type": c.audio_type.value, - "confidence": c.confidence, - } - for c in call.classification_history - ], - metadata_={"services": list(call.services)}, - ) - session.add(record) + for attempt in range(3): + try: + await _write_call_record(call, final_status) + return + except Exception as e: + if attempt == 2: + logger.error( + f"Call record lost: id={call.id} number={call.remote_number} " + f"status={final_status.value}: {e}" + ) + return + await asyncio.sleep(2**attempt) - # Each transcript chunk gets its own row with a sequence number - # so the dashboard can render them in order with click-to-seek. - for seq, text in enumerate(call.transcript_chunks): - speaker = "unknown" - payload = text - if ":" in text: - head, rest = text.split(":", 1) - head = head.strip().lower() - if head in {"caller", "agent", "receptionist", "caller_message"}: - speaker = head if head != "caller_message" else "caller" - payload = rest.strip() - session.add(TranscriptChunk( - id=f"tc_{uuid.uuid4().hex[:10]}", - call_id=call.id, - seq=seq, - t_offset_ms=0, - speaker=speaker, - text=payload, - )) - except Exception as e: - logger.warning(f"Could not persist call {call.id}: {e}") + +async def _write_call_record(call: ActiveCall, final_status: CallStatus) -> None: + async with session_scope() as session: + record = CallRecord( + id=call.id, + direction=call.direction, + remote_number=call.remote_number, + status=final_status.value, + mode=call.mode.value, + intent=call.intent, + started_at=call.started_at, + ended_at=datetime.now(), + duration=int(call.duration), + hold_time=int(call.hold_time), + device_used=call.device, + call_flow_id=call.call_flow_id, + classification_timeline=[ + { + "timestamp": c.timestamp, + "audio_type": c.audio_type.value, + "confidence": c.confidence, + } + for c in call.classification_history + ], + metadata_={"services": list(call.services)}, + ) + session.add(record) + + # Each transcript chunk gets its own row with a sequence number + # so the dashboard can render them in order with click-to-seek. + for seq, text in enumerate(call.transcript_chunks): + speaker = "unknown" + payload = text + if ":" in text: + head, rest = text.split(":", 1) + head = head.strip().lower() + if head in {"caller", "agent", "receptionist", "caller_message"}: + speaker = head if head != "caller_message" else "caller" + payload = rest.strip() + session.add(TranscriptChunk( + id=f"tc_{uuid.uuid4().hex[:10]}", + call_id=call.id, + seq=seq, + t_offset_ms=0, + speaker=speaker, + text=payload, + )) diff --git a/services/hold_slayer.py b/services/hold_slayer.py index 4d6031b..b92f781 100644 --- a/services/hold_slayer.py +++ b/services/hold_slayer.py @@ -56,6 +56,29 @@ class HoldSlayerService: self.settings = settings self.tts = tts + async def _service_error(self, call_id: str, service: str, error: Exception) -> None: + """Surface a failed dependency as a typed event, not silence.""" + logger.error(f"โš ๏ธ {service} failed for {call_id}: {error}") + try: + await self.gateway.event_bus.publish(GatewayEvent( + type=EventType.ERROR, + call_id=call_id, + data={"service": service, "error": str(error)}, + message=f"โš ๏ธ {service} failed: {error}", + )) + except Exception: + pass + + async def _transcribe( + self, call_id: str, audio: bytes, prompt: Optional[str] = None + ) -> str: + """Transcribe with an explicit empty-string fallback on failure.""" + try: + return await self.transcription.transcribe(audio, prompt=prompt) + except Exception as e: + await self._service_error(call_id, "transcription", e) + return "" + async def run( self, call: ActiveCall, @@ -309,9 +332,10 @@ class HoldSlayerService: AudioClassification.IVR_PROMPT, AudioClassification.LIVE_HUMAN, ): - transcript = await self.transcription.transcribe( + transcript = await self._transcribe( + call.id, audio_chunk, - prompt="Phone IVR menu, customer service, press 1 for..." + prompt="Phone IVR menu, customer service, press 1 for...", ) if transcript: await self.call_manager.add_transcript(call.id, transcript) @@ -429,7 +453,7 @@ class HoldSlayerService: # Check for human if result.audio_type == AudioClassification.LIVE_HUMAN: # Verify with transcription - transcript = await self.transcription.transcribe(audio_chunk) + transcript = await self._transcribe(call.id, audio_chunk) if transcript: await self.call_manager.add_transcript(call.id, transcript) # If we got meaningful speech, it's probably a real person @@ -491,7 +515,7 @@ class HoldSlayerService: continue # Transcribe - transcript = await self.transcription.transcribe(audio_chunk) + transcript = await self._transcribe(call.id, audio_chunk) if not transcript: continue @@ -545,7 +569,7 @@ class HoldSlayerService: AudioClassification.IVR_PROMPT, AudioClassification.LIVE_HUMAN, ): - text = await self.transcription.transcribe(audio_chunk) + text = await self._transcribe(call.id, audio_chunk) if text: transcript_parts.append(text) @@ -713,7 +737,11 @@ class HoldSlayerService: os.close(fd) try: - ok = await self.tts.synthesize_to_file(text, tmp_path) + try: + ok = await self.tts.synthesize_to_file(text, tmp_path) + except Exception as e: + await self._service_error(call.id, "tts", e) + return False if not ok: logger.warning(f"๐Ÿ—ฃ๏ธ TTS synthesis returned no audio for: '{text[:60]}'") return False diff --git a/services/llm_client.py b/services/llm_client.py index b754afa..667c12a 100644 --- a/services/llm_client.py +++ b/services/llm_client.py @@ -327,15 +327,14 @@ class LLMClient: except httpx.HTTPStatusError as e: self._total_errors += 1 logger.error(f"LLM API error: {e.response.status_code} {e.response.text[:200]}") - return "" + raise except httpx.TimeoutException: self._total_errors += 1 logger.error(f"LLM API timeout after {self.timeout}s") - return "" - except Exception as e: + raise + except Exception: self._total_errors += 1 - logger.error(f"LLM client error: {e}") - return "" + raise @staticmethod def _parse_json_response(text: str) -> dict[str, Any]: diff --git a/services/notification.py b/services/notification.py index 0a17999..c1d859d 100644 --- a/services/notification.py +++ b/services/notification.py @@ -67,7 +67,6 @@ class NotificationService: self._event_bus = event_bus self._settings = settings self._task: Optional[asyncio.Task] = None - self._sms_sender: Optional[Any] = None # Track what we've already notified (avoid spam) self._notified: dict[str, set[str]] = {} # call_id -> set of event types @@ -214,43 +213,3 @@ class NotificationService: # WebSocket notifications go through the event bus # (the WebSocket handler in the API reads from EventBus directly) - - # SMS for critical notifications - if ( - notification.priority == NotificationPriority.CRITICAL - and self._settings.notify_sms_number - ): - await self._send_sms(notification) - - async def _send_sms(self, notification: Notification) -> None: - """ - Send an SMS notification. - - Uses a simple HTTP-based SMS gateway. In production, - this would use Twilio, AWS SNS, or similar. - """ - phone = self._settings.notify_sms_number - if not phone: - return - - try: - import httpx - - # Generic webhook-based SMS (configure your provider) - # This is a placeholder โ€” wire up your preferred SMS provider - logger.info(f"๐Ÿ“ฑ SMS โ†’ {phone}: {notification.title}") - - # Example: Twilio-style API - # async with httpx.AsyncClient() as client: - # await client.post( - # "https://api.twilio.com/2010-04-01/Accounts/.../Messages.json", - # data={ - # "To": phone, - # "From": self._settings.sip_trunk.did, - # "Body": f"{notification.title}\n{notification.message}", - # }, - # auth=(account_sid, auth_token), - # ) - - except Exception as e: - logger.error(f"SMS send failed: {e}") diff --git a/services/receptionist.py b/services/receptionist.py index 1ecc071..8999b30 100644 --- a/services/receptionist.py +++ b/services/receptionist.py @@ -203,6 +203,19 @@ class ReceptionistService: # State machine steps # ---------------------------------------------------------------- + async def _service_error(self, call_id: str, service: str, error: Exception) -> None: + """Surface a failed dependency as a typed event, not silence.""" + logger.error(f"โš ๏ธ {service} failed for {call_id}: {error}") + try: + await self.gateway.event_bus.publish(GatewayEvent( + type=EventType.ERROR, + call_id=call_id, + data={"service": service, "error": str(error)}, + message=f"โš ๏ธ {service} failed: {error}", + )) + except Exception: + pass + async def _greet(self, call: ActiveCall, sip_leg_id: str) -> None: await self.gateway.event_bus.publish(GatewayEvent( type=EventType.RECEPTIONIST_GREETING, @@ -250,7 +263,11 @@ class ReceptionistService: if not audio or self.transcription is None: return "" - return await self.transcription.transcribe(bytes(audio)) + try: + return await self.transcription.transcribe(bytes(audio)) + except Exception as e: + await self._service_error(call.id, "transcription", e) + return "" async def _classify( self, @@ -291,7 +308,7 @@ class ReceptionistService: system=self.settings.llm_persona, ) except Exception as e: - logger.warning(f"Receptionist LLM classify failed: {e}") + await self._service_error(call.id, "llm", e) return { "intent": transcript, "urgency": "normal", @@ -304,10 +321,14 @@ class ReceptionistService: routing_decision: Optional[RoutingDecision], classification: dict, ) -> RoutingAction: - """Rules win on conflict; otherwise use the LLM's recommendation.""" - if routing_decision and routing_decision.action.type not in ( - RoutingActionType.TAKE_MESSAGE, - ): + """Rules win on conflict; otherwise use the LLM's recommendation. + + A decision counts as a rule only when one actually matched + (matched_rule_id set) โ€” the no-rule default is take_message and + must stay overridable by the LLM. A matched TAKE_MESSAGE rule + wins like any other rule. + """ + if routing_decision and routing_decision.matched_rule_id: return routing_decision.action recommended = (classification.get("recommended_action") or "ring").lower() @@ -347,36 +368,42 @@ class ReceptionistService: call.id, media_pipeline=media, leg_ids=[sip_leg_id] ) try: - await asyncio.sleep(self.settings.message_max_seconds) + # Record up to the cap, but stop early once the caller hangs + # up (leg termination ends the call via the leg-state wiring). + deadline = _time.monotonic() + self.settings.message_max_seconds + while _time.monotonic() < deadline: + await asyncio.sleep(1.0) + if self.gateway.call_manager.get_call(call.id) is None: + break finally: session = await recording_svc.stop_recording( call.id, media_pipeline=media ) - message_text = "" - rec_path = session.filepath_mixed if session else None - if rec_path and Path(rec_path).exists() and self.transcription is not None: - try: - audio_bytes = Path(rec_path).read_bytes() - message_text = await self.transcription.transcribe(audio_bytes) - except Exception as e: - logger.warning(f"Receptionist transcribe failed: {e}") + message_text = "" + rec_path = session.filepath_mixed if session else None + if rec_path and Path(rec_path).exists() and self.transcription is not None: + try: + audio_bytes = Path(rec_path).read_bytes() + message_text = await self.transcription.transcribe(audio_bytes) + except Exception as e: + await self._service_error(call.id, "transcription", e) - if message_text: - call.transcript_chunks.append(f"caller_message: {message_text}") + if message_text: + call.transcript_chunks.append(f"caller_message: {message_text}") - await self.gateway.event_bus.publish(GatewayEvent( - type=EventType.RECEPTIONIST_MESSAGE_SAVED, - call_id=call.id, - data={ - "path": rec_path, - "transcript": message_text, - "caller": call.remote_number, - }, - message=f"๐Ÿ“ฅ Message saved from {call.remote_number}", - )) + await self.gateway.event_bus.publish(GatewayEvent( + type=EventType.RECEPTIONIST_MESSAGE_SAVED, + call_id=call.id, + data={ + "path": rec_path, + "transcript": message_text, + "caller": call.remote_number, + }, + message=f"๐Ÿ“ฅ Message saved from {call.remote_number}", + )) - await self._hangup(call, sip_leg_id) + await self._hangup(call, sip_leg_id) # ---------------------------------------------------------------- # Helpers @@ -394,7 +421,11 @@ class ReceptionistService: fd, tmp_path = tempfile.mkstemp(suffix=".wav", prefix=f"recept_{call.id}_") os.close(fd) try: - ok = await tts.synthesize_to_file(text, tmp_path) + try: + ok = await tts.synthesize_to_file(text, tmp_path) + except Exception as e: + await self._service_error(call.id, "tts", e) + return if not ok: return await media.play_wav(sip_leg_id, tmp_path) diff --git a/services/recording.py b/services/recording.py index f80b551..375e306 100644 --- a/services/recording.py +++ b/services/recording.py @@ -91,6 +91,7 @@ class RecordingService: filepath_agent=filepath_agent, started_at=datetime.now(), sample_rate=self._sample_rate, + leg_ids=leg_ids, ) # Start PJSUA2 recording if media pipeline is available @@ -159,26 +160,38 @@ class RecordingService: @staticmethod async def _persist_recording(session: "RecordingSession") -> None: - """Write a recordings row for this session. Failures are non-fatal.""" - try: - import uuid as _uuid - from db.database import RecordingRecord, get_session_factory + """Write a recordings row for this session, with bounded retry. - async with get_session_factory()() as db: - db.add(RecordingRecord( - id=f"rec_{_uuid.uuid4().hex[:10]}", - call_id=session.call_id, - path=session.filepath_mixed or "", - format="wav", - duration_s=float(session.duration_seconds or 0), - size_bytes=int(session.file_size_bytes or 0), - channels=1, - started_at=session.started_at, - ended_at=session.stopped_at, - )) - await db.commit() - except Exception as e: - logger.warning(f"Recording persistence failed: {e}") + Non-fatal for the call, but a lost row means the dashboard can + never find the WAV โ€” so failures log at ERROR, not warning. + """ + import uuid as _uuid + + from db.database import RecordingRecord, session_scope + + for attempt in range(3): + try: + async with session_scope() as db: + db.add(RecordingRecord( + id=f"rec_{_uuid.uuid4().hex[:10]}", + call_id=session.call_id, + path=session.filepath_mixed or "", + format="wav", + duration_s=float(session.duration_seconds or 0), + size_bytes=int(session.file_size_bytes or 0), + channels=1, + started_at=session.started_at, + ended_at=session.stopped_at, + )) + return + except Exception as e: + if attempt == 2: + logger.error( + f"Recording row lost for {session.call_id} " + f"(path={session.filepath_mixed}): {e}" + ) + return + await asyncio.sleep(2 ** attempt) async def _recording_timeout(self, call_id: str) -> None: """Auto-stop recording after max duration.""" @@ -239,6 +252,7 @@ class RecordingSession: filepath_agent: Optional[str] = None, started_at: Optional[datetime] = None, sample_rate: int = 16000, + leg_ids: Optional[list[str]] = None, ): self.call_id = call_id self.filepath_mixed = filepath_mixed @@ -249,7 +263,7 @@ class RecordingSession: self.duration_seconds: Optional[int] = None self.file_size_bytes: Optional[int] = None self.sample_rate = sample_rate - self._leg_ids: list[str] = [] + self._leg_ids: list[str] = list(leg_ids or []) def to_dict(self) -> dict: return { diff --git a/services/transcription.py b/services/transcription.py index 183fed1..9a745cd 100644 --- a/services/transcription.py +++ b/services/transcription.py @@ -27,6 +27,8 @@ class TranscriptionService: def __init__(self, settings: SpeachesSettings): self.settings = settings self._client: Optional[httpx.AsyncClient] = None + # Last-known reachability, surfaced by /health (None = no requests yet) + self.available: Optional[bool] = None async def _get_client(self) -> httpx.AsyncClient: """Get or create the HTTP client.""" @@ -60,6 +62,9 @@ class TranscriptionService: # Convert raw PCM to WAV format for the API wav_data = self._pcm_to_wav(audio_data) + # Raises on failure โ€” callers decide the per-call fallback and + # publish a service-error event; swallowing here made a down + # Speaches look like "the AI is deciding badly". try: response = await client.post( "/v1/audio/transcriptions", @@ -72,44 +77,13 @@ class TranscriptionService: }, ) response.raise_for_status() - text = response.text.strip() - logger.debug(f"Transcription: '{text}'") - return text - - except httpx.HTTPStatusError as e: - logger.error(f"Speaches API error: {e.response.status_code} {e.response.text}") - return "" - except httpx.ConnectError: - logger.error(f"Cannot connect to Speaches at {self.settings.url}") - return "" - except Exception as e: - logger.error(f"Transcription failed: {e}") - return "" - - async def transcribe_stream( - self, - audio_data: bytes, - language: str = "en", - ): - """ - Stream transcription โ€” for real-time results. - - Uses Speaches streaming endpoint if available, - falls back to chunked transcription. - - Yields: - str: Partial transcription chunks - """ - # For now, do chunked transcription - # TODO: Implement WebSocket streaming when Speaches supports it - chunk_size = 16000 * 2 * 3 # 3 seconds of 16kHz 16-bit mono - - for i in range(0, len(audio_data), chunk_size): - chunk = audio_data[i:i + chunk_size] - if len(chunk) > 0: - text = await self.transcribe(chunk, language) - if text: - yield text + except Exception: + self.available = False + raise + self.available = True + text = response.text.strip() + logger.debug(f"Transcription: '{text}'") + return text async def close(self) -> None: """Close the HTTP client.""" diff --git a/services/tts.py b/services/tts.py index 2865c60..81e6e53 100644 --- a/services/tts.py +++ b/services/tts.py @@ -22,6 +22,8 @@ class TTSService: def __init__(self, settings: TTSSettings): self.settings = settings self._client: Optional[httpx.AsyncClient] = None + # Last-known reachability, surfaced by /health (None = no requests yet) + self.available: Optional[bool] = None async def _get_client(self) -> httpx.AsyncClient: if self._client is None or self._client.is_closed: @@ -54,19 +56,17 @@ class TTSService: "sample_rate": self.settings.sample_rate, } + # Raises on failure โ€” callers decide the per-call fallback and + # publish a service-error event; swallowing here made a down + # Rhema look like "the AI went quiet". try: response = await client.post("/v1/audio/speech", json=body) response.raise_for_status() - return response.content - except httpx.HTTPStatusError as e: - logger.error(f"Rhema TTS error: {e.response.status_code} {e.response.text}") - return b"" - except httpx.ConnectError: - logger.error(f"Cannot connect to Rhema at {self.settings.base_url}") - return b"" - except Exception as e: - logger.error(f"TTS synthesis failed: {e}") - return b"" + except Exception: + self.available = False + raise + self.available = True + return response.content async def synthesize_to_file( self, diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 570bae2..1cc1d05 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -102,14 +102,9 @@ class TestEngineEventFunnel: 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 diff --git a/tests/test_health_policy.py b/tests/test_health_policy.py new file mode 100644 index 0000000..4d6930a --- /dev/null +++ b/tests/test_health_policy.py @@ -0,0 +1,89 @@ +""" +Engine-mode and event-bus-integrity tests. + +The mock engine must be requested explicitly; a full subscriber queue +drops its oldest event but never loses the subscription; history is +replayable to late joiners. +""" + +import asyncio + +import pytest + +from config import Settings +from core.event_bus import EventBus +from core.gateway import build_sip_engine +from core.media_pipeline import MediaPipeline +from core.sip_engine import MockSIPEngine +from models.events import EventType, GatewayEvent + + +def _noop(*args, **kwargs): + pass + + +class TestEngineMode: + def _build(self, settings: Settings): + return build_sip_engine( + settings, + MediaPipeline(sample_rate=16000), + on_leg_state_change=_noop, + on_device_registered=_noop, + on_incoming_call=_noop, + ) + + def test_mock_engine_only_when_asked(self): + settings = Settings(use_mock_sip=True) + assert isinstance(self._build(settings), MockSIPEngine) + + def test_unconfigured_trunk_refuses_to_build(self): + settings = Settings(use_mock_sip=False) + settings.sip_trunk.host = "sip.yourprovider.com" + with pytest.raises(RuntimeError, match="not configured"): + self._build(settings) + + +def _event(i: int) -> GatewayEvent: + return GatewayEvent( + type=EventType.CALL_INITIATED, + call_id=f"call_{i}", + data={}, + message=f"event {i}", + ) + + +class TestEventBusIntegrity: + async def test_overflow_drops_oldest_keeps_subscription(self): + bus = EventBus() + sub = bus.subscribe(max_size=3) + + for i in range(5): + await bus.publish(_event(i)) + + assert bus.subscriber_count == 1 # never evicted + assert sub.dropped == 2 + received = [await asyncio.wait_for(sub.get(), 1.0) for _ in range(3)] + assert [e.call_id for e in received] == ["call_2", "call_3", "call_4"] + + async def test_replay_last_seeds_history(self): + bus = EventBus() + for i in range(10): + await bus.publish(_event(i)) + + sub = bus.subscribe(replay_last=3) + received = [await asyncio.wait_for(sub.get(), 1.0) for _ in range(3)] + assert [e.call_id for e in received] == ["call_7", "call_8", "call_9"] + + async def test_replay_respects_type_filter(self): + bus = EventBus() + await bus.publish(_event(1)) + await bus.publish(GatewayEvent( + type=EventType.HUMAN_DETECTED, call_id="call_h", data={}, message="x" + )) + + sub = bus.subscribe( + event_types={EventType.HUMAN_DETECTED}, replay_last=5 + ) + event = await asyncio.wait_for(sub.get(), 1.0) + assert event.call_id == "call_h" + assert sub._queue.empty() diff --git a/tests/test_receptionist.py b/tests/test_receptionist.py index d0b32f0..e90b923 100644 --- a/tests/test_receptionist.py +++ b/tests/test_receptionist.py @@ -24,10 +24,27 @@ class TestReceptionistDecide: gw = _make_gateway() svc = ReceptionistService(gw) rule_action = RoutingAction(type=RoutingActionType.REJECT, message="nope") - decision = RoutingDecision(action=rule_action, reason="rule said so") + decision = RoutingDecision( + action=rule_action, + matched_rule_id="rule_1", + matched_rule_name="block", + reason="rule said so", + ) chosen = svc._decide(decision, {"recommended_action": "ring"}) assert chosen.type == RoutingActionType.REJECT + def test_matched_take_message_rule_beats_llm(self): + gw = _make_gateway() + svc = ReceptionistService(gw) + decision = RoutingDecision( + action=RoutingAction(type=RoutingActionType.TAKE_MESSAGE), + matched_rule_id="rule_2", + matched_rule_name="voicemail-hours", + reason="matched rule 'voicemail-hours'", + ) + chosen = svc._decide(decision, {"recommended_action": "ring"}) + assert chosen.type == RoutingActionType.TAKE_MESSAGE + def test_falls_back_to_llm_when_rule_is_default_take_message(self): gw = _make_gateway() svc = ReceptionistService(gw) diff --git a/tests/test_services.py b/tests/test_services.py index ab9b5bf..fc54b4e 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -123,14 +123,14 @@ class TestLLMClient: assert result["key"] == "value" @pytest.mark.asyncio - async def test_chat_http_error_returns_empty(self): - """Verify HTTP errors return empty string gracefully.""" + async def test_chat_error_raises(self): + """Failures propagate to the caller (which owns the fallback).""" client = self._make_client() with patch.object(client._client, "post", new_callable=AsyncMock) as mock_post: mock_post.side_effect = Exception("Connection refused") - result = await client.chat("test", system="test") - assert result == "" + with pytest.raises(Exception, match="Connection refused"): + await client.chat("test", system="test") assert client._total_errors == 1 @pytest.mark.asyncio