Function bodies 220 total

    async def generate(self, text: str) -> bytes:
        import httpx
        import scipy.signal
        import soundfile as sf

        async with httpx.AsyncClient() as client:
            # Request speech generation
            gen_resp = await client.post(
                f"{self.base_url}/generate",
                json={
                    "profile_id": self.profile_id,
                    "text": text,
                    "language": self.language,
                },
                timeout=30,
            )
            gen_resp.raise_for_status()
            audio_id = gen_resp.json()["id"]

            # Download WAV audio
            audio_resp = await client.get(
                f"{self.base_url}/audio/{audio_id}",
                timeout=30,
            )
            audio_resp.raise_for_status()

        # Decode WAV to float32 (VoiceBox outputs 24kHz)
        samples, orig_rate = sf.read(io.BytesIO(audio_resp.content), dtype="float32")
        if len(samples) == 0:
   

class WebSocketHandler:
    def __init__(self, config: AgentConfig, stt_engine=None, tts_engine=None):
        self.config = config
        self.stt_engine = stt_engine
        self.tts_engine = tts_engine

    def set_tts_engine(self, engine):
        """Hot-swap the TTS engine. New sessions get the new engine immediately."""
        self.tts_engine = engine

    async def handle_connection(self, request):
        """aiohttp WebSocket handler — registered as a route on the HTTP app."""
        ws = web.WebSocketResponse()
        await ws.prepare(request)

        db = await get_db()
        system_prompt = self.config.get_system_prompt()

        # Load active flows for this session
        flow_tracker = await FlowTracker.load_active_flows(db)

        session = ConversationSession(
            db, system_prompt, tts_engine=self.tts_engine, flow_tracker=flow_tracker
        )
        print(f"  [WS] New connection: call {session.call_id} ({len(flow_tracker.flows)} active flows)")

  

    def __init__(self, config: AgentConfig, stt_engine=None, tts_engine=None):
        self.config = config
        self.stt_engine = stt_engine
        self.tts_engine = tts_engine

    def set_tts_engine(self, engine):
        """Hot-swap the TTS engine. New sessions get the new engine immediately."""
        self.tts_engine = engine

    async def handle_connection(self, request):
        """aiohttp WebSocket handler — registered as a route on the HTTP app."""
        ws = web.WebSocketResponse()
        await ws.prepare(request)

        db = await get_db()
        system_prompt = self.config.get_system_prompt()

        # Load active flows for this session
        flow_tracker = await FlowTracker.load_active_flows(db)

        session = ConversationSession(
            db, system_prompt, tts_engine=self.tts_engine, flow_tracker=flow_tracker
        )
        print(f"  [WS] New connection: call {session.call_id} ({len(flow_tracker.flows)} active flows)")

        # Audio state per connection
        audio_buffer = bytearray()
        last_speech_time = 0
        is_speaking = False
        processing_audio = False

        async def send_callback(msg):
            try:
                await ws.send_json(msg)
            except (ConnectionResetError, Exception):
                pass

        try:
            await 

def _sanitize_row(row_dict):
    """Convert non-JSON-serializable values (datetime, date) to ISO strings."""
    for k, v in row_dict.items():
        if isinstance(v, datetime):
            row_dict[k] = v.isoformat()
        elif isinstance(v, date):
            row_dict[k] = v.isoformat()
    return row_dict

def _convert_query(query):
    """Convert SQLite-style `?` placeholders to PostgreSQL `$N` style."""
    counter = 0

    def _replace(match):
        nonlocal counter
        counter += 1
        return f"${counter}"

    return re.sub(r"\?", _replace, query)

def sql_today():
    """Return a SQL expression for 'today' that works on both backends.

    SQLite uses DATE('now'), PostgreSQL uses CURRENT_DATE.
    Called from query sites that need date comparisons.
    """
    from database.db import is_postgres
    return "CURRENT_DATE" if is_postgres() else "DATE('now')"

