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

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

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

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}

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}'",
         ))
 

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()
 

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():

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()

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"}

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 ###

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

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)

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

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

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"]

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=[
+            metadata_={"services": list(call.services)},
+        ))
+
+
+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
-            ],
-            metadata_={"services": list(call.services)},
-        )
-        session.add(record)
+        ]
+        record.metadata_ = {"services": list(call.services)}
 
-        # 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()
+        # 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,
             ))

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,

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