diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..d58477c --- /dev/null +++ b/alembic.ini @@ -0,0 +1,42 @@ +# Alembic configuration. The database URL is not set here — env.py +# reads it from config.Settings (environment / .env), so CLI runs and +# app startup migrate the same database the app uses. + +[alembic] +script_location = db/migrations +prepend_sys_path = . +path_separator = os + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/api/call_history.py b/api/call_history.py index 59edbc4..aff3f5b 100644 --- a/api/call_history.py +++ b/api/call_history.py @@ -37,23 +37,7 @@ async def list_history( limit=limit, offset=offset, ) - return [ - { - "id": r.id, - "direction": r.direction, - "remote_number": r.remote_number, - "status": r.status, - "mode": r.mode, - "intent": r.intent, - "started_at": r.started_at.isoformat() if r.started_at else None, - "ended_at": r.ended_at.isoformat() if r.ended_at else None, - "duration": r.duration, - "hold_time": r.hold_time, - "device_used": r.device_used, - "summary": r.summary, - } - for r in rows - ] + return [store.record_summary(r) for r in rows] @router.get("/{call_id}/record") @@ -62,40 +46,14 @@ async def get_record(call_id: str, db: AsyncSession = Depends(get_db)): row = await store.get_record(db, call_id) if not row: raise HTTPException(status_code=404, detail=f"Call {call_id} not found") - return { - "id": row.id, - "direction": row.direction, - "remote_number": row.remote_number, - "status": row.status, - "mode": row.mode, - "intent": row.intent, - "started_at": row.started_at.isoformat() if row.started_at else None, - "ended_at": row.ended_at.isoformat() if row.ended_at else None, - "duration": row.duration, - "hold_time": row.hold_time, - "device_used": row.device_used, - "summary": row.summary, - "action_items": row.action_items, - "sentiment": row.sentiment, - "call_flow_id": row.call_flow_id, - "classification_timeline": row.classification_timeline, - } + return store.record_detail(row) @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 store.get_transcript_chunks(db, call_id) - return [ - { - "seq": c.seq, - "t_offset_ms": c.t_offset_ms, - "speaker": c.speaker, - "text": c.text, - "confidence": c.confidence, - } - for c in rows - ] + return [store.chunk_to_dict(c) for c in rows] @router.get("/{call_id}/recording") diff --git a/api/calls.py b/api/calls.py index eb46c5d..e6f84c6 100644 --- a/api/calls.py +++ b/api/calls.py @@ -40,12 +40,7 @@ async def make_call( call_flow_id=request.call_flow_id, services=request.services, ) - return CallResponse( - call_id=call.id, - status=call.status.value, - number=request.number, - mode=request.mode.value, - ) + return CallResponse.from_call(call) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: @@ -81,11 +76,8 @@ async def hold_slayer( call_flow_id=request.call_flow_id, device=request.transfer_to, ) - return CallResponse( - call_id=call.id, - status="navigating_ivr", - number=request.number, - mode="hold_slayer", + return CallResponse.from_call( + call, message="Hold Slayer activated. I'll ring you when a human picks up. ☕", ) except ValueError as e: @@ -113,21 +105,7 @@ async def get_call( if not call: raise HTTPException(status_code=404, detail=f"Call {call_id} not found") - return CallStatusResponse( - call_id=call.id, - status=call.status.value, - direction=call.direction, - remote_number=call.remote_number, - mode=call.mode.value, - duration=call.duration, - hold_time=call.hold_time, - audio_type=call.current_classification.value, - intent=call.intent, - transcript_excerpt=call.transcript[-500:] if call.transcript else None, - classification_history=call.classification_history[-50:], - current_step=call.current_step_id, - services=call.services, - ) + return CallStatusResponse.from_call(call) @router.post("/{call_id}/transfer") diff --git a/api/devices.py b/api/devices.py index aaded66..b550fa1 100644 --- a/api/devices.py +++ b/api/devices.py @@ -1,19 +1,19 @@ """ Device Management API — Register and manage phones/softphones. +Row mapping lives in call_persistence; this layer works with the +Device domain model only. """ import uuid -from datetime import datetime from fastapi import APIRouter, Depends, HTTPException -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 Device as DeviceDB from db.database import get_db -from models.device import Device, DeviceCreate, DeviceStatus, DeviceUpdate +from models.device import Device, DeviceCreate, DeviceUpdate +from services import call_persistence as store router = APIRouter() @@ -25,45 +25,18 @@ async def register_device( db: AsyncSession = Depends(get_db), ): """Register a new device with the gateway.""" - device_id = f"dev_{uuid.uuid4().hex[:8]}" - - # Save to DB - db_device = DeviceDB( - id=device_id, - name=device.name, - type=device.type.value, - sip_uri=device.sip_uri, - phone_number=device.phone_number, - priority=device.priority, - capabilities=device.capabilities, - is_online="false", - ) - db.add(db_device) - await db.flush() - - # Register with gateway - dev = Device(id=device_id, **device.model_dump()) + dev = Device(id=f"dev_{uuid.uuid4().hex[:8]}", **device.model_dump()) + await store.create_device_row(db, dev) gateway.register_device(dev) - return dev -@router.get("/", response_model=list[DeviceStatus]) +@router.get("/", response_model=list[Device]) async def list_devices( gateway: AIPSTNGateway = Depends(get_gateway), ): """List all registered devices and their status.""" - return [ - DeviceStatus( - id=d.id, - name=d.name, - type=d.type, - is_online=d.is_online, - last_seen=d.last_seen, - can_receive_call=d.can_receive_call, - ) - for d in gateway.devices.values() - ] + return list(gateway.devices.values()) @router.get("/{device_id}", response_model=Device) @@ -90,22 +63,11 @@ async def update_device( if not device: raise HTTPException(status_code=404, detail=f"Device {device_id} not found") - # Update in-memory update_data = update.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(device, key, value) - # Update in DB - result = await db.execute( - select(DeviceDB).where(DeviceDB.id == device_id) - ) - db_device = result.scalar_one_or_none() - if db_device: - for key, value in update_data.items(): - if key == "type" and value is not None: - value = value.value if hasattr(value, "value") else value - setattr(db_device, key, value) - + await store.update_device_row(db, device_id, update_data) return device @@ -120,12 +82,5 @@ async def unregister_device( raise HTTPException(status_code=404, detail=f"Device {device_id} not found") gateway.unregister_device(device_id) - - result = await db.execute( - select(DeviceDB).where(DeviceDB.id == device_id) - ) - db_device = result.scalar_one_or_none() - if db_device: - await db.delete(db_device) - + await store.delete_device_row(db, device_id) return {"status": "unregistered", "device_id": device_id} diff --git a/core/call_manager.py b/core/call_manager.py index 1640996..dea04ab 100644 --- a/core/call_manager.py +++ b/core/call_manager.py @@ -5,15 +5,19 @@ Central nervous system of the gateway. Tracks all active calls, publishes events, and coordinates between SIP engine and services. """ -import asyncio import logging import uuid -from collections.abc import AsyncIterator from datetime import datetime from typing import Optional -from core.event_bus import EventBus, EventSubscription -from models.call import ActiveCall, AudioClassification, CallMode, CallStatus, ClassificationResult +from core.event_bus import EventBus +from models.call import ( + ActiveCall, + CallMode, + CallStatus, + ClassificationResult, + TranscriptEntry, +) from models.events import EventType, GatewayEvent logger = logging.getLogger(__name__) @@ -26,10 +30,11 @@ class CallManager: The single source of truth for what's happening on the gateway. """ - def __init__(self, event_bus: EventBus, on_call_ended=None): + def __init__(self, event_bus: EventBus, on_call_created=None, 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_created = on_call_created # async callback(call) self._on_call_ended = on_call_ended # async callback(call, final_status) # ================================================================ @@ -67,6 +72,14 @@ class CallManager: message=f"📞 Calling {remote_number} ({mode.value})", )) + # Durable in_progress row — a crash mid-call must not erase the + # call from history. The hook does its own retrying/logging. + if self._on_call_created is not None: + try: + await self._on_call_created(call) + except Exception as e: + logger.warning(f"on_call_created hook failed for {call_id}: {e}") + return call async def update_status(self, call_id: str, status: CallStatus) -> None: @@ -135,18 +148,26 @@ class CallManager: message=f"🎵 Audio: {result.audio_type.value} ({result.confidence:.0%})", )) - async def add_transcript(self, call_id: str, text: str) -> None: - """Add a transcript chunk to a call.""" + async def add_transcript( + self, call_id: str, text: str, speaker: str = "unknown" + ) -> None: + """Add a transcript entry to a call, stamped with its offset.""" call = self._active_calls.get(call_id) if not call: return - call.transcript_chunks.append(text) + anchor = call.connected_at or call.started_at + entry = TranscriptEntry( + t_offset_ms=int((datetime.now() - anchor).total_seconds() * 1000), + speaker=speaker, + text=text, + ) + call.transcript_chunks.append(entry) await self.event_bus.publish(GatewayEvent( type=EventType.TRANSCRIPT_CHUNK, call_id=call_id, - data={"text": text}, + data={"text": text, "speaker": speaker, "t_offset_ms": entry.t_offset_ms}, message=f"📝 '{text[:80]}...' " if len(text) > 80 else f"📝 '{text}'", )) diff --git a/core/gateway.py b/core/gateway.py index 608f058..5f8f94e 100644 --- a/core/gateway.py +++ b/core/gateway.py @@ -86,11 +86,16 @@ class AIPSTNGateway: self, settings: Settings, sip_engine: Optional[SIPEngine] = None, + on_call_created=None, on_call_ended=None, ): self.settings = settings self.event_bus = EventBus() - self.call_manager = CallManager(self.event_bus, on_call_ended=on_call_ended) + self.call_manager = CallManager( + self.event_bus, + on_call_created=on_call_created, + on_call_ended=on_call_ended, + ) self.media_pipeline = MediaPipeline(sample_rate=16000) self.sip_engine: SIPEngine = sip_engine or MockSIPEngine() diff --git a/db/database.py b/db/database.py index 5ddd6cf..f7bbe2a 100644 --- a/db/database.py +++ b/db/database.py @@ -6,7 +6,7 @@ PostgreSQL via asyncpg + SQLAlchemy async. from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from datetime import datetime +from pathlib import Path from sqlalchemy import ( JSON, @@ -51,7 +51,6 @@ class CallRecord(Base): hold_time = Column(Integer, default=0) # seconds spent on hold device_used = Column(String) recording_path = Column(String, nullable=True) - transcript = Column(Text, nullable=True) summary = Column(Text, nullable=True) action_items = Column(JSON, nullable=True) sentiment = Column(String, nullable=True) @@ -94,7 +93,7 @@ class Device(Base): sip_uri = Column(String, nullable=True) # sip:robert@gateway.helu.ca phone_number = Column(String, nullable=True) # For PSTN devices priority = Column(Integer, default=10) # Routing priority (lower = higher priority) - is_online = Column(String, default="false") + is_online = Column(Boolean, default=False, nullable=False) capabilities = Column(JSON, default=list) # ["voice", "video", "sms"] dnd = Column(Boolean, default=False, nullable=False) last_seen = Column(DateTime, nullable=True) @@ -211,11 +210,31 @@ async def get_db() -> AsyncIterator[AsyncSession]: yield session +# The autogenerated baseline revision — a schema created by the old +# create_all path is identical to it, so such databases are stamped +# here and then migrated forward like any other. +_BASELINE_REVISION = "1173a71329ed" + + +def _upgrade_to_head(connection) -> None: + from alembic import command + from alembic.config import Config + from sqlalchemy import inspect + + cfg = Config(str(Path(__file__).resolve().parent.parent / "alembic.ini")) + cfg.attributes["connection"] = connection + + inspector = inspect(connection) + if not inspector.has_table("alembic_version") and inspector.has_table("call_records"): + command.stamp(cfg, _BASELINE_REVISION) + command.upgrade(cfg, "head") + + async def init_db(): - """Create all tables. For development; use Alembic migrations in production.""" + """Bring the schema to Alembic head (tests create tables directly).""" engine = get_engine() async with engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) + await conn.run_sync(_upgrade_to_head) async def close_db(): diff --git a/db/migrations/env.py b/db/migrations/env.py new file mode 100644 index 0000000..c55fcd0 --- /dev/null +++ b/db/migrations/env.py @@ -0,0 +1,68 @@ +""" +Alembic environment — async engine against Base.metadata. + +Two entry paths: +- CLI (``alembic upgrade head``): builds an async engine from + Settings.database_url and runs migrations on it. +- App startup (db.database.init_db): passes an already-open + connection via ``config.attributes["connection"]`` so migrations + run inside the app's engine instead of opening a second one. +""" + +import asyncio +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import pool +from sqlalchemy.ext.asyncio import create_async_engine + +from config import get_settings +from db.database import Base + +config = context.config + +# Only configure logging on standalone CLI runs — inside the app this +# would clobber uvicorn's logger setup. +if config.config_file_name is not None and config.attributes.get("connection") is None: + fileConfig(config.config_file_name, disable_existing_loggers=False) + +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + """Emit SQL to stdout without a live connection (--sql mode).""" + context.configure( + url=get_settings().database_url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + engine = create_async_engine(get_settings().database_url, poolclass=pool.NullPool) + async with engine.connect() as connection: + await connection.run_sync(do_run_migrations) + await engine.dispose() + + +def run_migrations_online() -> None: + connection = config.attributes.get("connection") + if connection is not None: + do_run_migrations(connection) + else: + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/db/migrations/script.py.mako b/db/migrations/script.py.mako new file mode 100644 index 0000000..fbc4b07 --- /dev/null +++ b/db/migrations/script.py.mako @@ -0,0 +1,26 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/db/migrations/versions/1173a71329ed_baseline_schema.py b/db/migrations/versions/1173a71329ed_baseline_schema.py new file mode 100644 index 0000000..2330617 --- /dev/null +++ b/db/migrations/versions/1173a71329ed_baseline_schema.py @@ -0,0 +1,129 @@ +"""baseline schema + +Revision ID: 1173a71329ed +Revises: +Create Date: 2026-07-10 07:19:08.321778 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '1173a71329ed' +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('call_flows', + sa.Column('id', sa.String(), nullable=False), + sa.Column('name', sa.String(), nullable=False), + sa.Column('phone_number', sa.String(), nullable=False), + sa.Column('description', sa.Text(), nullable=True), + sa.Column('steps', sa.JSON(), nullable=False), + sa.Column('last_verified', sa.DateTime(), nullable=True), + sa.Column('avg_hold_time', sa.Integer(), nullable=True), + sa.Column('success_rate', sa.Float(), nullable=True), + sa.Column('times_used', sa.Integer(), nullable=True), + sa.Column('last_used', sa.DateTime(), nullable=True), + sa.Column('notes', sa.Text(), nullable=True), + sa.Column('tags', sa.JSON(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.Column('updated_at', sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_call_flows_phone_number'), 'call_flows', ['phone_number'], unique=False) + op.create_table('call_records', + sa.Column('id', sa.String(), nullable=False), + sa.Column('direction', sa.String(), nullable=False), + sa.Column('remote_number', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('mode', sa.String(), nullable=False), + sa.Column('intent', sa.Text(), nullable=True), + sa.Column('started_at', sa.DateTime(), nullable=True), + sa.Column('ended_at', sa.DateTime(), nullable=True), + sa.Column('duration', sa.Integer(), nullable=True), + sa.Column('hold_time', sa.Integer(), nullable=True), + sa.Column('device_used', sa.String(), nullable=True), + sa.Column('recording_path', sa.String(), nullable=True), + sa.Column('transcript', sa.Text(), nullable=True), + sa.Column('summary', sa.Text(), nullable=True), + sa.Column('action_items', sa.JSON(), nullable=True), + sa.Column('sentiment', sa.String(), nullable=True), + sa.Column('call_flow_id', sa.String(), nullable=True), + sa.Column('classification_timeline', sa.JSON(), nullable=True), + sa.Column('metadata', sa.JSON(), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_call_records_remote_number'), 'call_records', ['remote_number'], unique=False) + op.create_table('devices', + sa.Column('id', sa.String(), nullable=False), + sa.Column('name', sa.String(), nullable=False), + sa.Column('type', sa.String(), nullable=False), + sa.Column('sip_uri', sa.String(), nullable=True), + sa.Column('phone_number', sa.String(), nullable=True), + sa.Column('priority', sa.Integer(), nullable=True), + sa.Column('is_online', sa.String(), nullable=True), + sa.Column('capabilities', sa.JSON(), nullable=True), + sa.Column('dnd', sa.Boolean(), nullable=False), + sa.Column('last_seen', sa.DateTime(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.Column('updated_at', sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + op.create_table('recordings', + sa.Column('id', sa.String(), nullable=False), + sa.Column('call_id', sa.String(), nullable=False), + sa.Column('path', sa.String(), nullable=False), + sa.Column('format', sa.String(), nullable=True), + sa.Column('duration_s', sa.Float(), nullable=True), + sa.Column('size_bytes', sa.Integer(), nullable=True), + sa.Column('channels', sa.Integer(), nullable=True), + sa.Column('started_at', sa.DateTime(), nullable=True), + sa.Column('ended_at', sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_recordings_call_id'), 'recordings', ['call_id'], unique=False) + op.create_table('routing_rules', + sa.Column('id', sa.String(), nullable=False), + sa.Column('name', sa.String(), nullable=False), + sa.Column('priority', sa.Integer(), nullable=False), + sa.Column('enabled', sa.Boolean(), nullable=False), + sa.Column('match', sa.JSON(), nullable=False), + sa.Column('action', sa.JSON(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.Column('updated_at', sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + op.create_table('transcript_chunks', + sa.Column('id', sa.String(), nullable=False), + sa.Column('call_id', sa.String(), nullable=False), + sa.Column('seq', sa.Integer(), nullable=False), + sa.Column('t_offset_ms', sa.Integer(), nullable=True), + sa.Column('speaker', sa.String(), nullable=True), + sa.Column('text', sa.Text(), nullable=False), + sa.Column('confidence', sa.Float(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_transcript_chunks_call_id'), 'transcript_chunks', ['call_id'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_transcript_chunks_call_id'), table_name='transcript_chunks') + op.drop_table('transcript_chunks') + op.drop_table('routing_rules') + op.drop_index(op.f('ix_recordings_call_id'), table_name='recordings') + op.drop_table('recordings') + op.drop_table('devices') + op.drop_index(op.f('ix_call_records_remote_number'), table_name='call_records') + op.drop_table('call_records') + op.drop_index(op.f('ix_call_flows_phone_number'), table_name='call_flows') + op.drop_table('call_flows') + # ### end Alembic commands ### diff --git a/db/migrations/versions/5187577efc23_drop_dead_transcript_column_boolean_is_.py b/db/migrations/versions/5187577efc23_drop_dead_transcript_column_boolean_is_.py new file mode 100644 index 0000000..9eaf9f5 --- /dev/null +++ b/db/migrations/versions/5187577efc23_drop_dead_transcript_column_boolean_is_.py @@ -0,0 +1,45 @@ +"""drop dead transcript column, boolean is_online + +Revision ID: 5187577efc23 +Revises: 1173a71329ed +Create Date: 2026-07-10 07:19:40.741327 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '5187577efc23' +down_revision: Union[str, None] = '1173a71329ed' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Transcript text lives solely in transcript_chunks rows now. + op.drop_column('call_records', 'transcript') + + # String "true"/"false" (or NULL) -> real boolean; NULLs become false. + # batch mode so the table-recreate path works on SQLite too. + with op.batch_alter_table('devices') as batch_op: + batch_op.alter_column( + 'is_online', + existing_type=sa.VARCHAR(), + type_=sa.Boolean(), + nullable=False, + postgresql_using="coalesce(lower(is_online) in ('true', 't', '1'), false)", + ) + + +def downgrade() -> None: + with op.batch_alter_table('devices') as batch_op: + batch_op.alter_column( + 'is_online', + existing_type=sa.Boolean(), + type_=sa.VARCHAR(), + nullable=True, + postgresql_using="case when is_online then 'true' else 'false' end", + ) + op.add_column('call_records', sa.Column('transcript', sa.TEXT(), nullable=True)) diff --git a/main.py b/main.py index dd11fc7..d9a95fd 100644 --- a/main.py +++ b/main.py @@ -26,7 +26,7 @@ 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.call_persistence import persist_call_on_create, persist_call_on_end from services.hold_slayer import HoldSlayerService from services.notification import NotificationService from services.receptionist import ReceptionistService @@ -139,7 +139,11 @@ async def lifespan(app: FastAPI): # === 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) + gateway = AIPSTNGateway( + settings=settings, + on_call_created=persist_call_on_create, + on_call_ended=persist_call_on_end, + ) classifier = AudioClassifier(settings.classifier) transcription = TranscriptionService(settings.speaches) diff --git a/models/call.py b/models/call.py index 785e795..d9863c4 100644 --- a/models/call.py +++ b/models/call.py @@ -55,6 +55,14 @@ class ClassificationResult(BaseModel): details: Optional[dict] = None # Extra analysis data +class TranscriptEntry(BaseModel): + """One transcribed utterance, offset from call start for seek.""" + + t_offset_ms: int + speaker: str = "unknown" # caller / agent / receptionist / unknown + text: str + + class ActiveCall(BaseModel): """In-memory state for an active call.""" @@ -71,7 +79,7 @@ class ActiveCall(BaseModel): hold_started_at: Optional[datetime] = None current_classification: AudioClassification = AudioClassification.UNKNOWN classification_history: list[ClassificationResult] = Field(default_factory=list) - transcript_chunks: list[str] = Field(default_factory=list) + transcript_chunks: list[TranscriptEntry] = Field(default_factory=list) current_step_id: Optional[str] = None # Current position in call flow services: list[str] = Field(default_factory=list) # Active services on this call @@ -92,7 +100,7 @@ class ActiveCall(BaseModel): @property def transcript(self) -> str: """Full transcript so far.""" - return "\n".join(self.transcript_chunks) + return "\n".join(e.text for e in self.transcript_chunks) def summary(self) -> dict: """Compact summary for list views.""" @@ -145,6 +153,16 @@ class CallResponse(BaseModel): mode: str message: Optional[str] = None + @classmethod + def from_call(cls, call: "ActiveCall", message: Optional[str] = None) -> "CallResponse": + return cls( + call_id=call.id, + status=call.status.value, + number=call.remote_number, + mode=call.mode.value, + message=message, + ) + class CallStatusResponse(BaseModel): """Full status of an active or completed call.""" @@ -163,6 +181,25 @@ class CallStatusResponse(BaseModel): current_step: Optional[str] = None services: list[str] = Field(default_factory=list) + @classmethod + def from_call(cls, call: "ActiveCall") -> "CallStatusResponse": + """The one ActiveCall → status-response mapping.""" + return cls( + call_id=call.id, + status=call.status.value, + direction=call.direction, + remote_number=call.remote_number, + mode=call.mode.value, + duration=call.duration, + hold_time=call.hold_time, + audio_type=call.current_classification.value, + intent=call.intent, + transcript_excerpt=call.transcript[-500:] if call.transcript else None, + classification_history=call.classification_history[-20:], + current_step=call.current_step_id, + services=call.services, + ) + class TransferRequest(BaseModel): """Request to transfer a call to a device.""" diff --git a/models/device.py b/models/device.py index 641b538..481b766 100644 --- a/models/device.py +++ b/models/device.py @@ -8,7 +8,7 @@ from datetime import datetime from enum import Enum from typing import Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, computed_field class DeviceType(str, Enum): @@ -43,6 +43,7 @@ class Device(DeviceBase): created_at: Optional[datetime] = None updated_at: Optional[datetime] = None + @computed_field # serialized so API consumers see routability directly @property def can_receive_call(self) -> bool: """Can this device receive a call right now?""" @@ -71,14 +72,3 @@ class DeviceUpdate(BaseModel): phone_number: Optional[str] = None priority: Optional[int] = None capabilities: Optional[list[str]] = None - - -class DeviceStatus(BaseModel): - """Lightweight device status for list views.""" - - id: str - name: str - type: DeviceType - is_online: bool - last_seen: Optional[datetime] = None - can_receive_call: bool diff --git a/pyproject.toml b/pyproject.toml index 70b0676..ed8ab45 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,3 +70,7 @@ line-length = 100 [tool.ruff.lint] select = ["E", "F", "I", "N", "W", "UP"] + +[tool.ruff.lint.per-file-ignores] +# Alembic-generated migrations keep the standard template style. +"db/migrations/versions/*" = ["E501", "UP007", "UP035", "W291"] diff --git a/services/call_persistence.py b/services/call_persistence.py index d12c8c5..6390eda 100644 --- a/services/call_persistence.py +++ b/services/call_persistence.py @@ -22,8 +22,10 @@ from db.database import ( TranscriptChunk, session_scope, ) +from db.database import Device as DeviceRow from models.call import ActiveCall, CallStatus from models.call_flow import CallFlow, CallFlowStep +from models.device import Device logger = logging.getLogger(__name__) @@ -45,6 +47,45 @@ def flow_to_model(row: StoredCallFlow) -> CallFlow: ) +def record_summary(row: CallRecord) -> dict: + """The one CallRecord-row → list-item mapping.""" + return { + "id": row.id, + "direction": row.direction, + "remote_number": row.remote_number, + "status": row.status, + "mode": row.mode, + "intent": row.intent, + "started_at": row.started_at.isoformat() if row.started_at else None, + "ended_at": row.ended_at.isoformat() if row.ended_at else None, + "duration": row.duration, + "hold_time": row.hold_time, + "device_used": row.device_used, + "summary": row.summary, + } + + +def record_detail(row: CallRecord) -> dict: + """Full CallRecord-row mapping, superset of record_summary.""" + return record_summary(row) | { + "action_items": row.action_items, + "sentiment": row.sentiment, + "call_flow_id": row.call_flow_id, + "classification_timeline": row.classification_timeline, + } + + +def chunk_to_dict(row: TranscriptChunk) -> dict: + """The one TranscriptChunk-row → dict mapping.""" + return { + "seq": row.seq, + "t_offset_ms": row.t_offset_ms, + "speaker": row.speaker, + "text": row.text, + "confidence": row.confidence, + } + + # ================================================================ # Call flows # ================================================================ @@ -95,6 +136,49 @@ async def create_flow( return row +# ================================================================ +# Devices +# ================================================================ + +async def create_device_row(session: AsyncSession, device: Device) -> None: + """The one Device-model → row mapping.""" + session.add(DeviceRow( + id=device.id, + name=device.name, + type=device.type.value, + sip_uri=device.sip_uri, + phone_number=device.phone_number, + priority=device.priority, + capabilities=device.capabilities, + is_online=device.is_online, + )) + await session.flush() + + +async def update_device_row( + session: AsyncSession, device_id: str, values: dict +) -> None: + result = await session.execute( + select(DeviceRow).where(DeviceRow.id == device_id) + ) + row = result.scalar_one_or_none() + if row is None: + return + for key, value in values.items(): + if key == "type" and value is not None: + value = value.value if hasattr(value, "value") else value + setattr(row, key, value) + + +async def delete_device_row(session: AsyncSession, device_id: str) -> None: + result = await session.execute( + select(DeviceRow).where(DeviceRow.id == device_id) + ) + row = result.scalar_one_or_none() + if row is not None: + await session.delete(row) + + # ================================================================ # Call history / records # ================================================================ @@ -156,71 +240,96 @@ async def latest_recording( return result.scalars().first() +async def persist_call_on_create(call: ActiveCall) -> None: + """Insert an in_progress CallRecord the moment a call starts. + + Wired into CallManager as its on_call_created hook — a crash + mid-call leaves this row behind instead of erasing the call from + history. persist_call_on_end updates it to the terminal state. + """ + await _with_retry(_insert_in_progress_record, call) + + async def persist_call_on_end(call: ActiveCall, final_status: CallStatus) -> None: - """Insert a CallRecord and any transcript chunks for `call`. + """Finalize the CallRecord and write transcript chunks for `call`. Wired into CallManager as its on_call_ended hook by the - composition root in main.py. Retries briefly — losing the row - means the call never happened as far as history is concerned, so - the final failure logs at ERROR with the payload identifiers. + composition root in main.py. """ + await _with_retry(_finalize_call_record, call, final_status) + + +async def _with_retry(write, call: ActiveCall, *args) -> None: + """Losing the row means the call never happened as far as history + is concerned, so the final failure logs at ERROR with identifiers.""" for attempt in range(3): try: - await _write_call_record(call, final_status) + await write(call, *args) return except Exception as e: if attempt == 2: logger.error( - f"Call record lost: id={call.id} number={call.remote_number} " - f"status={final_status.value}: {e}" + f"Call record lost ({write.__name__}): id={call.id} " + f"number={call.remote_number}: {e}" ) return await asyncio.sleep(2**attempt) -async def _write_call_record(call: ActiveCall, final_status: CallStatus) -> None: +async def _insert_in_progress_record(call: ActiveCall) -> None: async with session_scope() as session: - record = CallRecord( + session.add(CallRecord( id=call.id, direction=call.direction, remote_number=call.remote_number, - status=final_status.value, + status="in_progress", mode=call.mode.value, intent=call.intent, started_at=call.started_at, - ended_at=datetime.now(), - duration=int(call.duration), - hold_time=int(call.hold_time), device_used=call.device, call_flow_id=call.call_flow_id, - classification_timeline=[ - { - "timestamp": c.timestamp, - "audio_type": c.audio_type.value, - "confidence": c.confidence, - } - for c in call.classification_history - ], metadata_={"services": list(call.services)}, - ) - session.add(record) + )) - # Each transcript chunk gets its own row with a sequence number - # so the dashboard can render them in order with click-to-seek. - for seq, text in enumerate(call.transcript_chunks): - speaker = "unknown" - payload = text - if ":" in text: - head, rest = text.split(":", 1) - head = head.strip().lower() - if head in {"caller", "agent", "receptionist", "caller_message"}: - speaker = head if head != "caller_message" else "caller" - payload = rest.strip() + +async def _finalize_call_record(call: ActiveCall, final_status: CallStatus) -> None: + async with session_scope() as session: + record = await get_record(session, call.id) + if record is None: + # The create-time insert failed (or predates the hook); + # write the whole row now instead. + record = CallRecord(id=call.id) + session.add(record) + + record.direction = call.direction + record.remote_number = call.remote_number + record.status = final_status.value + record.mode = call.mode.value + record.intent = call.intent + record.started_at = call.started_at + record.ended_at = datetime.now() + record.duration = int(call.duration) + record.hold_time = int(call.hold_time) + record.device_used = call.device + record.call_flow_id = call.call_flow_id + record.classification_timeline = [ + { + "timestamp": c.timestamp, + "audio_type": c.audio_type.value, + "confidence": c.confidence, + } + for c in call.classification_history + ] + record.metadata_ = {"services": list(call.services)} + + # Each transcript entry gets its own row with a sequence number + # and real offset so the dashboard can render click-to-seek. + for seq, entry in enumerate(call.transcript_chunks): session.add(TranscriptChunk( id=f"tc_{uuid.uuid4().hex[:10]}", call_id=call.id, seq=seq, - t_offset_ms=0, - speaker=speaker, - text=payload, + t_offset_ms=entry.t_offset_ms, + speaker=entry.speaker, + text=entry.text, )) diff --git a/services/receptionist.py b/services/receptionist.py index 8999b30..4cf21ca 100644 --- a/services/receptionist.py +++ b/services/receptionist.py @@ -134,13 +134,9 @@ class ReceptionistService: transcript = await self._listen(call, sip_leg_id) if transcript: - call.transcript_chunks.append(f"caller: {transcript}") - await self.gateway.event_bus.publish(GatewayEvent( - type=EventType.TRANSCRIPT_CHUNK, - call_id=call.id, - data={"text": transcript, "speaker": "caller"}, - message=f"📝 caller: {transcript[:80]}", - )) + await self.gateway.call_manager.add_transcript( + call.id, transcript, speaker="caller" + ) classification = await self._classify(call, transcript, routing_decision) call.intent = classification.get("intent") @@ -390,7 +386,9 @@ class ReceptionistService: await self._service_error(call.id, "transcription", e) if message_text: - call.transcript_chunks.append(f"caller_message: {message_text}") + await self.gateway.call_manager.add_transcript( + call.id, message_text, speaker="caller" + ) await self.gateway.event_bus.publish(GatewayEvent( type=EventType.RECEPTIONIST_MESSAGE_SAVED, diff --git a/tests/test_data_layer.py b/tests/test_data_layer.py new file mode 100644 index 0000000..513b19c --- /dev/null +++ b/tests/test_data_layer.py @@ -0,0 +1,187 @@ +""" +Data-layer tests. + +Alembic migrations produce the schema the ORM models declare (and +adopt a pre-Alembic database); a call gets a durable in_progress row +the moment it starts; transcript entries carry real offsets; the +consolidated response models map from the domain in one place. +""" + +import asyncio + +import pytest +from sqlalchemy import Boolean, inspect, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import StaticPool + +import db.database as dbmod +from core.call_manager import CallManager +from core.event_bus import EventBus +from db.database import Base +from models.call import ActiveCall, CallStatus, CallStatusResponse +from models.device import Device, DeviceType +from services import call_persistence as store + +# ================================================================ +# Alembic migrations +# ================================================================ + +def _run_alembic(connection, revision: str) -> None: + from alembic import command + from alembic.config import Config + + cfg = Config("alembic.ini") + cfg.attributes["connection"] = connection + command.upgrade(cfg, revision) + + +def _schema_info(sync_conn) -> dict: + inspector = inspect(sync_conn) + return { + "tables": set(inspector.get_table_names()) - {"alembic_version"}, + "call_record_cols": {c["name"] for c in inspector.get_columns("call_records")}, + "is_online_type": next( + c["type"] for c in inspector.get_columns("devices") + if c["name"] == "is_online" + ), + } + + +class TestMigrations: + async def test_upgrade_head_matches_models(self, tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/mig.db") + async with engine.begin() as conn: + await conn.run_sync(dbmod._upgrade_to_head) + async with engine.connect() as conn: + info = await conn.run_sync(_schema_info) + await engine.dispose() + + assert info["tables"] == set(Base.metadata.tables) + assert "transcript" not in info["call_record_cols"] + assert isinstance(info["is_online_type"], Boolean) + + async def test_adopts_pre_alembic_schema(self, tmp_path): + """A create_all-era database (baseline schema, no alembic_version) + is stamped and migrated forward instead of failing.""" + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/legacy.db") + async with engine.begin() as conn: + await conn.run_sync( + lambda c: _run_alembic(c, dbmod._BASELINE_REVISION) + ) + await conn.execute(text("DROP TABLE alembic_version")) + + async with engine.begin() as conn: + await conn.run_sync(dbmod._upgrade_to_head) + async with engine.connect() as conn: + info = await conn.run_sync(_schema_info) + await engine.dispose() + + assert "transcript" not in info["call_record_cols"] + assert isinstance(info["is_online_type"], Boolean) + + +# ================================================================ +# Durable call rows +# ================================================================ + +@pytest.fixture +async def mem_db(monkeypatch): + 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) + monkeypatch.setattr(dbmod, "_engine", engine) + monkeypatch.setattr(dbmod, "_session_factory", factory) + yield factory + await engine.dispose() + + +class TestDurableCallRows: + async def test_in_progress_row_from_the_start(self, mem_db): + cm = CallManager( + EventBus(), + on_call_created=store.persist_call_on_create, + on_call_ended=store.persist_call_on_end, + ) + call = await cm.create_call("+15551230000", intent="dispute a charge") + + async with mem_db() as session: + row = await store.get_record(session, call.id) + assert row is not None + assert row.status == "in_progress" + assert row.ended_at is None + assert row.intent == "dispute a charge" + + await cm.add_transcript(call.id, "hello, billing please", speaker="caller") + await cm.end_call(call.id, CallStatus.COMPLETED) + + async with mem_db() as session: + row = await store.get_record(session, call.id) + chunks = await store.get_transcript_chunks(session, call.id) + assert row.status == "completed" + assert row.ended_at is not None + assert [(c.seq, c.speaker, c.text) for c in chunks] == [ + (0, "caller", "hello, billing please") + ] + + async def test_end_without_create_still_writes_row(self, mem_db): + """If the create-time insert never happened, finalize inserts.""" + cm = CallManager(EventBus(), on_call_ended=store.persist_call_on_end) + call = await cm.create_call("+15551230001") + await cm.end_call(call.id, CallStatus.FAILED) + + async with mem_db() as session: + row = await store.get_record(session, call.id) + assert row is not None + assert row.status == "failed" + + +# ================================================================ +# Transcript offsets +# ================================================================ + +class TestTranscriptOffsets: + async def test_entries_carry_offset_and_speaker(self): + cm = CallManager(EventBus()) + call = await cm.create_call("+15551230002") + + await cm.add_transcript(call.id, "one") + await asyncio.sleep(0.02) + await cm.add_transcript(call.id, "two", speaker="agent") + + first, second = call.transcript_chunks + assert first.t_offset_ms >= 0 + assert second.t_offset_ms > first.t_offset_ms + assert first.speaker == "unknown" + assert second.speaker == "agent" + assert call.transcript == "one\ntwo" + + +# ================================================================ +# Consolidated response models +# ================================================================ + +class TestResponseModels: + def test_status_response_from_call(self): + call = ActiveCall(id="call_x", remote_number="+15550000000", intent="pay bill") + resp = CallStatusResponse.from_call(call) + assert resp.call_id == "call_x" + assert resp.status == "initiating" + assert resp.remote_number == "+15550000000" + assert resp.intent == "pay bill" + + def test_device_serializes_routability(self): + device = Device( + id="dev_1", + name="Desk Phone", + type=DeviceType.SIP_PHONE, + sip_uri="sip:desk@gw", + is_online=True, + ) + assert device.model_dump()["can_receive_call"] is True + device.dnd = True + assert device.model_dump()["can_receive_call"] is False