class DBSession:
    """Unified database session wrapping either aiosqlite or asyncpg."""

    def __init__(self, backend, resource):
        """
        backend: 'sqlite' or 'postgres'
        resource: aiosqlite.Connection (sqlite) or asyncpg.Pool (postgres)
        """
        self._backend = backend
        self._resource = resource

    @property
    def is_postgres(self):
        return self._backend == "postgres"

    async def fetch_one(self, query, params=None):
        """Execute query and return a single row as a dict, or None."""
        if self.is_postgres:
            q = _convert_query(query)
            row = await self._resource.fetchrow(q, *(params or []))
            return _sanitize_row(dict(row)) if row else None
        else:
            cursor = await self._resource.execute(query, params or [])
            row = await cursor.fetchone()
            if row is None:
                return None
            return {k: row[k] for k in row.keys()}

    async def fetch_a

    def __init__(self, backend, resource):
        """
        backend: 'sqlite' or 'postgres'
        resource: aiosqlite.Connection (sqlite) or asyncpg.Pool (postgres)
        """
        self._backend = backend
        self._resource = resource

    async def fetch_one(self, query, params=None):
        """Execute query and return a single row as a dict, or None."""
        if self.is_postgres:
            q = _convert_query(query)
            row = await self._resource.fetchrow(q, *(params or []))
            return _sanitize_row(dict(row)) if row else None
        else:
            cursor = await self._resource.execute(query, params or [])
            row = await cursor.fetchone()
            if row is None:
                return None
            return {k: row[k] for k in row.keys()}

    async def fetch_all(self, query, params=None):
        """Execute query and return all rows as a list of dicts."""
        if self.is_postgres:
            q = _convert_query(query)
            rows = await self._resource.fetch(q, *(params or []))
            return [_sanitize_row(dict(r)) for r in rows]
        else:
            cursor = await self._resource.execute(query, params or [])
            rows = await cursor.fetchall()
            return [{k: r[k] for k in r.keys()} for r in rows]

    async def fetch_val(self, query, params=None):
        """Execute query and return a single scalar value."""
        if self.is_postgres:
            q = _convert_query(query)
            return await self._resource.fetchval(q, *(params or []))
        else:
            cursor = await self._resource.execute(query, params or [])
            row = await cursor.fetchone()
            return row[0] if row else None

    async def execute(self, query, params=None):
        """Execute a query (INSERT/UPDATE/DELETE) without returning rows."""
        if self.is_postgres:
            q = _convert_query(query)
            await self._resource.execute(q, *(params or []))
        else:
            await self._resource.execute(query, params or [])

    async def insert_returning_id(self, query, params=None):
        """Execute an INSERT and return the new row's id.

        For PostgreSQL, appends RETURNING id to the query.
        For SQLite, uses cursor.lastrowid.
        """
        if self.is_postgres:
            q = _convert_query(query)
            if "RETURNING" not in q.upper():
                q = q.rstrip().rstrip(";") + " RETURNING id"
            return await self._resource.fetchval(q, *(params or []))
        else:
            cursor = await self._resource.execute(query, params or [])
            await self._resource.commit()
            return cursor.lastrowid

    async def execute_with_rowcount(self, query, params=None):
        """Execute a query and return the number of affected rows."""
        if self.is_postgres:
            q = _convert_query(query)
            status = await self._resource.execute(q, *(params or []))
            # asyncpg returns status like 'DELETE 1' or 'UPDATE 0'
            parts = status.split()
            return int(parts[-1]) if parts else 0
        else:
            cursor = await self._resource.execute(query, params or [])
            return cursor.rowcount

    async def commit(self):
        """Commit transaction. No-op for asyncpg (auto-commit)."""
        if not self.is_postgres:
            await self._resource.commit()

    async def close(self):
        """Close the session. No-op for asyncpg (pool-managed)."""
        if not self.is_postgres:
            await self._resource.close()

async def init_pg():
    """Create the PostgreSQL connection pool and initialize schema."""
    global _pool
    database_url = os.environ.get("DATABASE_URL", "")
    if not database_url:
        raise ValueError("DATABASE_URL environment variable is required for PostgreSQL")

    _pool = await asyncpg.create_pool(database_url, min_size=2, max_size=10)

    async with _pool.acquire() as conn:
        await conn.execute(PG_SCHEMA)

def get_pg_pool():
    """Return the global asyncpg connection pool."""
    return _pool

async def close_pg():
    """Close the PostgreSQL connection pool."""
    global _pool
    if _pool:
        await _pool.close()
        _pool = None

def is_postgres():
    """Check if the current backend is PostgreSQL."""
    if os.environ.get("DB_BACKEND", "").lower() == "postgres":
        return True
    return bool(os.environ.get("DATABASE_URL"))

async def init_db():
    """Initialize the database (SQLite or PostgreSQL based on DB_BACKEND)."""
    if is_postgres():
        from database.db_postgres import init_pg
        await init_pg()
    else:
        os.makedirs(os.path.dirname(DB_PATH) or ".", exist_ok=True)
        async with aiosqlite.connect(DB_PATH) as db:
            await db.executescript(SCHEMA)
            await db.commit()

