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 db.database import get_db
-from core.gateway import AIPSTNGateway
-from db.database import StoredCallFlow, 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
+    if await store.get_flow(db, flow_id):
-    existing = await db.execute(
-        select(StoredCallFlow).where(StoredCallFlow.id == flow_id)
-    )
-    if existing.scalar_one_or_none():
         raise HTTPException(
             status_code=409,
             detail=f"Call flow '{flow_id}' already exists. Use PUT to update.",
         )
 
-    db_flow = StoredCallFlow(
+    row = await store.create_flow(
-        id=flow_id,
+        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 = await store.list_flows(db)
-    rows = result.scalars().all()
-
     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(
+    row = await store.get_flow(db, flow_id)
-        select(StoredCallFlow).where(StoredCallFlow.id == flow_id)
-    )
-    row = result.scalar_one_or_none()
     if not row:
         raise HTTPException(status_code=404, detail=f"Call flow '{flow_id}' not found")
-
+    return store.flow_to_model(row)
-    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,
-    )
 
 
 @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(
+    row = await store.get_flow_by_number(db, phone_number)
-        select(StoredCallFlow).where(StoredCallFlow.phone_number == phone_number)
-    )
-    row = result.scalar_one_or_none()
     if not row:
         raise HTTPException(
             status_code=404,
             detail=f"No call flow found for {phone_number}",
         )
-
+    return store.flow_to_model(row)
-    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,
-    )
 
 
 @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(
+    row = await store.get_flow(db, flow_id)
-        select(StoredCallFlow).where(StoredCallFlow.id == flow_id)
-    )
-    row = result.scalar_one_or_none()
     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 store.flow_to_model(row)
-    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,
-    )
 
 
 @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(
+    row = await store.get_flow(db, flow_id)
-        select(StoredCallFlow).where(StoredCallFlow.id == flow_id)
-    )
-    row = result.scalar_one_or_none()
     if not row:
         raise HTTPException(status_code=404, detail=f"Call flow '{flow_id}' not found")
 

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 (
+from db.database import get_db
-    CallRecord,
+from services import call_persistence as store
-    RecordingRecord,
-    TranscriptChunk,
-    get_db,
-)
 
 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,
+    number: str | None = None,
-    status: Optional[str] = None,
+    status: str | None = None,
-    since: Optional[datetime] = None,
+    since: datetime | None = None,
-    until: Optional[datetime] = 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))
+    rows = await store.search_history(
-    if number:
+        db,
-        stmt = stmt.where(CallRecord.remote_number == number)
+        number=number,
-    if status:
+        status=status,
-        stmt = stmt.where(CallRecord.status == status)
+        since=since,
-    if since:
+        until=until,
-        stmt = stmt.where(CallRecord.started_at >= since)
+        limit=limit,
-    if until:
+        offset=offset,
-        stmt = stmt.where(CallRecord.started_at <= until)
+    )
-
-    rows = (await db.execute(stmt.offset(offset).limit(limit))).scalars().all()
     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(
+    row = await store.get_record(db, call_id)
-        select(CallRecord).where(CallRecord.id == call_id)
-    )).scalar_one_or_none()
     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(
+    rows = await store.get_transcript_chunks(db, call_id)
-        select(TranscriptChunk)
-        .where(TranscriptChunk.call_id == call_id)
-        .order_by(TranscriptChunk.seq)
-    )).scalars().all()
     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(
+    row = await store.latest_recording(db, call_id)
-        select(RecordingRecord)
-        .where(RecordingRecord.call_id == call_id)
-        .order_by(desc(RecordingRecord.started_at))
-    )).scalar_one_or_none()
     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))

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
+    legs = gateway.call_manager.legs_for_call(call_id)
-    for leg_id, cid in gateway.call_manager._call_legs.items():
+    if not legs:
-        if cid == call_id:
+        raise HTTPException(status_code=409, detail="No active SIP leg found for this call")
-            await gateway.sip_engine.send_dtmf(leg_id, digits)
-            return {"status": "sent", "digits": digits}
 
