r/hold-slayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stage 5: Alembic migrations, durable call rows, one transcript truth

Browse Source 
Alembic replaces create_all as the schema authority: async env.py
against Base.metadata (CLI and in-app entry paths share it via
config.attributes["connection"]), an autogenerated baseline of the
create_all-era schema, and init_db now runs upgrade head — stamping
the baseline first on a pre-Alembic database so existing deployments
adopt cleanly. create_all remains for tests only.

Calls are durable from the start: CallManager gains an
on_call_created hook (wired to persist_call_on_create) that inserts
an in_progress CallRecord the moment a call is created;
persist_call_on_end finalizes that same row. A SIGKILL mid-call now
leaves an in_progress row instead of erasing the call from history
(verified live against the dev database).

One transcript representation: ActiveCall.transcript_chunks holds
TranscriptEntry (t_offset_ms, speaker, text) — add_transcript stamps
real offsets from connect time, receptionist passes speaker instead
of encoding it into "caller: ..." strings, persisted chunks carry
real seek offsets, and the dead CallRecord.transcript Text column is
dropped by migration. Device.is_online migrates String → Boolean
(with a USING cast for existing rows).

Model de-triplication: CallResponse/CallStatusResponse build via
from_call classmethods (one ActiveCall→response mapping);
DeviceStatus deleted — can_receive_call is a computed field on
Device and the list endpoint returns the domain model; all row↔dict
and row↔domain mapping now lives in call_persistence.py
(record_summary/record_detail/chunk_to_dict + device row functions).

New tests/test_data_layer.py: upgrade-head-matches-models,
pre-Alembic adoption, durable in_progress rows, end-without-create
fallback, transcript offsets, consolidated response models.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Robert Helewka
2026-07-10 07:42:52 -04:00
parent 4048ce1db6
commit f7a11f2f20
18 changed files with 777 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
alembic.ini Normal file
View File

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

48
api/call_history.py
View File

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

30
api/calls.py
View File

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

65
api/devices.py
View File

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

39
core/call_manager.py
View File

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

7
core/gateway.py
View File

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

29
db/database.py
View File

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

68
db/migrations/env.py Normal file
View File

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

26
db/migrations/script.py.mako Normal file
View File

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

129
db/migrations/versions/1173a71329ed_baseline_schema.py Normal file
View File

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

45
db/migrations/versions/5187577efc23_drop_dead_transcript_column_boolean_is_.py Normal file
View File

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

8
main.py
View File

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

41
models/call.py
View File

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

14
models/device.py
View File

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

4
pyproject.toml
View File

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

173
services/call_persistence.py
View File

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

14
services/receptionist.py
View File

@@ -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,

187
tests/test_data_layer.py Normal file
View File

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