async def get_db() -> DBSession:
    """Return a DBSession wrapping the appropriate backend."""
    if is_postgres():
        from database.db_postgres import get_pg_pool
        pool = get_pg_pool()
        return DBSession("postgres", pool)
    else:
        db = await aiosqlite.connect(DB_PATH)
        db.row_factory = aiosqlite.Row
        await db.execute("PRAGMA foreign_keys = ON")
        return DBSession("sqlite", db)

async def close_db_pool():
    """Close the PostgreSQL pool on shutdown. No-op for SQLite."""
    if is_postgres():
        from database.db_postgres import close_pg
        await close_pg()

class Customer:
    id: Optional[int] = None
    name: str = ""
    phone: Optional[str] = None
    email: Optional[str] = None
    created_at: Optional[str] = None
    updated_at: Optional[str] = None

    @classmethod
    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

    def to_dict(self):
        return {k: v for k, v in asdict(self).items() if v is not None}

    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

class Service:
    id: Optional[int] = None
    name: str = ""
    duration_minutes: int = 60
    description: Optional[str] = None

    @classmethod
    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

    def to_dict(self):
        return {k: v for k, v in asdict(self).items() if v is not None}

    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

class Availability:
    id: Optional[int] = None
    date: str = ""
    time: str = ""
    service_id: Optional[int] = None
    is_booked: bool = False

    @classmethod
    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

    def to_dict(self):
        return asdict(self)

    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

class Booking:
    id: Optional[int] = None
    booking_ref: str = ""
    customer_id: Optional[int] = None
    date: str = ""
    time: str = ""
    service_id: Optional[int] = None
    status: str = "confirmed"
    notes: Optional[str] = None
    created_at: Optional[str] = None

    @classmethod
    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

    def to_dict(self):
        return {k: v for k, v in asdict(self).items() if v is not None}

    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

class Ticket:
    id: Optional[int] = None
    ticket_ref: str = ""
    customer_id: Optional[int] = None
    issue_summary: str = ""
    issue_details: Optional[str] = None
    priority: str = "medium"
    category: Optional[str] = None
    status: str = "open"
    created_at: Optional[str] = None

    @classmethod
    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

    def to_dict(self):
        return {k: v for k, v in asdict(self).items() if v is not None}

    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

class CallLog:
    id: Optional[int] = None
    call_id: str = ""
    started_at: Optional[str] = None
    ended_at: Optional[str] = None
    duration_seconds: Optional[int] = None
    transcript: Optional[str] = None
    actions_taken: Optional[str] = None
    outcome: Optional[str] = None

    @classmethod
    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

    def to_dict(self):
        return {k: v for k, v in asdict(self).items() if v is not None}

    def from_row(cls, row):
        if row is None:
            return None
        return cls(**{k: row[k] for k in row.keys()})

async def generate_booking_ref(db) -> str:
    count = await db.fetch_val("SELECT COUNT(*) FROM bookings")
    return f"BK-{(count or 0) + 1:04d}"

async def generate_ticket_ref(db) -> str:
    count = await db.fetch_val("SELECT COUNT(*) FROM tickets")
    return f"TK-{(count or 0) + 1:04d}"