-    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}

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.

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)):
+async def list_rules(routing: RoutingService = Depends(get_routing_service)):
-    if gateway._routing is None:
+    return sorted(routing.rules, key=lambda r: (r.priority, r.id))
-        raise HTTPException(status_code=503, detail="Routing service not ready")
-    return sorted(gateway._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:
+    return await routing.create_rule(payload)
-        raise HTTPException(status_code=503, detail="Routing service not ready")
-    return await gateway._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:
+    rule = await routing.update_rule(rule_id, payload)
-        raise HTTPException(status_code=503, detail="Routing service not ready")
-    rule = await gateway._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:
+    ok = await routing.delete_rule(rule_id)
-        raise HTTPException(status_code=503, detail="Routing service not ready")
-    ok = await gateway._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}

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

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.
+The application service that FastAPI and MCP talk to for live-call
-This is the top-level object that FastAPI and MCP talk to.
+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:
+def build_sip_engine(
-    """Pull the user part out of a SIP URI (sip:+15551212@host → +15551212)."""
+    settings: Settings,
-    if not sip_uri:
+    media_pipeline: MediaPipeline,
-        return ""
+    on_leg_state_change: Callable,
-    s = sip_uri.strip()
+    on_device_registered: Callable,
-    if s.startswith("<") and ">" in s:
+    on_incoming_call: Callable,
-        s = s[1:s.index(">")]
+) -> SIPEngine:
-    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:
     """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,
+                media_pipeline=media_pipeline,
-                on_leg_state_change=gateway._on_sip_leg_state,
+                on_leg_state_change=on_leg_state_change,
-                on_device_registered=gateway._on_sip_device_registered,
+                on_device_registered=on_device_registered,
-                on_incoming_call=gateway._on_sip_incoming_call,
+                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:
+    Owns live-call operations (make/transfer/hangup), the device
-    - SIP engine (signaling + media)
+    registry, and per-call background tasks. Services are attached by
-    - Call manager (state + events)
+    the composition root; mode handlers launch per-call services
-    - Hold Slayer service
+    (hold slayer) without the gateway knowing their types.
-    - Audio classifier
-    - Transcription service
-    - Device management
     """
 
     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())
+        # Attached by the composition root (attach_services)
-        self._hold_slayer = None
-        self._audio_classifier = None
-        self._transcription = None
         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 attach_services(self, tts=None) -> None:
-    def from_config(cls, sip_engine: Optional[SIPEngine] = None) -> "AIPSTNGateway":
+        """Attach shared services the gateway must manage on shutdown."""
-        """Create gateway from environment config."""
+        self._tts = tts
-        settings = get_settings()
+
-        gw = cls(settings=settings)
+    def register_mode_handler(self, mode: CallMode, handler: Callable) -> None:
-        if sip_engine is not None:
+        """Register a launcher called as handler(call, sip_leg_id, call_flow_id)."""
-            gw.sip_engine = sip_engine
+        self._mode_handlers[mode] = handler
-        else:
-            gw.sip_engine = _build_sip_engine(settings, gw)
-        return gw
 
     # ================================================================
     # 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")
+        logger.info("   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
 
         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
+        # Hand off to the registered per-mode launcher (e.g. hold slayer)
-        if mode == CallMode.HOLD_SLAYER:
+        handler = self._mode_handlers.get(mode)
-            from services.hold_slayer import HoldSlayerService
+        if handler is not None:
-
+            handler(call, sip_leg_id, call_flow_id)
-            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}",
-            )
 
         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
+        pstn_leg_id = next(
-        for leg_id, cid in self.call_manager._call_legs.items():
+            (
-            if cid == call_id and leg_id != device_leg_id:
+                leg_id
-                pstn_leg_id = leg_id
+                for leg_id in self.call_manager.legs_for_call(call_id)
-                break
+                if leg_id != device_leg_id
+            ),
+            None,
+        )
 
         if pstn_leg_id:
             # Bridge the PSTN leg and device leg
