diff --git a/api/call_flows.py b/api/call_flows.py index e5d4bf2..0790920 100644 --- a/api/call_flows.py +++ b/api/call_flows.py @@ -2,26 +2,21 @@ Call Flows API — Store and manage IVR navigation trees. The system gets smarter every time you call somewhere. +Thin HTTP layer over the shared data functions in call_persistence. """ -import uuid -from datetime import datetime - from fastapi import APIRouter, Depends, HTTPException from slugify import slugify -from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from api.deps import get_gateway -from core.gateway import AIPSTNGateway -from db.database import StoredCallFlow, get_db +from db.database import get_db from models.call_flow import ( CallFlow, CallFlowCreate, - CallFlowStep, CallFlowSummary, CallFlowUpdate, ) +from services import call_persistence as store router = APIRouter() @@ -34,39 +29,23 @@ async def create_call_flow( """Store a new call flow for a phone number.""" flow_id = slugify(flow.name) - # Check if ID already exists - existing = await db.execute( - select(StoredCallFlow).where(StoredCallFlow.id == flow_id) - ) - if existing.scalar_one_or_none(): + if await store.get_flow(db, flow_id): raise HTTPException( status_code=409, detail=f"Call flow '{flow_id}' already exists. Use PUT to update.", ) - db_flow = StoredCallFlow( - id=flow_id, + row = await store.create_flow( + db, + flow_id=flow_id, name=flow.name, phone_number=flow.phone_number, description=flow.description, steps=[s.model_dump() for s in flow.steps], tags=flow.tags, notes=flow.notes, - last_verified=datetime.now(), - ) - db.add(db_flow) - await db.flush() - - return CallFlow( - id=flow_id, - name=flow.name, - phone_number=flow.phone_number, - description=flow.description, - steps=flow.steps, - tags=flow.tags, - notes=flow.notes, - last_verified=datetime.now(), ) + return store.flow_to_model(row) @router.get("/", response_model=list[CallFlowSummary]) @@ -74,9 +53,7 @@ async def list_call_flows( db: AsyncSession = Depends(get_db), ): """List all stored call flows.""" - result = await db.execute(select(StoredCallFlow)) - rows = result.scalars().all() - + rows = await store.list_flows(db) return [ CallFlowSummary( id=row.id, @@ -100,26 +77,10 @@ async def get_call_flow( db: AsyncSession = Depends(get_db), ): """Get a stored call flow by ID.""" - result = await db.execute( - select(StoredCallFlow).where(StoredCallFlow.id == flow_id) - ) - row = result.scalar_one_or_none() + row = await store.get_flow(db, flow_id) if not row: raise HTTPException(status_code=404, detail=f"Call flow '{flow_id}' not found") - - return CallFlow( - id=row.id, - name=row.name, - phone_number=row.phone_number, - description=row.description or "", - steps=[CallFlowStep(**s) for s in row.steps], - tags=row.tags or [], - notes=row.notes, - avg_hold_time=row.avg_hold_time, - success_rate=row.success_rate, - last_used=row.last_used, - times_used=row.times_used or 0, - ) + return store.flow_to_model(row) @router.get("/by-number/{phone_number}", response_model=CallFlow) @@ -128,29 +89,13 @@ async def get_flow_for_number( db: AsyncSession = Depends(get_db), ): """Look up stored call flow by phone number.""" - result = await db.execute( - select(StoredCallFlow).where(StoredCallFlow.phone_number == phone_number) - ) - row = result.scalar_one_or_none() + row = await store.get_flow_by_number(db, phone_number) if not row: raise HTTPException( status_code=404, detail=f"No call flow found for {phone_number}", ) - - return CallFlow( - id=row.id, - name=row.name, - phone_number=row.phone_number, - description=row.description or "", - steps=[CallFlowStep(**s) for s in row.steps], - tags=row.tags or [], - notes=row.notes, - avg_hold_time=row.avg_hold_time, - success_rate=row.success_rate, - last_used=row.last_used, - times_used=row.times_used or 0, - ) + return store.flow_to_model(row) @router.put("/{flow_id}", response_model=CallFlow) @@ -160,10 +105,7 @@ async def update_call_flow( db: AsyncSession = Depends(get_db), ): """Update an existing call flow.""" - result = await db.execute( - select(StoredCallFlow).where(StoredCallFlow.id == flow_id) - ) - row = result.scalar_one_or_none() + row = await store.get_flow(db, flow_id) if not row: raise HTTPException(status_code=404, detail=f"Call flow '{flow_id}' not found") @@ -181,20 +123,7 @@ async def update_call_flow( row.last_verified = update.last_verified await db.flush() - - return CallFlow( - id=row.id, - name=row.name, - phone_number=row.phone_number, - description=row.description or "", - steps=[CallFlowStep(**s) for s in row.steps], - tags=row.tags or [], - notes=row.notes, - avg_hold_time=row.avg_hold_time, - success_rate=row.success_rate, - last_used=row.last_used, - times_used=row.times_used or 0, - ) + return store.flow_to_model(row) @router.delete("/{flow_id}") @@ -203,10 +132,7 @@ async def delete_call_flow( db: AsyncSession = Depends(get_db), ): """Delete a stored call flow.""" - result = await db.execute( - select(StoredCallFlow).where(StoredCallFlow.id == flow_id) - ) - row = result.scalar_one_or_none() + row = await store.get_flow(db, flow_id) if not row: raise HTTPException(status_code=404, detail=f"Call flow '{flow_id}' not found") diff --git a/api/call_history.py b/api/call_history.py index 7105945..59edbc4 100644 --- a/api/call_history.py +++ b/api/call_history.py @@ -1,22 +1,18 @@ """ Call History API — Read-only access to persisted call records, transcript chunks, and recording files for the dashboard. +Thin HTTP layer over the shared data functions in call_persistence. """ +import os from datetime import datetime -from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import FileResponse -from sqlalchemy import desc, select from sqlalchemy.ext.asyncio import AsyncSession -from db.database import ( - CallRecord, - RecordingRecord, - TranscriptChunk, - get_db, -) +from db.database import get_db +from services import call_persistence as store router = APIRouter() @@ -25,24 +21,22 @@ router = APIRouter() async def list_history( limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0), - number: Optional[str] = None, - status: Optional[str] = None, - since: Optional[datetime] = None, - until: Optional[datetime] = None, + number: str | None = None, + status: str | None = None, + since: datetime | None = None, + until: datetime | None = None, db: AsyncSession = Depends(get_db), ): """Paged list of past calls, newest first.""" - stmt = select(CallRecord).order_by(desc(CallRecord.started_at)) - if number: - stmt = stmt.where(CallRecord.remote_number == number) - if status: - stmt = stmt.where(CallRecord.status == status) - if since: - stmt = stmt.where(CallRecord.started_at >= since) - if until: - stmt = stmt.where(CallRecord.started_at <= until) - - rows = (await db.execute(stmt.offset(offset).limit(limit))).scalars().all() + rows = await store.search_history( + db, + number=number, + status=status, + since=since, + until=until, + limit=limit, + offset=offset, + ) return [ { "id": r.id, @@ -65,9 +59,7 @@ async def list_history( @router.get("/{call_id}/record") async def get_record(call_id: str, db: AsyncSession = Depends(get_db)): """Full CallRecord with classification_timeline.""" - row = (await db.execute( - select(CallRecord).where(CallRecord.id == call_id) - )).scalar_one_or_none() + row = await store.get_record(db, call_id) if not row: raise HTTPException(status_code=404, detail=f"Call {call_id} not found") return { @@ -93,11 +85,7 @@ async def get_record(call_id: str, db: AsyncSession = Depends(get_db)): @router.get("/{call_id}/transcript") async def get_transcript(call_id: str, db: AsyncSession = Depends(get_db)): """Ordered transcript chunks for a call.""" - rows = (await db.execute( - select(TranscriptChunk) - .where(TranscriptChunk.call_id == call_id) - .order_by(TranscriptChunk.seq) - )).scalars().all() + rows = await store.get_transcript_chunks(db, call_id) return [ { "seq": c.seq, @@ -113,14 +101,9 @@ async def get_transcript(call_id: str, db: AsyncSession = Depends(get_db)): @router.get("/{call_id}/recording") async def get_recording(call_id: str, db: AsyncSession = Depends(get_db)): """Stream the WAV recording for a call.""" - row = (await db.execute( - select(RecordingRecord) - .where(RecordingRecord.call_id == call_id) - .order_by(desc(RecordingRecord.started_at)) - )).scalar_one_or_none() + row = await store.latest_recording(db, call_id) if not row or not row.path: raise HTTPException(status_code=404, detail="Recording not found") - import os if not os.path.exists(row.path): raise HTTPException(status_code=404, detail="Recording file missing on disk") return FileResponse(row.path, media_type="audio/wav", filename=os.path.basename(row.path)) diff --git a/api/calls.py b/api/calls.py index a612fcc..eb46c5d 100644 --- a/api/calls.py +++ b/api/calls.py @@ -172,10 +172,9 @@ async def send_dtmf( if not call: raise HTTPException(status_code=404, detail=f"Call {call_id} not found") - # Find the PSTN leg for this call - for leg_id, cid in gateway.call_manager._call_legs.items(): - if cid == call_id: - await gateway.sip_engine.send_dtmf(leg_id, digits) - return {"status": "sent", "digits": digits} + legs = gateway.call_manager.legs_for_call(call_id) + if not legs: + raise HTTPException(status_code=409, detail="No active SIP leg found for this call") - raise HTTPException(status_code=500, detail="No active SIP leg found for this call") + await gateway.sip_engine.send_dtmf(legs[0], digits) + return {"status": "sent", "digits": digits} diff --git a/api/deps.py b/api/deps.py index a8beedb..9698c2f 100644 --- a/api/deps.py +++ b/api/deps.py @@ -18,6 +18,14 @@ def get_gateway(request: Request) -> AIPSTNGateway: return gateway +def get_routing_service(request: Request): + """Get the routing service from app state.""" + routing = getattr(request.app.state, "routing_service", None) + if routing is None: + raise HTTPException(status_code=503, detail="Routing service not ready") + return routing + + def require_token(authorization: str | None = Header(default=None)) -> None: """ Enforce the static bearer token (API_TOKEN) on REST routes. diff --git a/api/routing.py b/api/routing.py index 9f7330e..614871e 100644 --- a/api/routing.py +++ b/api/routing.py @@ -6,7 +6,7 @@ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from api.deps import get_gateway +from api.deps import get_gateway, get_routing_service from core.gateway import AIPSTNGateway from db.database import Device as DeviceDB from db.database import get_db @@ -15,36 +15,31 @@ from models.routing import ( RoutingRuleCreate, RoutingRuleUpdate, ) +from services.routing import RoutingService router = APIRouter() @router.get("/rules", response_model=list[RoutingRule]) -async def list_rules(gateway: AIPSTNGateway = Depends(get_gateway)): - if gateway._routing is None: - raise HTTPException(status_code=503, detail="Routing service not ready") - return sorted(gateway._routing.rules, key=lambda r: (r.priority, r.id)) +async def list_rules(routing: RoutingService = Depends(get_routing_service)): + return sorted(routing.rules, key=lambda r: (r.priority, r.id)) @router.post("/rules", response_model=RoutingRule, status_code=201) async def create_rule( payload: RoutingRuleCreate, - gateway: AIPSTNGateway = Depends(get_gateway), + routing: RoutingService = Depends(get_routing_service), ): - if gateway._routing is None: - raise HTTPException(status_code=503, detail="Routing service not ready") - return await gateway._routing.create_rule(payload) + return await routing.create_rule(payload) @router.put("/rules/{rule_id}", response_model=RoutingRule) async def update_rule( rule_id: str, payload: RoutingRuleUpdate, - gateway: AIPSTNGateway = Depends(get_gateway), + routing: RoutingService = Depends(get_routing_service), ): - if gateway._routing is None: - raise HTTPException(status_code=503, detail="Routing service not ready") - rule = await gateway._routing.update_rule(rule_id, payload) + rule = await routing.update_rule(rule_id, payload) if rule is None: raise HTTPException(status_code=404, detail=f"Rule {rule_id} not found") return rule @@ -53,11 +48,9 @@ async def update_rule( @router.delete("/rules/{rule_id}") async def delete_rule( rule_id: str, - gateway: AIPSTNGateway = Depends(get_gateway), + routing: RoutingService = Depends(get_routing_service), ): - if gateway._routing is None: - raise HTTPException(status_code=503, detail="Routing service not ready") - ok = await gateway._routing.delete_rule(rule_id) + ok = await routing.delete_rule(rule_id) if not ok: raise HTTPException(status_code=404, detail=f"Rule {rule_id} not found") return {"status": "deleted", "rule_id": rule_id} diff --git a/core/call_manager.py b/core/call_manager.py index 3ea6ee4..1640996 100644 --- a/core/call_manager.py +++ b/core/call_manager.py @@ -26,11 +26,11 @@ class CallManager: The single source of truth for what's happening on the gateway. """ - def __init__(self, event_bus: EventBus): + def __init__(self, event_bus: EventBus, on_call_ended=None): self.event_bus = event_bus self._active_calls: dict[str, ActiveCall] = {} self._call_legs: dict[str, str] = {} # SIP leg ID -> call ID mapping - self._on_call_ended = None # async callback(call: ActiveCall, final_status) + self._on_call_ended = on_call_ended # async callback(call, final_status) # ================================================================ # Call Lifecycle diff --git a/core/gateway.py b/core/gateway.py index da3178b..327e6a1 100644 --- a/core/gateway.py +++ b/core/gateway.py @@ -1,16 +1,18 @@ """ -AI PSTN Gateway — The main orchestrator. +AI PSTN Gateway — call operations and device registry. -Ties together SIP engine, call manager, event bus, and all services. -This is the top-level object that FastAPI and MCP talk to. +The application service that FastAPI and MCP talk to for live-call +work. Composition happens in main.py's lifespan: services are built +there and attached; this module never imports from services/. """ import asyncio import logging +from collections.abc import Callable from datetime import datetime from typing import Optional -from config import Settings, get_settings +from config import Settings from core.call_manager import CallManager from core.dial_plan import is_emergency_number, next_extension from core.event_bus import EventBus @@ -18,28 +20,19 @@ from core.media_pipeline import MediaPipeline from core.sip_engine import MockSIPEngine, SIPEngine from core.sippy_engine import SippyEngine from models.call import ActiveCall, CallMode, CallStatus -from models.call_flow import CallFlow from models.device import Device, DeviceType from models.events import EventType, GatewayEvent logger = logging.getLogger(__name__) -def _extract_number(sip_uri: str) -> str: - """Pull the user part out of a SIP URI (sip:+15551212@host → +15551212).""" - if not sip_uri: - return "" - s = sip_uri.strip() - if s.startswith("<") and ">" in s: - s = s[1:s.index(">")] - if s.startswith("sip:"): - s = s[4:] - if "@" in s: - s = s.split("@", 1)[0] - return s - - -def _build_sip_engine(settings: Settings, gateway: "AIPSTNGateway") -> SIPEngine: +def build_sip_engine( + settings: Settings, + media_pipeline: MediaPipeline, + on_leg_state_change: Callable, + on_device_registered: Callable, + on_incoming_call: Callable, +) -> SIPEngine: """Build the appropriate SIP engine from config.""" trunk = settings.sip_trunk gw_sip = settings.gateway_sip @@ -57,10 +50,10 @@ def _build_sip_engine(settings: Settings, gateway: "AIPSTNGateway") -> SIPEngine trunk_transport=trunk.transport, 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, + 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") @@ -72,35 +65,31 @@ class AIPSTNGateway: """ The AI PSTN Gateway. - Central coordination point for: - - SIP engine (signaling + media) - - Call manager (state + events) - - Hold Slayer service - - Audio classifier - - Transcription service - - Device management + Owns live-call operations (make/transfer/hangup), the device + registry, and per-call background tasks. Services are attached by + the composition root; mode handlers launch per-call services + (hold slayer) without the gateway knowing their types. """ def __init__( self, settings: Settings, sip_engine: Optional[SIPEngine] = None, + on_call_ended=None, ): self.settings = settings self.event_bus = EventBus() - self.call_manager = CallManager(self.event_bus) + self.call_manager = CallManager(self.event_bus, on_call_ended=on_call_ended) self.media_pipeline = MediaPipeline(sample_rate=16000) self.sip_engine: SIPEngine = sip_engine or MockSIPEngine() - # Services (initialized in start()) - self._hold_slayer = None - self._audio_classifier = None - self._transcription = None + # Attached by the composition root (attach_services) self._tts = None - self._routing = None - self._receptionist = None - # Device registry (loaded from DB on start) + # Per-call-mode launchers registered by the composition root + self._mode_handlers: dict[CallMode, Callable] = {} + + # Device registry self._devices: dict[str, Device] = {} # Background tasks (per-call services, receptionist sessions) — @@ -117,23 +106,20 @@ class AIPSTNGateway: 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.""" - settings = get_settings() - gw = cls(settings=settings) - if sip_engine is not None: - gw.sip_engine = sip_engine - else: - gw.sip_engine = _build_sip_engine(settings, gw) - return gw + def attach_services(self, tts=None) -> None: + """Attach shared services the gateway must manage on shutdown.""" + self._tts = tts + + def register_mode_handler(self, mode: CallMode, handler: Callable) -> None: + """Register a launcher called as handler(call, sip_leg_id, call_flow_id).""" + self._mode_handlers[mode] = handler # ================================================================ # Lifecycle # ================================================================ async def start(self) -> None: - """Boot the gateway — start SIP engine and services.""" + """Boot the gateway — start media pipeline and SIP engine.""" logger.info("🔥 Starting AI PSTN Gateway...") # Start media pipeline first so SIP engine can hand it RTP streams @@ -141,26 +127,7 @@ class AIPSTNGateway: # Start SIP engine await self.sip_engine.start() - logger.info(f" SIP Engine: ready") - - # Import services here to avoid circular imports - from services.audio_classifier import AudioClassifier - from services.transcription import TranscriptionService - from services.tts import TTSService - from services.routing import RoutingService - from services.receptionist import ReceptionistService - - self._audio_classifier = AudioClassifier(self.settings.classifier) - self._transcription = TranscriptionService(self.settings.speaches) - self._tts = TTSService(self.settings.tts) - self._routing = RoutingService(self) - await self._routing.start() - self._receptionist = ReceptionistService(self) - - # Persist completed calls to the database for history/playback. - from services.call_persistence import persist_call_on_end - - self.call_manager._on_call_ended = persist_call_on_end + logger.info(" SIP Engine: ready") self._started_at = datetime.now() @@ -283,24 +250,10 @@ class AIPSTNGateway: await self.call_manager.update_status(call.id, CallStatus.FAILED) raise - # If hold_slayer mode, launch the Hold Slayer service - if mode == CallMode.HOLD_SLAYER: - from services.hold_slayer import HoldSlayerService - - hold_slayer = HoldSlayerService( - gateway=self, - call_manager=self.call_manager, - sip_engine=self.sip_engine, - classifier=self._audio_classifier, - transcription=self._transcription, - settings=self.settings, - tts=self._tts, - ) - # Launch as background task — don't block - self.spawn( - hold_slayer.run(call, sip_leg_id, call_flow_id), - name=f"holdslayer_{call.id}", - ) + # Hand off to the registered per-mode launcher (e.g. hold slayer) + handler = self._mode_handlers.get(mode) + if handler is not None: + handler(call, sip_leg_id, call_flow_id) return call @@ -321,11 +274,14 @@ class AIPSTNGateway: self.call_manager.map_leg(device_leg_id, call_id) # Get the original PSTN leg - pstn_leg_id = None - for leg_id, cid in self.call_manager._call_legs.items(): - if cid == call_id and leg_id != device_leg_id: - pstn_leg_id = leg_id - break + pstn_leg_id = next( + ( + leg_id + for leg_id in self.call_manager.legs_for_call(call_id) + if leg_id != device_leg_id + ), + None, + ) if pstn_leg_id: # Bridge the PSTN leg and device leg @@ -342,9 +298,8 @@ class AIPSTNGateway: raise ValueError(f"Call {call_id} not found") # Hang up all legs associated with this call - for leg_id, cid in list(self.call_manager._call_legs.items()): - if cid == call_id: - await self.sip_engine.hangup(leg_id) + for leg_id in self.call_manager.legs_for_call(call_id): + await self.sip_engine.hangup(leg_id) await self.call_manager.end_call(call_id) @@ -470,73 +425,6 @@ class AIPSTNGateway: }, )) - async def _on_sip_incoming_call( - self, from_uri: str, to_uri: str, leg_id: str - ) -> None: - """ - Called by SippyEngine when an inbound INVITE arrives. - - Evaluates routing rules, then either: - - Rejects (rule says reject/DND) - - Answers + hands off to the AI Receptionist - """ - import uuid as _uuid - from models.call import CallMode, CallStatus - from models.routing import RoutingActionType - - caller_number = _extract_number(from_uri) - dnis = _extract_number(to_uri) - - # Create a call record so the dashboard sees the ringing call. - call = await self.call_manager.create_call( - remote_number=caller_number, - mode=CallMode.RECEPTIONIST, - intent=None, - call_flow_id=None, - device=None, - ) - # Mark inbound - call.direction = "inbound" - self.call_manager.map_leg(leg_id, call.id) - await self.call_manager.update_status(call.id, CallStatus.RINGING) - - decision = ( - await self._routing.evaluate(caller_number, dnis) - if self._routing is not None - else None - ) - - if decision is not None: - await self.event_bus.publish(GatewayEvent( - type=EventType.ROUTING_RULE_MATCHED, - call_id=call.id, - data={ - "matched_rule_id": decision.matched_rule_id, - "matched_rule_name": decision.matched_rule_name, - "action": decision.action.type.value, - "reason": decision.reason, - }, - message=decision.reason, - )) - - if decision.action.type in (RoutingActionType.REJECT, RoutingActionType.DND): - if hasattr(self.sip_engine, "reject_inbound"): - await self.sip_engine.reject_inbound(leg_id) - await self.call_manager.end_call(call.id, CallStatus.COMPLETED) - return - - # Answer the leg - if hasattr(self.sip_engine, "accept_inbound"): - await self.sip_engine.accept_inbound(leg_id) - await self.call_manager.update_status(call.id, CallStatus.CONNECTED) - - # Hand off to the AI Receptionist - if self._receptionist is not None and self.settings.receptionist.enabled: - self.spawn( - self._receptionist.handle(call, leg_id, decision), - name=f"receptionist_{call.id}", - ) - def preferred_device(self) -> Optional[Device]: """Get the highest-priority online device.""" online_devices = [ diff --git a/db/database.py b/db/database.py index e4d7377..c93d658 100644 --- a/db/database.py +++ b/db/database.py @@ -4,6 +4,8 @@ Database connection and session management. PostgreSQL via asyncpg + SQLAlchemy async. """ +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from datetime import datetime from sqlalchemy import ( @@ -204,8 +206,13 @@ def get_session_factory() -> async_sessionmaker[AsyncSession]: return _session_factory -async def get_db() -> AsyncSession: - """Dependency: yield an async database session.""" +@asynccontextmanager +async def session_scope() -> AsyncIterator[AsyncSession]: + """A commit-on-success session — the one session-lifecycle convention. + + REST handlers get it via the get_db dependency; services and MCP + tools use it directly. + """ factory = get_session_factory() async with factory() as session: try: @@ -216,6 +223,12 @@ async def get_db() -> AsyncSession: raise +async def get_db() -> AsyncIterator[AsyncSession]: + """FastAPI dependency: yield an async database session.""" + async with session_scope() as session: + yield session + + async def init_db(): """Create all tables. For development; use Alembic migrations in production.""" engine = get_engine() diff --git a/main.py b/main.py index 392fcce..5a6978d 100644 --- a/main.py +++ b/main.py @@ -21,9 +21,19 @@ from fastapi.staticfiles import StaticFiles from api import call_flows, call_history, calls, devices, routing, websocket from api.deps import require_token from config import Settings, get_settings -from core.gateway import AIPSTNGateway +from core.gateway import AIPSTNGateway, build_sip_engine from db.database import close_db, init_db from mcp_server.server import create_mcp_server +from models.call import CallMode +from services.audio_classifier import AudioClassifier +from services.call_persistence import persist_call_on_end +from services.hold_slayer import HoldSlayerService +from services.notification import NotificationService +from services.receptionist import ReceptionistService +from services.recording import RecordingService +from services.routing import RoutingService +from services.transcription import TranscriptionService +from services.tts import TTSService # Configure logging logging.basicConfig( @@ -126,23 +136,61 @@ async def lifespan(app: FastAPI): except Exception as e: _handle_db_error(e) - # Boot the telephony engine - gateway = AIPSTNGateway.from_config() + # === Composition root === + # Build the gateway and every service here, wiring them by + # constructor/registration — nothing constructs its own deps. + gateway = AIPSTNGateway(settings=settings, on_call_ended=persist_call_on_end) + + classifier = AudioClassifier(settings.classifier) + transcription = TranscriptionService(settings.speaches) + tts = TTSService(settings.tts) + routing_svc = RoutingService(gateway) + recording_svc = RecordingService() + receptionist = ReceptionistService( + gateway, + tts=tts, + transcription=transcription, + recording=recording_svc, + routing=routing_svc, + ) + gateway.attach_services(tts=tts) + + def launch_hold_slayer(call, sip_leg_id, call_flow_id): + svc = HoldSlayerService( + gateway=gateway, + call_manager=gateway.call_manager, + sip_engine=gateway.sip_engine, + classifier=classifier, + transcription=transcription, + settings=settings, + tts=tts, + ) + gateway.spawn( + svc.run(call, sip_leg_id, call_flow_id), + name=f"holdslayer_{call.id}", + ) + + 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, + ) + + await routing_svc.start() await gateway.start() app.state.gateway = gateway - - # Start auxiliary services - from services.notification import NotificationService - from services.recording import RecordingService + app.state.routing_service = routing_svc notification_svc = NotificationService(gateway.event_bus, settings) await notification_svc.start() app.state.notification_service = notification_svc - recording_svc = RecordingService() await recording_svc.start() app.state.recording_service = recording_svc - gateway._recording_service = recording_svc logger.info("=" * 60) logger.info("🔥 Hold Slayer Gateway is LIVE") diff --git a/mcp_server/server.py b/mcp_server/server.py index a389257..3c641d3 100644 --- a/mcp_server/server.py +++ b/mcp_server/server.py @@ -184,18 +184,12 @@ def create_mcp_server( Returns the IVR navigation tree if one exists. """ - from db.database import StoredCallFlow, get_session_factory - from sqlalchemy import select + from db.database import session_scope + from services import call_persistence as store try: - factory = get_session_factory() - async with factory() as session: - result = await session.execute( - select(StoredCallFlow).where( - StoredCallFlow.phone_number == phone_number - ) - ) - row = result.scalar_one_or_none() + async with session_scope() as session: + row = await store.get_flow_by_number(session, phone_number) if not row: return f"No stored call flow for {phone_number}." @@ -243,25 +237,26 @@ def create_mcp_server( """ from slugify import slugify as do_slugify - from db.database import StoredCallFlow, get_session_factory + from db.database import session_scope + from services import call_persistence as store try: steps = json.loads(steps_json) flow_id = do_slugify(name) - factory = get_session_factory() - async with factory() as session: - db_flow = StoredCallFlow( - id=flow_id, + async with session_scope() as session: + if await store.get_flow(session, flow_id): + return f"Call flow '{flow_id}' already exists." + await store.create_flow( + session, + flow_id=flow_id, name=name, phone_number=phone_number, description="Created by AI assistant", steps=steps, - notes=notes or None, tags=["ai-created"], + notes=notes or None, ) - session.add(db_flow) - await session.commit() return f"Call flow '{name}' saved for {phone_number} (ID: {flow_id})" except json.JSONDecodeError: @@ -283,12 +278,12 @@ def create_mcp_server( if not call: return f"Call {call_id} not found." - for leg_id, cid in gateway.call_manager._call_legs.items(): - if cid == call_id: - await gateway.sip_engine.send_dtmf(leg_id, digits) - return f"Sent DTMF '{digits}' on call {call_id}." + legs = gateway.call_manager.legs_for_call(call_id) + if not legs: + return f"No active SIP leg found for call {call_id}." - return f"No active SIP leg found for call {call_id}." + await gateway.sip_engine.send_dtmf(legs[0], digits) + return f"Sent DTMF '{digits}' on call {call_id}." @mcp.tool() async def get_call_transcript(call_id: str) -> str: @@ -318,16 +313,12 @@ def create_mcp_server( Returns the recording file path and status. """ - from db.database import CallRecord, get_session_factory - from sqlalchemy import select + from db.database import session_scope + from services import call_persistence as store try: - factory = get_session_factory() - async with factory() as session: - result = await session.execute( - select(CallRecord).where(CallRecord.id == call_id) - ) - record = result.scalar_one_or_none() + async with session_scope() as session: + record = await store.get_record(session, call_id) if not record: return f"No record found for call {call_id}." if not record.recording_path: @@ -348,16 +339,12 @@ def create_mcp_server( Returns the summary, action items, and sentiment analysis. """ - from db.database import CallRecord, get_session_factory - from sqlalchemy import select + from db.database import session_scope + from services import call_persistence as store try: - factory = get_session_factory() - async with factory() as session: - result = await session.execute( - select(CallRecord).where(CallRecord.id == call_id) - ) - record = result.scalar_one_or_none() + async with session_scope() as session: + record = await store.get_record(session, call_id) if not record: return f"No record found for call {call_id}." @@ -398,27 +385,17 @@ def create_mcp_server( intent: Filter by intent text (partial match) limit: Max results to return (default 10) """ - from db.database import CallRecord, get_session_factory - from sqlalchemy import select + from db.database import session_scope + from services import call_persistence as store try: - factory = get_session_factory() - async with factory() as session: - query = select(CallRecord).order_by( - CallRecord.started_at.desc() - ).limit(limit) - - if phone_number: - query = query.where( - CallRecord.remote_number.contains(phone_number) - ) - if intent: - query = query.where( - CallRecord.intent.icontains(intent) - ) - - result = await session.execute(query) - records = result.scalars().all() + async with session_scope() as session: + records = await store.search_history( + session, + number_contains=phone_number or None, + intent_contains=intent or None, + limit=limit, + ) if not records: return "No matching call records found." @@ -482,14 +459,12 @@ def create_mcp_server( @mcp.resource("gateway://call-flows") async def resource_call_flows() -> str: """List all stored call flows.""" - from db.database import StoredCallFlow, get_session_factory - from sqlalchemy import select + from db.database import session_scope + from services import call_persistence as store try: - factory = get_session_factory() - async with factory() as session: - result = await session.execute(select(StoredCallFlow)) - rows = result.scalars().all() + async with session_scope() as session: + rows = await store.list_flows(session) flows = [ { "id": r.id, diff --git a/pyproject.toml b/pyproject.toml index 877f185..70b0676 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,6 +50,7 @@ dev = [ "pytest-cov>=6.0.0", "httpx>=0.28.0", "ruff>=0.8.0", + "aiosqlite>=0.22.0", ] [tool.setuptools.packages.find] diff --git a/services/call_analytics.py b/services/call_analytics.py deleted file mode 100644 index 4fa67d7..0000000 --- a/services/call_analytics.py +++ /dev/null @@ -1,324 +0,0 @@ -""" -Call Analytics Service — Tracks call metrics and generates insights. - -Monitors call patterns, hold times, success rates, and IVR navigation -efficiency. Provides data for the dashboard and API. -""" - -import logging -from collections import defaultdict -from datetime import datetime, timedelta -from typing import Any, Optional - -from models.call import ActiveCall, AudioClassification, CallMode, CallStatus - -logger = logging.getLogger(__name__) - - -class CallAnalytics: - """ - In-memory call analytics engine. - - Tracks: - - Call success/failure rates - - Hold time statistics (avg, min, max, p95) - - IVR navigation efficiency - - Human detection accuracy - - Per-number/company patterns - - Time-of-day patterns - - In production, this would be backed by TimescaleDB or similar. - For now, we keep rolling windows in memory. - """ - - def __init__(self, max_history: int = 10000): - self._max_history = max_history - self._call_records: list[CallRecord] = [] - self._company_stats: dict[str, CompanyStats] = defaultdict(CompanyStats) - - # ================================================================ - # Record Calls - # ================================================================ - - def record_call(self, call: ActiveCall) -> None: - """ - Record a completed call for analytics. - - Called when a call ends (from CallManager). - """ - record = CallRecord( - call_id=call.id, - remote_number=call.remote_number, - mode=call.mode, - status=call.status, - intent=call.intent, - started_at=call.created_at, - duration_seconds=call.duration, - hold_time_seconds=call.hold_time, - classification_history=[ - r.audio_type.value for r in call.classification_history - ], - transcript_chunks=list(call.transcript_chunks), - services=list(call.services), - ) - - self._call_records.append(record) - - # Trim history - if len(self._call_records) > self._max_history: - self._call_records = self._call_records[-self._max_history :] - - # Update company stats - company_key = self._normalize_number(call.remote_number) - self._company_stats[company_key].update(record) - - logger.debug( - f"📊 Recorded call {call.id}: " - f"{call.status.value}, {call.duration}s, hold={call.hold_time}s" - ) - - # ================================================================ - # Aggregate Stats - # ================================================================ - - def get_summary(self, hours: int = 24) -> dict[str, Any]: - """Get summary statistics for the last N hours.""" - cutoff = datetime.now() - timedelta(hours=hours) - recent = [r for r in self._call_records if r.started_at >= cutoff] - - if not recent: - return { - "period_hours": hours, - "total_calls": 0, - "success_rate": 0.0, - "avg_hold_time": 0.0, - "avg_duration": 0.0, - } - - total = len(recent) - successful = sum(1 for r in recent if r.status in ( - CallStatus.COMPLETED, CallStatus.BRIDGED, CallStatus.HUMAN_DETECTED - )) - failed = sum(1 for r in recent if r.status == CallStatus.FAILED) - - hold_times = [r.hold_time_seconds for r in recent if r.hold_time_seconds > 0] - durations = [r.duration_seconds for r in recent if r.duration_seconds > 0] - - hold_slayer_calls = [r for r in recent if r.mode == CallMode.HOLD_SLAYER] - hold_slayer_success = sum( - 1 for r in hold_slayer_calls - if r.status in (CallStatus.BRIDGED, CallStatus.HUMAN_DETECTED) - ) - - return { - "period_hours": hours, - "total_calls": total, - "successful": successful, - "failed": failed, - "success_rate": round(successful / total, 3) if total else 0.0, - "avg_duration": round(sum(durations) / len(durations), 1) if durations else 0.0, - "max_duration": max(durations) if durations else 0, - "hold_time": { - "avg": round(sum(hold_times) / len(hold_times), 1) if hold_times else 0.0, - "min": min(hold_times) if hold_times else 0, - "max": max(hold_times) if hold_times else 0, - "p95": self._percentile(hold_times, 95) if hold_times else 0, - "total": sum(hold_times), - }, - "hold_slayer": { - "total": len(hold_slayer_calls), - "success": hold_slayer_success, - "success_rate": round( - hold_slayer_success / len(hold_slayer_calls), 3 - ) if hold_slayer_calls else 0.0, - }, - "by_mode": self._group_by_mode(recent), - "by_hour": self._group_by_hour(recent), - } - - def get_company_stats(self, number: str) -> dict[str, Any]: - """Get stats for a specific company/number.""" - key = self._normalize_number(number) - stats = self._company_stats.get(key) - if not stats: - return {"number": number, "total_calls": 0} - return stats.to_dict(number) - - def get_top_numbers(self, limit: int = 10) -> list[dict[str, Any]]: - """Get the most-called numbers with their stats.""" - sorted_stats = sorted( - self._company_stats.items(), - key=lambda x: x[1].total_calls, - reverse=True, - )[:limit] - return [stats.to_dict(number) for number, stats in sorted_stats] - - # ================================================================ - # Hold Time Trends - # ================================================================ - - def get_hold_time_trend( - self, - number: Optional[str] = None, - days: int = 7, - ) -> list[dict]: - """ - Get hold time trend data for graphing. - - Returns daily average hold times for the last N days. - """ - cutoff = datetime.now() - timedelta(days=days) - records = [r for r in self._call_records if r.started_at >= cutoff] - - if number: - key = self._normalize_number(number) - records = [r for r in records if self._normalize_number(r.remote_number) == key] - - # Group by day - by_day: dict[str, list[int]] = defaultdict(list) - for r in records: - day = r.started_at.strftime("%Y-%m-%d") - if r.hold_time_seconds > 0: - by_day[day].append(r.hold_time_seconds) - - trend = [] - for i in range(days): - date = (datetime.now() - timedelta(days=days - 1 - i)).strftime("%Y-%m-%d") - times = by_day.get(date, []) - trend.append({ - "date": date, - "avg_hold_time": round(sum(times) / len(times), 1) if times else 0, - "call_count": len(times), - "max_hold_time": max(times) if times else 0, - }) - - return trend - - # ================================================================ - # Helpers - # ================================================================ - - @staticmethod - def _normalize_number(number: str) -> str: - """Normalize phone number for grouping.""" - # Strip formatting, keep last 10 digits - digits = "".join(c for c in number if c.isdigit()) - return digits[-10:] if len(digits) >= 10 else digits - - @staticmethod - def _percentile(values: list, pct: int) -> float: - """Calculate percentile value.""" - if not values: - return 0.0 - sorted_vals = sorted(values) - idx = int(len(sorted_vals) * pct / 100) - idx = min(idx, len(sorted_vals) - 1) - return float(sorted_vals[idx]) - - @staticmethod - def _group_by_mode(records: list["CallRecord"]) -> dict[str, int]: - """Group call counts by mode.""" - by_mode: dict[str, int] = defaultdict(int) - for r in records: - by_mode[r.mode.value] += 1 - return dict(by_mode) - - @staticmethod - def _group_by_hour(records: list["CallRecord"]) -> dict[int, int]: - """Group call counts by hour of day.""" - by_hour: dict[int, int] = defaultdict(int) - for r in records: - by_hour[r.started_at.hour] += 1 - return dict(sorted(by_hour.items())) - - @property - def total_calls_recorded(self) -> int: - return len(self._call_records) - - -# ================================================================ -# Data Models -# ================================================================ - -class CallRecord: - """A completed call record for analytics.""" - - def __init__( - self, - call_id: str, - remote_number: str, - mode: CallMode, - status: CallStatus, - intent: Optional[str] = None, - started_at: Optional[datetime] = None, - duration_seconds: int = 0, - hold_time_seconds: int = 0, - classification_history: Optional[list[str]] = None, - transcript_chunks: Optional[list[str]] = None, - services: Optional[list[str]] = None, - ): - self.call_id = call_id - self.remote_number = remote_number - self.mode = mode - self.status = status - self.intent = intent - self.started_at = started_at or datetime.now() - self.duration_seconds = duration_seconds - self.hold_time_seconds = hold_time_seconds - self.classification_history = classification_history or [] - self.transcript_chunks = transcript_chunks or [] - self.services = services or [] - - -class CompanyStats: - """Aggregated stats for a specific company/phone number.""" - - def __init__(self): - self.total_calls = 0 - self.successful_calls = 0 - self.failed_calls = 0 - self.total_hold_time = 0 - self.hold_times: list[int] = [] - self.total_duration = 0 - self.last_called: Optional[datetime] = None - self.intents: dict[str, int] = defaultdict(int) - - def update(self, record: CallRecord) -> None: - """Update stats with a new call record.""" - self.total_calls += 1 - self.total_duration += record.duration_seconds - self.last_called = record.started_at - - if record.status in (CallStatus.COMPLETED, CallStatus.BRIDGED, CallStatus.HUMAN_DETECTED): - self.successful_calls += 1 - elif record.status == CallStatus.FAILED: - self.failed_calls += 1 - - if record.hold_time_seconds > 0: - self.total_hold_time += record.hold_time_seconds - self.hold_times.append(record.hold_time_seconds) - - if record.intent: - self.intents[record.intent] += 1 - - def to_dict(self, number: str) -> dict[str, Any]: - return { - "number": number, - "total_calls": self.total_calls, - "successful_calls": self.successful_calls, - "failed_calls": self.failed_calls, - "success_rate": round( - self.successful_calls / self.total_calls, 3 - ) if self.total_calls else 0.0, - "avg_hold_time": round( - self.total_hold_time / len(self.hold_times), 1 - ) if self.hold_times else 0.0, - "max_hold_time": max(self.hold_times) if self.hold_times else 0, - "avg_duration": round( - self.total_duration / self.total_calls, 1 - ) if self.total_calls else 0.0, - "last_called": self.last_called.isoformat() if self.last_called else None, - "top_intents": dict( - sorted(self.intents.items(), key=lambda x: x[1], reverse=True)[:5] - ), - } diff --git a/services/call_persistence.py b/services/call_persistence.py index 393954a..af57d77 100644 --- a/services/call_persistence.py +++ b/services/call_persistence.py @@ -1,25 +1,168 @@ """ -Call Persistence — Writes completed calls and their transcript chunks -to the database when CallManager.end_call() fires. +Call Persistence — the data-access layer for calls and call flows. + +Holds the on-hangup persistence hook plus the query/write functions +that both the REST handlers and the MCP tools call, so the two +surfaces can't drift. Every function takes an AsyncSession; callers +own the transaction (get_db for REST, session_scope for MCP/services). """ import logging import uuid from datetime import datetime -from db.database import CallRecord, TranscriptChunk, get_session_factory +from sqlalchemy import desc, select +from sqlalchemy.ext.asyncio import AsyncSession + +from db.database import ( + CallRecord, + RecordingRecord, + StoredCallFlow, + TranscriptChunk, + session_scope, +) from models.call import ActiveCall, CallStatus +from models.call_flow import CallFlow, CallFlowStep logger = logging.getLogger(__name__) +def flow_to_model(row: StoredCallFlow) -> CallFlow: + """The one StoredCallFlow-row → CallFlow-model mapping.""" + return CallFlow( + id=row.id, + name=row.name, + phone_number=row.phone_number, + description=row.description or "", + steps=[CallFlowStep(**s) for s in (row.steps or [])], + tags=row.tags or [], + notes=row.notes, + avg_hold_time=row.avg_hold_time, + success_rate=row.success_rate, + last_used=row.last_used, + times_used=row.times_used or 0, + ) + + +# ================================================================ +# Call flows +# ================================================================ + +async def get_flow(session: AsyncSession, flow_id: str) -> StoredCallFlow | None: + result = await session.execute( + select(StoredCallFlow).where(StoredCallFlow.id == flow_id) + ) + return result.scalar_one_or_none() + + +async def get_flow_by_number( + session: AsyncSession, phone_number: str +) -> StoredCallFlow | None: + result = await session.execute( + select(StoredCallFlow).where(StoredCallFlow.phone_number == phone_number) + ) + return result.scalar_one_or_none() + + +async def list_flows(session: AsyncSession) -> list[StoredCallFlow]: + result = await session.execute(select(StoredCallFlow)) + return list(result.scalars().all()) + + +async def create_flow( + session: AsyncSession, + flow_id: str, + name: str, + phone_number: str, + steps: list[dict], + description: str | None = None, + tags: list[str] | None = None, + notes: str | None = None, +) -> StoredCallFlow: + row = StoredCallFlow( + id=flow_id, + name=name, + phone_number=phone_number, + description=description, + steps=steps, + tags=tags, + notes=notes, + last_verified=datetime.now(), + ) + session.add(row) + await session.flush() + return row + + +# ================================================================ +# Call history / records +# ================================================================ + +async def get_record(session: AsyncSession, call_id: str) -> CallRecord | None: + result = await session.execute( + select(CallRecord).where(CallRecord.id == call_id) + ) + return result.scalar_one_or_none() + + +async def search_history( + session: AsyncSession, + number: str | None = None, + number_contains: str | None = None, + intent_contains: str | None = None, + status: str | None = None, + since: datetime | None = None, + until: datetime | None = None, + limit: int = 50, + offset: int = 0, +) -> list[CallRecord]: + stmt = select(CallRecord).order_by(desc(CallRecord.started_at)) + if number: + stmt = stmt.where(CallRecord.remote_number == number) + if number_contains: + stmt = stmt.where(CallRecord.remote_number.contains(number_contains)) + if intent_contains: + stmt = stmt.where(CallRecord.intent.icontains(intent_contains)) + if status: + stmt = stmt.where(CallRecord.status == status) + if since: + stmt = stmt.where(CallRecord.started_at >= since) + if until: + stmt = stmt.where(CallRecord.started_at <= until) + result = await session.execute(stmt.offset(offset).limit(limit)) + return list(result.scalars().all()) + + +async def get_transcript_chunks( + session: AsyncSession, call_id: str +) -> list[TranscriptChunk]: + result = await session.execute( + select(TranscriptChunk) + .where(TranscriptChunk.call_id == call_id) + .order_by(TranscriptChunk.seq) + ) + return list(result.scalars().all()) + + +async def latest_recording( + session: AsyncSession, call_id: str +) -> RecordingRecord | None: + result = await session.execute( + select(RecordingRecord) + .where(RecordingRecord.call_id == call_id) + .order_by(desc(RecordingRecord.started_at)) + ) + return result.scalars().first() + + async def persist_call_on_end(call: ActiveCall, final_status: CallStatus) -> None: """Insert a CallRecord and any transcript chunks for `call`. - Wired into CallManager via _on_call_ended in gateway.start(). + Wired into CallManager as its on_call_ended hook by the + composition root in main.py. """ try: - async with get_session_factory()() as session: + async with session_scope() as session: record = CallRecord( id=call.id, direction=call.direction, @@ -64,7 +207,5 @@ async def persist_call_on_end(call: ActiveCall, final_status: CallStatus) -> Non speaker=speaker, text=payload, )) - - await session.commit() except Exception as e: logger.warning(f"Could not persist call {call.id}: {e}") diff --git a/services/hold_slayer.py b/services/hold_slayer.py index d9cec27..4d6031b 100644 --- a/services/hold_slayer.py +++ b/services/hold_slayer.py @@ -23,35 +23,12 @@ from models.call import ActiveCall, AudioClassification, CallStatus, Classificat from models.call_flow import ActionType, CallFlow, CallFlowStep from models.events import EventType, GatewayEvent from services.audio_classifier import AudioClassifier +from services.llm_client import get_llm from services.transcription import TranscriptionService from services.tts import TTSService logger = logging.getLogger(__name__) -# LLM client is optional — imported at use time -_llm_client = None - - -def _get_llm(): - """Lazy-load LLM client (optional dependency).""" - global _llm_client - if _llm_client is None: - try: - from config import get_settings - from services.llm_client import LLMClient - - settings = get_settings() - _llm_client = LLMClient( - base_url=settings.llm.base_url, - model=settings.llm.model, - api_key=settings.llm.api_key.get_secret_value(), - timeout=settings.llm.timeout, - ) - except Exception as e: - logger.debug(f"LLM client not available: {e}") - _llm_client = False # Sentinel: don't retry - return _llm_client if _llm_client is not False else None - class HoldSlayerService: """ @@ -228,7 +205,7 @@ class HoldSlayerService: # Phase 2: LLM fallback if regex couldn't decide if not decision and transcript: - llm = _get_llm() + llm = get_llm() if llm: try: logger.info("🤖 Regex inconclusive, asking LLM...") diff --git a/services/llm_client.py b/services/llm_client.py index 623b157..b754afa 100644 --- a/services/llm_client.py +++ b/services/llm_client.py @@ -389,3 +389,31 @@ class LLMClient: "model": self.model, "base_url": self.base_url, } + + +# ================================================================ +# Shared lazy client +# ================================================================ + +_shared_client: Optional["LLMClient"] = None +_shared_failed = False + + +def get_llm() -> Optional["LLMClient"]: + """Lazily build the shared LLMClient from settings (None if unavailable).""" + global _shared_client, _shared_failed + if _shared_client is None and not _shared_failed: + try: + from config import get_settings + + settings = get_settings() + _shared_client = LLMClient( + base_url=settings.llm.base_url, + model=settings.llm.model, + api_key=settings.llm.api_key.get_secret_value(), + timeout=settings.llm.timeout, + ) + except Exception as e: + logger.debug(f"LLM client not available: {e}") + _shared_failed = True # don't retry + return _shared_client diff --git a/services/receptionist.py b/services/receptionist.py index a27e827..1ecc071 100644 --- a/services/receptionist.py +++ b/services/receptionist.py @@ -27,12 +27,100 @@ from models.routing import RoutingAction, RoutingActionType, RoutingDecision logger = logging.getLogger(__name__) -class ReceptionistService: - """Drives the receptionist state machine for a single inbound call.""" +def _extract_number(sip_uri: str) -> str: + """Pull the user part out of a SIP URI (sip:+15551212@host → +15551212).""" + if not sip_uri: + return "" + s = sip_uri.strip() + if s.startswith("<") and ">" in s: + s = s[1 : s.index(">")] + if s.startswith("sip:"): + s = s[4:] + if "@" in s: + s = s.split("@", 1)[0] + return s - def __init__(self, gateway): + +class ReceptionistService: + """Owns inbound-call policy: routing evaluation, screening, voicemail.""" + + def __init__( + self, + gateway, + tts=None, + transcription=None, + recording=None, + routing=None, + ): self.gateway = gateway self.settings = gateway.settings.receptionist + self.tts = tts + self.transcription = transcription + self.recording = recording + self.routing = routing + + async def on_inbound_call(self, from_uri: str, to_uri: str, leg_id: str) -> None: + """ + Entry point for an inbound INVITE (wired as the SIP engine's + on_incoming_call by the composition root). + + Evaluates routing rules, then either rejects (rule says + reject/DND) or answers and runs the screening flow. + """ + from models.call import CallMode + + gateway = self.gateway + caller_number = _extract_number(from_uri) + dnis = _extract_number(to_uri) + + # Create a call record so the dashboard sees the ringing call. + call = await gateway.call_manager.create_call( + remote_number=caller_number, + mode=CallMode.RECEPTIONIST, + intent=None, + call_flow_id=None, + device=None, + ) + call.direction = "inbound" + gateway.call_manager.map_leg(leg_id, call.id) + await gateway.call_manager.update_status(call.id, CallStatus.RINGING) + + decision = ( + await self.routing.evaluate(caller_number, dnis) + if self.routing is not None + else None + ) + + if decision is not None: + await gateway.event_bus.publish(GatewayEvent( + type=EventType.ROUTING_RULE_MATCHED, + call_id=call.id, + data={ + "matched_rule_id": decision.matched_rule_id, + "matched_rule_name": decision.matched_rule_name, + "action": decision.action.type.value, + "reason": decision.reason, + }, + message=decision.reason, + )) + + if decision.action.type in (RoutingActionType.REJECT, RoutingActionType.DND): + if hasattr(gateway.sip_engine, "reject_inbound"): + await gateway.sip_engine.reject_inbound(leg_id) + await gateway.call_manager.end_call(call.id, CallStatus.COMPLETED) + return + + # Answer the leg + if hasattr(gateway.sip_engine, "accept_inbound"): + await gateway.sip_engine.accept_inbound(leg_id) + await gateway.call_manager.update_status(call.id, CallStatus.CONNECTED) + + # Screen the caller (unless the receptionist is disabled) + if self.settings.enabled: + gateway.spawn( + self.handle(call, leg_id, decision), + name=f"receptionist_{call.id}", + ) async def handle( self, @@ -89,7 +177,10 @@ class ReceptionistService: await self._speak( call, sip_leg_id, "One moment, I'll connect you now." ) - answered = await self.gateway._routing.ring_chain( + if self.routing is None: + await self._take_message(call, sip_leg_id) + return + answered = await self.routing.ring_chain( call.id, devices, action.ring_timeout ) if answered: @@ -156,10 +247,10 @@ class ReceptionistService: finally: tap.close() - if not audio: + if not audio or self.transcription is None: return "" - return await self.gateway._transcription.transcribe(bytes(audio)) + return await self.transcription.transcribe(bytes(audio)) async def _classify( self, @@ -168,9 +259,9 @@ class ReceptionistService: routing_decision: Optional[RoutingDecision], ) -> dict: """Ask the LLM to interpret the caller's utterance.""" - from services.hold_slayer import _get_llm + from services.llm_client import get_llm - llm = _get_llm() + llm = get_llm() if llm is None or not transcript.strip(): return { "intent": transcript or "unknown", @@ -245,7 +336,7 @@ class ReceptionistService: await self._speak(call, sip_leg_id, self.settings.message_prompt) media = self.gateway.media_pipeline - recording_svc = getattr(self.gateway, "_recording_service", None) + recording_svc = self.recording if recording_svc is None or media is None: logger.warning("Receptionist: recording unavailable, ending call") await self._hangup(call, sip_leg_id) @@ -264,10 +355,10 @@ class ReceptionistService: message_text = "" rec_path = session.filepath_mixed if session else None - if rec_path and Path(rec_path).exists(): + 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.gateway._transcription.transcribe(audio_bytes) + message_text = await self.transcription.transcribe(audio_bytes) except Exception as e: logger.warning(f"Receptionist transcribe failed: {e}") @@ -292,7 +383,7 @@ class ReceptionistService: # ---------------------------------------------------------------- async def _speak(self, call: ActiveCall, sip_leg_id: str, text: str) -> None: - tts = self.gateway._tts + tts = self.tts media = self.gateway.media_pipeline if tts is None or media is None or not text.strip(): return diff --git a/tests/test_structure.py b/tests/test_structure.py new file mode 100644 index 0000000..beb9d5b --- /dev/null +++ b/tests/test_structure.py @@ -0,0 +1,227 @@ +""" +Composition and API-surface tests. + +Covers the gateway composed the way main.py's lifespan composes it +(mode handlers, on_call_ended hook, receptionist-owned inbound +policy) and the REST routes running against a real (SQLite) database +through the shared data layer in services/call_persistence.py. +""" + +import httpx +import pytest +from pydantic import SecretStr +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import StaticPool + +import main +from config import ReceptionistSettings, Settings, get_settings +from core.gateway import AIPSTNGateway +from db.database import Base, CallRecord, get_db +from models.call import CallMode, CallStatus +from models.routing import RoutingAction, RoutingActionType, RoutingDecision +from services.receptionist import ReceptionistService + +# ================================================================ +# Gateway composition +# ================================================================ + +class TestGatewayComposition: + async def test_mode_handler_launches_per_call(self): + gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4)) + launched: list[tuple] = [] + gateway.register_mode_handler( + CallMode.HOLD_SLAYER, + lambda call, leg_id, flow_id: launched.append((call.id, leg_id, flow_id)), + ) + + call = await gateway.make_call("+15551234567", mode=CallMode.HOLD_SLAYER, + call_flow_id="acme-main") + + assert launched == [(call.id, gateway.call_manager.legs_for_call(call.id)[0], "acme-main")] + + async def test_direct_mode_needs_no_handler(self): + gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4)) + call = await gateway.make_call("+15551234567") + assert call.status == CallStatus.RINGING + + async def test_on_call_ended_hook_from_constructor(self): + ended: list[tuple] = [] + + async def hook(call, status): + ended.append((call.id, status)) + + gateway = AIPSTNGateway( + settings=Settings(max_concurrent_calls=4), on_call_ended=hook + ) + call = await gateway.make_call("+15551234567") + await gateway.hangup_call(call.id) + + assert ended == [(call.id, CallStatus.COMPLETED)] + + +# ================================================================ +# Receptionist-owned inbound policy +# ================================================================ + +class _StubRouting: + def __init__(self, decision): + self._decision = decision + + async def evaluate(self, caller_number, dnis): + return self._decision + + +class TestInboundPolicy: + def _gateway(self) -> AIPSTNGateway: + settings = Settings(max_concurrent_calls=4) + settings.receptionist = ReceptionistSettings(enabled=False) + return AIPSTNGateway(settings=settings) + + async def test_inbound_call_answered_and_tracked(self): + gateway = self._gateway() + receptionist = ReceptionistService(gateway) + + await receptionist.on_inbound_call( + "sip:+16135550100@pstn", "sip:+15551234567@gw", "leg_in1" + ) + + calls = list(gateway.call_manager.active_calls.values()) + assert len(calls) == 1 + call = calls[0] + assert call.direction == "inbound" + assert call.remote_number == "+16135550100" + assert call.status == CallStatus.CONNECTED + assert gateway.call_manager.legs_for_call(call.id) == ["leg_in1"] + + async def test_reject_rule_declines_before_answer(self): + gateway = self._gateway() + decision = RoutingDecision( + action=RoutingAction(type=RoutingActionType.REJECT), + matched_rule_id="rule_x", + matched_rule_name="block", + reason="matched rule 'block'", + ) + receptionist = ReceptionistService(gateway, routing=_StubRouting(decision)) + + await receptionist.on_inbound_call( + "sip:+18005550100@pstn", "sip:+15551234567@gw", "leg_in2" + ) + + assert gateway.call_manager.active_calls == {} + + +# ================================================================ +# REST routes on the shared data layer (real SQLite) +# ================================================================ + +@pytest.fixture +async def client(monkeypatch): + monkeypatch.setattr(get_settings(), "api_token", SecretStr("")) + + engine = create_async_engine( + "sqlite+aiosqlite:///:memory:", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + factory = async_sessionmaker(engine, expire_on_commit=False) + + async def _get_db(): + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + main.app.dependency_overrides[get_db] = _get_db + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: + c.db_factory = factory + yield c + main.app.dependency_overrides.pop(get_db, None) + await engine.dispose() + + +FLOW_PAYLOAD = { + "name": "Acme Main Line", + "phone_number": "+18005551234", + "description": "Main IVR", + "steps": [ + { + "id": "step1", + "description": "Press 2 for billing", + "action": "dtmf", + "action_value": "2", + } + ], + "tags": ["test"], +} + + +class TestCallFlowRoutes: + async def test_crud_round_trip(self, client): + resp = await client.post("/api/call-flows/", json=FLOW_PAYLOAD) + assert resp.status_code == 200, resp.text + flow_id = resp.json()["id"] + assert flow_id == "acme-main-line" + + resp = await client.post("/api/call-flows/", json=FLOW_PAYLOAD) + assert resp.status_code == 409 + + resp = await client.get("/api/call-flows/") + assert [f["id"] for f in resp.json()] == [flow_id] + + resp = await client.get(f"/api/call-flows/{flow_id}") + assert resp.json()["steps"][0]["action_value"] == "2" + + resp = await client.get("/api/call-flows/by-number/+18005551234") + assert resp.json()["id"] == flow_id + + resp = await client.put( + f"/api/call-flows/{flow_id}", json={"notes": "updated"} + ) + assert resp.json()["notes"] == "updated" + + resp = await client.delete(f"/api/call-flows/{flow_id}") + assert resp.json()["status"] == "deleted" + + resp = await client.get(f"/api/call-flows/{flow_id}") + assert resp.status_code == 404 + + +class TestCallHistoryRoutes: + async def test_history_and_record(self, client): + resp = await client.get("/api/calls/history") + assert resp.status_code == 200 + assert resp.json() == [] + + async with client.db_factory() as session: + session.add(CallRecord( + id="call_hist1", + direction="outbound", + remote_number="+18005551234", + status="completed", + mode="hold_slayer", + intent="dispute charge", + duration=120, + hold_time=90, + )) + await session.commit() + + resp = await client.get("/api/calls/history") + assert [r["id"] for r in resp.json()] == ["call_hist1"] + + resp = await client.get("/api/calls/history?number=%2B18005551234") + assert len(resp.json()) == 1 + + resp = await client.get("/api/calls/call_hist1/record") + assert resp.json()["intent"] == "dispute charge" + + resp = await client.get("/api/calls/call_missing/record") + assert resp.status_code == 404 + + resp = await client.get("/api/calls/call_hist1/transcript") + assert resp.json() == []