async def seed_db():
    db = await get_db()
    try:
        # Check if already seeded
        count = await db.fetch_val("SELECT COUNT(*) FROM services")
        if count == 0:
            # Seed services
            for svc in SERVICES:
                await db.execute(
                    "INSERT INTO services (name, duration_minutes, description) VALUES (?, ?, ?)",
                    (svc["name"], svc["duration_minutes"], svc["description"]),
                )

            # Seed customers
            for cust in SAMPLE_CUSTOMERS:
                await db.execute(
                    "INSERT INTO customers (name, phone, email) VALUES (?, ?, ?)",
                    (cust["name"], cust["phone"], cust["email"]),
                )

            # Seed availability for next 7 days
            random.seed(42)  # Reproducible for testing
            today = date.today()
            slots_created = 0
            for day_offset in range(7):
                d = today + timedelta(days=day_o

async def _seed_default_flow(db):
    """Create the default 'Customer Service' flow if no flows exist."""
    count = await db.fetch_val("SELECT COUNT(*) FROM flows")
    if count > 0:
        return

    # ---- Drawflow visual format ----
    def _node(nid, ntype, label, data, inputs, outputs, px, py):
        """Build a single Drawflow node dict."""
        type_labels = {
            'entry': 'Entry', 'message': 'Message', 'question': 'Question',
            'tool_call': 'Tool Call', 'condition': 'Condition',
            'transfer': 'Transfer', 'end': 'End',
        }
        html = (
            f'<div class="df-node type-{ntype}">'
            f'<div class="df-node-header"><span class="node-icon"></span>'
            f'<span class="node-type-label">{type_labels[ntype]}</span></div>'
            f'<div class="df-node-body">{label}</div></div>'
        )
        inp = {}
        for k, conns in inputs.items():
            inp[k] = {"connections": [{"node": str(n), "input": k.replace

async function fetchJSON(url) {
    const res = await apiFetch(url);
    return res.json();
}

function loadUser() {
    const user = getUserInfo();
    if (user) {
        const name = user.name || user.email || 'User';
        document.getElementById('userName').textContent = name;
        document.getElementById('userAvatar').textContent = name.charAt(0).toUpperCase();
        document.getElementById('greetingText').textContent = 'Welcome back, ' + name.split(' ')[0];
    }
}

function showTab(name, btn) {
    document.querySelectorAll('.tab-panel').forEach(el => el.classList.remove('active'));
    document.querySelectorAll('.dash-tab').forEach(el => el.classList.remove('active'));
    document.getElementById('tab-' + name).classList.add('active');
    if (btn) btn.classList.add('active');
}

async function loadAgents() {
    const grid = document.getElementById('agentsGrid');
    try {
        const flows = await fetchJSON('/api/flows');
        let html = '';
        if (flows.length === 0) {
            html += `<a href="onboard.html" class="agent-card agent-card-new">
                <span class="plus">+</span>
                <span>Create your first agent</span>
            </a>`;
        } else {
            for (const flow of flows) {
                const isActive = flow.is_active;
                html += `<a href="setup.html" class="agent-card">
                    <div class="agent-status">
                        <span class="agent-dot${isActive ? ' live' : ''}"></span>
                        ${isActive ? 'Active' : 'Inactive'}
                    </div>
                    <div class="agent-name">${escapeHtml(flow.name)}</div>
                    <div class="agent-meta">Flow #${flow.id}</div>
                </a>`;
            }
            html += `<a href="on

function showPhoneModal() {
    document.getElementById('phoneModal').classList.add('visible');
}

function hidePhoneModal() {
    document.getElementById('phoneModal').classList.remove('visible');
}

async function loadStats() {
    try {
        const stats = await fetchJSON('/api/stats');
        document.getElementById('statCalls').textContent = stats.total_calls;
        document.getElementById('statBookings').textContent = stats.total_bookings;
        document.getElementById('statTickets').textContent = stats.total_tickets;
        document.getElementById('statCustomers').textContent = stats.total_customers;
    } catch (e) { /* ignore */ }
}

async function loadCalls() {
    try {
        const calls = await fetchJSON('/api/calls');
        const tbody = document.getElementById('callsTable');
        if (calls.length === 0) {
            tbody.innerHTML = '<tr><td colspan="5" style="text-align:center;color:var(--text-muted)">No calls yet</td></tr>';
            return;
        }
        tbody.innerHTML = calls.map(c => {
            const started = c.started_at ? new Date(c.started_at).toLocaleString() : '-';
            const duration = c.duration_seconds != null ? c.duration_seconds + 's' : '-';
            const badge = c.outcome ? `<span class="badge badge-${c.outcome === 'booking_made' ? 'confirmed' : 'open'}">${c.outcome}</span>` : '-';
            const transcriptBtn = c.transcript && c.transcript.length > 0
                ? `<button style="padding:4px 8px;font-size:0.75rem;background:var(--bg-card);color:var(--text-secondary);border:1px solid var(--border);border-radius:6px;cursor:pointer;font-family:inherit" oncli

function toggleTranscript(btn, transcriptJSON) {
    const existing = btn.parentElement.querySelector('.transcript-viewer');
    if (existing) {
        existing.remove();
        btn.textContent = 'View';
        return;
    }
    const transcript = JSON.parse(transcriptJSON);
    const viewer = document.createElement('div');
    viewer.className = 'transcript-viewer';
    viewer.innerHTML = transcript.map(t =>
        `<div class="t-${t.role}"><strong>${t.role === 'user' ? 'User' : 'Agent'}:</strong> ${escapeHtml(t.text)}</div>`
    ).join('');
    btn.parentElement.appendChild(viewer);
    btn.textContent = 'Hide';
}