@@ -342,8 +298,7 @@ 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()):
+        for leg_id in self.call_manager.legs_for_call(call_id):
-            if cid == 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 = [

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:
+@asynccontextmanager
-    """Dependency: yield an async database session."""
+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()

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
+        # === Composition root ===
-        gateway = AIPSTNGateway.from_config()
+        # 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
-
+        app.state.routing_service = routing_svc
-        # Start auxiliary services
-        from services.notification import NotificationService
-        from services.recording import RecordingService
 
         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")

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 db.database import session_scope
-        from sqlalchemy import select
+        from services import call_persistence as store
 
         try:
-            factory = get_session_factory()
+            async with session_scope() as session:
-            async with factory() as session:
+                row = await store.get_flow_by_number(session, phone_number)
-                result = await session.execute(
-                    select(StoredCallFlow).where(
-                        StoredCallFlow.phone_number == phone_number
-                    )
-                )
-                row = result.scalar_one_or_none()
                 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 session_scope() as session:
-            async with factory() as session:
+                if await store.get_flow(session, flow_id):
-                db_flow = StoredCallFlow(
+                    return f"Call flow '{flow_id}' already exists."
-                    id=flow_id,
+                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,13 +278,13 @@ 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():
+        legs = gateway.call_manager.legs_for_call(call_id)
-            if cid == call_id:
+        if not legs:
-                await gateway.sip_engine.send_dtmf(leg_id, digits)
-                return f"Sent DTMF '{digits}' on 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 db.database import session_scope
-        from sqlalchemy import select
+        from services import call_persistence as store
 
         try:
-            factory = get_session_factory()
+            async with session_scope() as session:
-            async with factory() as session:
+                record = await store.get_record(session, call_id)
-                result = await session.execute(
-                    select(CallRecord).where(CallRecord.id == call_id)
-                )
-                record = result.scalar_one_or_none()
                 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 db.database import session_scope
-        from sqlalchemy import select
+        from services import call_persistence as store
 
         try:
-            factory = get_session_factory()
+            async with session_scope() as session:
-            async with factory() as session:
+                record = await store.get_record(session, call_id)
-                result = await session.execute(
-                    select(CallRecord).where(CallRecord.id == call_id)
-                )
-                record = result.scalar_one_or_none()
                 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 db.database import session_scope
-        from sqlalchemy import select
+        from services import call_persistence as store
 
         try:
-            factory = get_session_factory()
+            async with session_scope() as session:
-            async with factory() as session:
+                records = await store.search_history(
-                query = select(CallRecord).order_by(
+                    session,
-                    CallRecord.started_at.desc()
+                    number_contains=phone_number or None,
-                ).limit(limit)
+                    intent_contains=intent or None,
-
+                    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()
 
                 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 db.database import session_scope
-        from sqlalchemy import select
+        from services import call_persistence as store
 
         try:
-            factory = get_session_factory()
+            async with session_scope() as session:
-            async with factory() as session:
+                rows = await store.list_flows(session)
-                result = await session.execute(select(StoredCallFlow))
-                rows = result.scalars().all()
                 flows = [
                     {
                         "id": r.id,

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]

services/call_analytics.py

@@ -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]
-            ),
-        }

services/call_persistence.py

@@ -1,25 +1,168 @@
 """
-Call Persistence — Writes completed calls and their transcript chunks
+Call Persistence — the data-access layer for calls and call flows.
-to the database when CallManager.end_call() fires.
+
+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}")

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

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

services/receptionist.py

@@ -27,12 +27,100 @@ from models.routing import RoutingAction, RoutingActionType, RoutingDecision
 logger = logging.getLogger(__name__)
 
 
-class ReceptionistService:
+def _extract_number(sip_uri: str) -> str:
-    """Drives the receptionist state machine for a single inbound call."""
+    """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

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() == []