call-center/services/asterisk_bridge_service/bridge_processing.py

from __future__ import annotations

from datetime import datetime, timezone
import hashlib
import json
import logging
import os
from pathlib import Path
import threading
from typing import Any

import httpx
from sqlalchemy import select, update

from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.reporting_facts import upsert_reporting_interaction_fact
from services.shared.sql_models import (
    AsteriskCallActionLogRow,
    AsteriskCallLinkRow,
    AsteriskEventLogRow,
    EscalationRow,
    Interaction,
    VoiceAISessionRow,
)


logger = logging.getLogger("uvicorn.error")


def _bridge_app():
    import services.asterisk_bridge_service.app as bridge_app

    return bridge_app


def _parse_iso_timestamp(raw: str | None) -> datetime | None:
    if not raw:
        return None
    try:
        value = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def _seconds_between(start_ts: str | None, end_ts: str | None) -> int | None:
    start = _parse_iso_timestamp(start_ts)
    end = _parse_iso_timestamp(end_ts)
    if not start or not end:
        return None
    return max(int((end - start).total_seconds()), 0)


def _upsert_voice_reporting_fact(
    session,
    link: AsteriskCallLinkRow,
    *,
    status: str,
    answered: bool | None = None,
    abandoned: bool | None = None,
    wait_seconds: int | None = None,
    handle_seconds: int | None = None,
    closed_at: str | None = None,
) -> None:
    interaction_id = str(link.interaction_id or "").strip()
    if not interaction_id:
        return
    payload: dict[str, Any] = {
        "interaction_id": interaction_id,
        "channel": "voice",
        "queue_id": link.queue_id,
        "agent_id": link.claimed_by_user,
        "status": status,
        "created_at": link.started_at,
        "source": "asterisk-bridge",
    }
    if answered is not None:
        payload["answered"] = answered
    if abandoned is not None:
        payload["abandoned"] = abandoned
    if wait_seconds is not None:
        payload["wait_seconds"] = wait_seconds
    if handle_seconds is not None:
        payload["handle_seconds"] = handle_seconds
    if closed_at is not None:
        payload["closed_at"] = closed_at
    upsert_reporting_interaction_fact(session, payload)


def create_bridge_log(
    session,
    *,
    ami_event_name: str,
    call_id: str,
    linked_id: str | None,
    payload: dict[str, Any],
) -> AsteriskEventLogRow:
    now = utc_now_iso()
    row = AsteriskEventLogRow(
        bridge_event_id=new_id("abe"),
        ami_event_name=ami_event_name,
        call_id=call_id,
        linked_id=linked_id,
        interaction_id=None,
        recording_id=None,
        forward_status="received",
        payload_json=json.dumps(payload, ensure_ascii=False),
        last_error=None,
        created_at=now,
        updated_at=now,
    )
    session.add(row)
    session.flush()
    return row


def mark_log_forwarded(
    session,
    row: AsteriskEventLogRow,
    *,
    interaction_id: str | None = None,
    recording_id: str | None = None,
) -> None:
    row.interaction_id = interaction_id or row.interaction_id
    row.recording_id = recording_id or row.recording_id
    row.forward_status = "forwarded"
    row.last_error = None
    row.updated_at = utc_now_iso()
    session.flush()


def mark_log_failed(session, row: AsteriskEventLogRow, message: str) -> None:
    row.forward_status = "failed"
    row.last_error = message[:4000]
    row.updated_at = utc_now_iso()
    session.flush()


def stable_source_event_id(*parts: str) -> str:
    raw = ":".join(str(part).strip() for part in parts if str(part).strip())
    if len(raw) <= 64:
        return raw
    digest = hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]
    return f"{raw[:47]}:{digest}"


def bridge_source_event_id(row: AsteriskEventLogRow, effect: str) -> str:
    return stable_source_event_id(row.bridge_event_id, effect)


def reconcile_source_event_id(call_id: str, effect: str) -> str:
    return stable_source_event_id("reconcile", call_id, effect)


def load_bridge_event_row(session, bridge_event_id: str) -> AsteriskEventLogRow | None:
    return session.execute(
        select(AsteriskEventLogRow).where(AsteriskEventLogRow.bridge_event_id == bridge_event_id)
    ).scalar_one_or_none()


def claim_bridge_event_for_processing(
    bridge_event_id: str,
    *,
    allowed_statuses: tuple[str, ...] = ("received", "failed"),
) -> bool:
    if not allowed_statuses:
        return False

    session = get_session()
    try:
        result = session.execute(
            update(AsteriskEventLogRow)
            .where(AsteriskEventLogRow.bridge_event_id == bridge_event_id)
            .where(AsteriskEventLogRow.forward_status.in_(allowed_statuses))
            .values(
                forward_status="processing",
                last_error=None,
                updated_at=utc_now_iso(),
            )
        )
        session.commit()
        return int(result.rowcount or 0) > 0
    finally:
        session.close()


def process_claimed_bridge_event(bridge_event_id: str) -> AsteriskEventLogRow | None:
    bridge = _bridge_app()
    session = get_session()
    try:
        row = bridge._load_bridge_event_row(session, bridge_event_id)
        if row is None:
            return None
        if row.forward_status != "processing":
            return row
        try:
            bridge._process_bridge_row(session, row)
            session.commit()
            logger.warning(
                "bridge.processed event=%s call_id=%s final_status=%s",
                row.ami_event_name,
                row.call_id,
                row.forward_status,
            )
        except Exception as exc:
            session.rollback()
            row = bridge._load_bridge_event_row(session, bridge_event_id)
            if row is None:
                return None
            bridge._mark_log_failed(session, row, str(exc))
            session.commit()
            logger.warning(
                "bridge.process_failed event=%s call_id=%s error=%s",
                row.ami_event_name,
                row.call_id,
                str(exc)[:500],
            )
        session.refresh(row)
        return row
    finally:
        session.close()


def create_action_log(
    session,
    *,
    call_id: str,
    interaction_id: str | None,
    action_type: str,
    actor_user: str,
    actor_role: str,
    request_payload: dict[str, Any],
) -> AsteriskCallActionLogRow:
    row = AsteriskCallActionLogRow(
        action_id=new_id("aca"),
        call_id=call_id,
        interaction_id=interaction_id,
        action_type=action_type,
        actor_user=actor_user,
        actor_role=actor_role,
        request_json=json.dumps(request_payload, ensure_ascii=False),
        result_status="ok",
        ami_action_id=None,
        error=None,
        created_at=utc_now_iso(),
    )
    session.add(row)
    return row


def set_action_result(
    session,
    row: AsteriskCallActionLogRow,
    *,
    result_status: str,
    ami_action_id: str | None = None,
    error: str | None = None,
) -> None:
    row.result_status = result_status
    row.ami_action_id = ami_action_id
    row.error = (error or "")[:4000] or None


def find_call_link(session, call_id: str) -> AsteriskCallLinkRow | None:
    return session.execute(
        select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
    ).scalar_one_or_none()


def find_voice_ai_session_by_call_id(session, call_id: str) -> VoiceAISessionRow | None:
    return session.execute(
        select(VoiceAISessionRow)
        .where(VoiceAISessionRow.call_id == call_id)
        .order_by(VoiceAISessionRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()


def find_interaction_in_db_by_call_id(session, call_id: str) -> Interaction | None:
    marker = f"[{call_id}]"
    return session.execute(
        select(Interaction)
        .where(Interaction.subject.like(f"%{marker}%"))
        .order_by(Interaction.id.desc())
        .limit(1)
    ).scalars().first()


def recover_call_link_from_started_event(
    session,
    *,
    call_id: str,
    exclude_bridge_event_id: str,
) -> AsteriskCallLinkRow | None:
    bridge = _bridge_app()
    existing = bridge._find_call_link(session, call_id)
    if existing:
        return existing

    started_events = session.execute(
        select(AsteriskEventLogRow)
        .where(AsteriskEventLogRow.ami_event_name == f"{bridge._ami_prefix()}CallStarted")
        .where(AsteriskEventLogRow.call_id == call_id)
        .order_by(AsteriskEventLogRow.id.desc())
    ).scalars().all()

    for started in started_events:
        if started.bridge_event_id == exclude_bridge_event_id:
            continue
        if started.forward_status == "forwarded":
            return bridge._find_call_link(session, call_id)
        try:
            bridge._process_bridge_row(session, started)
        except Exception as exc:
            bridge._mark_log_failed(session, started, str(exc))
        recovered = bridge._find_call_link(session, call_id)
        if recovered:
            return recovered
    return None


def find_forwarded_log(
    session,
    *,
    event_name: str,
    call_id: str,
    exclude_bridge_event_id: str,
) -> AsteriskEventLogRow | None:
    rows = session.execute(
        select(AsteriskEventLogRow)
        .where(AsteriskEventLogRow.ami_event_name == event_name)
        .where(AsteriskEventLogRow.call_id == call_id)
        .where(AsteriskEventLogRow.forward_status == "forwarded")
        .order_by(AsteriskEventLogRow.id.desc())
    ).scalars().all()
    for row in rows:
        if row.bridge_event_id != exclude_bridge_event_id:
            return row
    return None


def create_interaction(*, caller_number: str | None, queue_id: str, call_id: str) -> dict[str, Any]:
    bridge = _bridge_app()
    subject_value = caller_number.strip() if caller_number else "unknown"
    return bridge._post_json(
        f"{bridge._interaction_service_url()}/interactions",
        {
            "channel": "voice",
            "subject": f"Inbound call {subject_value} [{call_id}]",
            "queue_id": queue_id,
            "priority": 3,
        },
    )


def find_interaction_by_call_id(call_id: str) -> dict[str, Any] | None:
    bridge = _bridge_app()
    with httpx.Client(timeout=bridge._forward_timeout_seconds(45.0)) as client:
        response = bridge._request_with_bridge_auth(
            client,
            method="GET",
            url=f"{bridge._interaction_service_url()}/interactions",
        )
        payload = response.json()
    if not isinstance(payload, list):
        return None
    marker = f"[{call_id}]"
    for item in payload:
        if not isinstance(item, dict):
            continue
        subject = str(item.get("subject") or "")
        if marker in subject:
            return item
    return None


def resolve_started_interaction_id(
    session,
    *,
    call_id: str,
    queue_id: str,
    caller_number: str | None,
) -> tuple[AsteriskCallLinkRow | None, str]:
    bridge = _bridge_app()
    existing_link = bridge._find_call_link(session, call_id)
    interaction_id = None
    if existing_link:
        interaction_id = str(existing_link.interaction_id or "").strip() or None
    if interaction_id:
        return existing_link, interaction_id

    recovered_interaction = bridge._find_interaction_in_db_by_call_id(session, call_id)
    if recovered_interaction and str(recovered_interaction.interaction_id or "").strip():
        return existing_link, str(recovered_interaction.interaction_id).strip()

    create_error: Exception | None = None
    try:
        created = bridge._create_interaction(
            caller_number=caller_number,
            queue_id=queue_id,
            call_id=call_id,
        )
        created_interaction_id = str((created or {}).get("interaction_id") or "").strip()
        if created_interaction_id:
            return existing_link, created_interaction_id
        create_error = RuntimeError("Interaction service did not return interaction_id")
    except Exception as exc:
        create_error = exc

    recovered_interaction = bridge._find_interaction_in_db_by_call_id(session, call_id)
    if recovered_interaction and str(recovered_interaction.interaction_id or "").strip():
        return existing_link, str(recovered_interaction.interaction_id).strip()

    recovered_payload = bridge._find_interaction_by_call_id(call_id)
    recovered_interaction_id = str((recovered_payload or {}).get("interaction_id") or "").strip()
    if recovered_interaction_id:
        return existing_link, recovered_interaction_id

    if create_error is not None:
        raise create_error
    raise RuntimeError("Unable to resolve interaction_id for started call")


def callcontrol_side_effect_timeout_seconds() -> float:
    bridge = _bridge_app()
    return min(bridge._forward_timeout_seconds(45.0), 2.5)


def callcontrol_side_effect_max_attempts() -> int:
    return 1


def emit_voice_event(
    *,
    event_type: str,
    call_id: str,
    interaction_id: str | None,
    source_event_id: str | None = None,
    payload: dict[str, Any],
) -> dict[str, Any]:
    bridge = _bridge_app()
    return bridge._post_json(
        f"{bridge._voice_adapter_service_url()}/integrations/voice/events",
        {
            "event_type": event_type,
            "call_id": call_id,
            "interaction_id": interaction_id,
            "source_event_id": source_event_id,
            "payload": payload,
        },
    )


def assign_interaction(*, interaction_id: str, assignee: str) -> dict[str, Any]:
    bridge = _bridge_app()
    return bridge._patch_json(
        f"{bridge._interaction_service_url()}/interactions/{interaction_id}/assign",
        {"assignee": assignee},
        timeout_seconds=bridge._callcontrol_side_effect_timeout_seconds(),
        max_attempts=bridge._callcontrol_side_effect_max_attempts(),
        retry_backoff_seconds=0.0,
    )


def upload_recording(
    *,
    local_path: Path,
    call_id: str,
    interaction_id: str | None,
    source_event_id: str | None,
    file_name: str,
    mime_type: str | None,
    duration_seconds: int | None,
) -> dict[str, Any]:
    bridge = _bridge_app()
    data = {
        "call_id": call_id,
        "file_name": file_name,
    }
    if interaction_id:
        data["interaction_id"] = interaction_id
    if source_event_id:
        data["source_event_id"] = source_event_id
    if mime_type:
        data["mime_type"] = mime_type
    if duration_seconds is not None:
        data["duration_seconds"] = str(duration_seconds)

    last_exc: Exception | None = None
    with local_path.open("rb") as handle, httpx.Client(timeout=bridge._forward_timeout_seconds(60.0)) as client:
        for attempt in range(1, bridge._forward_max_attempts() + 1):
            try:
                response = bridge._request_with_bridge_auth(
                    client,
                    method="POST",
                    url=f"{bridge._recording_service_url()}/recordings/import-upload",
                    data=data,
                    files={"file": (file_name, handle, mime_type or "application/octet-stream")},
                    retry_reset=lambda: handle.seek(0),
                )
                return response.json()
            except httpx.HTTPStatusError as exc:
                last_exc = exc
                status = exc.response.status_code if exc.response is not None else 0
                if status < 500 or attempt >= bridge._forward_max_attempts():
                    raise
            except httpx.RequestError as exc:
                last_exc = exc
                if attempt >= bridge._forward_max_attempts():
                    raise
            handle.seek(0)
    if last_exc is not None:
        raise last_exc
    raise RuntimeError("Failed to upload recording")


def normalize_duration(value: Any) -> int | None:
    if value in (None, ""):
        return None
    try:
        parsed = int(str(value))
    except ValueError:
        return None
    return max(parsed, 0)


def process_call_started(
    session,
    row: AsteriskEventLogRow,
    payload: dict[str, Any],
) -> None:
    bridge = _bridge_app()
    forwarded = bridge._find_forwarded_log(
        session,
        event_name=row.ami_event_name,
        call_id=row.call_id,
        exclude_bridge_event_id=row.bridge_event_id,
    )
    if forwarded:
        bridge._mark_log_forwarded(session, row, interaction_id=forwarded.interaction_id)
        return

    queue_code = str(payload.get("QueueCode") or "").strip()
    queue_id = bridge._queue_map().get(queue_code)
    if not queue_id:
        raise RuntimeError(f"Unknown QueueCode: {queue_code or '<empty>'}")

    caller_number = str(payload.get("CallerNumber") or "").strip() or None
    existing_link, interaction_id = bridge._resolve_started_interaction_id(
        session,
        call_id=row.call_id,
        queue_id=queue_id,
        caller_number=caller_number,
    )

    bridge._emit_voice_event(
        event_type="call.started",
        call_id=row.call_id,
        interaction_id=interaction_id,
        source_event_id=bridge._bridge_source_event_id(row, "call.started"),
        payload={
            "source": "asterisk",
            "caller_number": payload.get("CallerNumber"),
            "caller_name": payload.get("CallerName"),
            "extension": payload.get("Extension"),
            "context": payload.get("Context"),
            "linked_id": row.linked_id,
            "direction": payload.get("Direction", "inbound"),
        },
    )

    now = utc_now_iso()
    channel_name = bridge._resolve_channel_from_payload(payload)
    link = existing_link or AsteriskCallLinkRow(
        call_id=row.call_id,
        linked_id=row.linked_id,
        queue_code=queue_code,
        queue_id=queue_id,
        interaction_id=interaction_id,
        caller_number=caller_number,
        caller_name=str(payload.get("CallerName") or "").strip() or None,
        status="active",
        telephony_status="ringing",
        claimed_by_user=None,
        claimed_at=None,
        operator_extension=None,
        channel_name=channel_name,
        started_at=now,
        connected_at=None,
        ended_at=None,
        updated_at=now,
    )
    if not existing_link:
        session.add(link)
    else:
        link.linked_id = row.linked_id
        link.queue_code = queue_code
        link.queue_id = queue_id
        link.interaction_id = interaction_id
        link.caller_number = caller_number
        link.caller_name = str(payload.get("CallerName") or "").strip() or None
        link.status = "active"
        link.telephony_status = "ringing"
        if channel_name:
            link.channel_name = channel_name
        link.updated_at = now

    _upsert_voice_reporting_fact(session, link, status="new")

    ai_config = bridge._ai_voice_config_for_queue(queue_code)
    if ai_config:
        link.ai_state = "queued"
        link.ai_handoff_reason = None
        link.ai_last_model_at = now
        logger.warning(
            "bridge.call_started_ai call_id=%s interaction_id=%s queue_code=%s queue_id=%s",
            row.call_id,
            interaction_id,
            queue_code,
            queue_id,
        )
        try:
            started = bridge._start_voice_ai_session(
                call_id=row.call_id,
                linked_id=row.linked_id,
                interaction_id=interaction_id,
                queue_id=queue_id,
                queue_code=queue_code,
                caller_number=caller_number,
                caller_name=str(payload.get("CallerName") or "").strip() or None,
                ai_config=ai_config,
            )
            link.voice_session_id = str((started or {}).get("voice_session_id") or "").strip() or None
            link.ai_session_id = str((started or {}).get("ai_session_id") or "").strip() or None
            link.ai_state = str((started or {}).get("status") or "queued").strip() or "queued"
            link.ai_last_model_at = utc_now_iso()
            logger.warning(
                "bridge.call_started_ai_result call_id=%s voice_session_id=%s ai_session_id=%s ai_state=%s",
                row.call_id,
                link.voice_session_id,
                link.ai_session_id,
                link.ai_state,
            )
            try:
                bridge._append_interaction_timeline(
                    interaction_id=interaction_id,
                    action="ai.session_started",
                    metadata={
                        "call_id": row.call_id,
                        "voice_session_id": link.voice_session_id,
                        "ai_session_id": link.ai_session_id,
                        "queue_code": queue_code,
                    },
                )
            except Exception:
                pass
        except Exception as exc:
            link.ai_state = "error"
            link.ai_handoff_reason = str(exc)[:1000]
            link.ai_last_model_at = utc_now_iso()
            logger.warning(
                "bridge.call_started_ai_error call_id=%s queue_code=%s error=%s",
                row.call_id,
                queue_code,
                str(exc)[:500],
            )
            try:
                bridge._append_interaction_timeline(
                    interaction_id=interaction_id,
                    action="ai.error",
                    metadata={
                        "call_id": row.call_id,
                        "queue_code": queue_code,
                        "error": str(exc)[:500],
                    },
                )
            except Exception:
                pass

    bridge._mark_log_forwarded(session, row, interaction_id=interaction_id)


def process_audio_bridge_requested(
    session,
    row: AsteriskEventLogRow,
    payload: dict[str, Any],
) -> None:
    bridge = _bridge_app()
    forwarded = bridge._find_forwarded_log(
        session,
        event_name=row.ami_event_name,
        call_id=row.call_id,
        exclude_bridge_event_id=row.bridge_event_id,
    )
    if forwarded:
        bridge._mark_log_forwarded(session, row, interaction_id=forwarded.interaction_id)
        return

    link = bridge._find_call_link(session, row.call_id)
    if not link:
        link = bridge._recover_call_link_from_started_event(
            session,
            call_id=row.call_id,
            exclude_bridge_event_id=row.bridge_event_id,
        )
    if not link:
        raise RuntimeError(f"Unknown CallID: {row.call_id}")
    voice_session_id = str(link.voice_session_id or "").strip() or None
    ai_session_id = str(link.ai_session_id or "").strip() or None
    ai_state = str(link.ai_state or "").strip() or None
    if not str(link.voice_session_id or "").strip():
        voice_session = find_voice_ai_session_by_call_id(session, row.call_id)
        if voice_session is not None:
            voice_session_id = str(voice_session.session_id or "").strip() or None
            ai_session_id = str(voice_session.ai_session_id or "").strip() or None
            if ai_state in {None, "", "queued"}:
                ai_state = str(voice_session.status or "greeting").strip() or "greeting"
    if not voice_session_id:
        raise RuntimeError("Voice AI session is not ready for AudioSocket bridge")

    media_uuid = bridge._first_non_empty(payload.get("MediaUUID"), payload.get("MediaUuid"))
    if not media_uuid:
        raise RuntimeError("MediaUUID is required for AudioSocket bridge request")
    channel = bridge._resolve_channel_from_payload(payload)
    service_address = bridge._first_non_empty(
        payload.get("AudioSocketService"),
        payload.get("ServiceAddress"),
    )
    logger.warning(
        "bridge.audio_bridge_requested call_id=%s voice_session_id=%s ai_session_id=%s ai_state=%s media_uuid=%s service=%s channel=%s",
        row.call_id,
        voice_session_id,
        ai_session_id,
        ai_state,
        media_uuid,
        service_address,
        channel,
    )
    bridge._register_voice_ai_media_bridge(
        voice_session_id=voice_session_id,
        event_type="requested",
        media_uuid=media_uuid,
        call_id=row.call_id,
        linked_id=row.linked_id,
        channel=channel,
        service_address=service_address,
    )
    logger.warning(
        "bridge.audio_bridge_registered call_id=%s voice_session_id=%s media_uuid=%s",
        row.call_id,
        voice_session_id,
        media_uuid,
    )
    link.voice_session_id = voice_session_id
    link.ai_session_id = ai_session_id
    if channel:
        link.channel_name = channel
    link.ai_state = "greeting"
    link.ai_last_model_at = utc_now_iso()
    link.updated_at = utc_now_iso()
    bridge._mark_log_forwarded(session, row, interaction_id=link.interaction_id)


def process_audio_bridge_ended(
    session,
    row: AsteriskEventLogRow,
    payload: dict[str, Any],
) -> None:
    bridge = _bridge_app()
    forwarded = bridge._find_forwarded_log(
        session,
        event_name=row.ami_event_name,
        call_id=row.call_id,
        exclude_bridge_event_id=row.bridge_event_id,
    )
    if forwarded:
        bridge._mark_log_forwarded(session, row, interaction_id=forwarded.interaction_id)
        return

    link = bridge._find_call_link(session, row.call_id)
    if not link:
        link = bridge._recover_call_link_from_started_event(
            session,
            call_id=row.call_id,
            exclude_bridge_event_id=row.bridge_event_id,
        )
    if not link:
        raise RuntimeError(f"Unknown CallID: {row.call_id}")
    media_uuid = bridge._first_non_empty(payload.get("MediaUUID"), payload.get("MediaUuid"))
    reason = bridge._first_non_empty(payload.get("Reason"), payload.get("TryExecStatus"))
    logger.warning(
        "bridge.audio_bridge_ended call_id=%s voice_session_id=%s media_uuid=%s reason=%s",
        row.call_id,
        str(link.voice_session_id or "").strip() or None,
        media_uuid,
        reason,
    )
    if str(link.voice_session_id or "").strip() and media_uuid:
        bridge._register_voice_ai_media_bridge(
            voice_session_id=link.voice_session_id,
            event_type="ended",
            media_uuid=media_uuid,
            call_id=row.call_id,
            linked_id=row.linked_id,
            channel=bridge._resolve_channel_from_payload(payload),
            service_address=bridge._first_non_empty(
                payload.get("AudioSocketService"),
                payload.get("ServiceAddress"),
            ),
            reason=reason,
        )
    link.ai_last_model_at = utc_now_iso()
    link.updated_at = utc_now_iso()
    bridge._mark_log_forwarded(session, row, interaction_id=link.interaction_id)


def process_call_ended(
    session,
    row: AsteriskEventLogRow,
    payload: dict[str, Any],
) -> None:
    bridge = _bridge_app()
    forwarded = bridge._find_forwarded_log(
        session,
        event_name=row.ami_event_name,
        call_id=row.call_id,
        exclude_bridge_event_id=row.bridge_event_id,
    )
    if forwarded:
        bridge._mark_log_forwarded(session, row, interaction_id=forwarded.interaction_id)
        return

    link = bridge._find_call_link(session, row.call_id)
    if not link:
        link = bridge._recover_call_link_from_started_event(
            session,
            call_id=row.call_id,
            exclude_bridge_event_id=row.bridge_event_id,
        )
    if not link:
        raise RuntimeError(f"Unknown CallID: {row.call_id}")

    bridge._emit_voice_event(
        event_type="call.ended",
        call_id=row.call_id,
        interaction_id=link.interaction_id,
        source_event_id=bridge._bridge_source_event_id(row, "call.ended"),
        payload={
            "source": "asterisk",
            "hangup_cause": payload.get("HangupCause"),
            "duration_seconds": bridge._normalize_duration(payload.get("DurationSeconds")),
            "linked_id": row.linked_id,
        },
    )
    now = utc_now_iso()
    link.status = "ended"
    link.telephony_status = "ended"
    link.ended_at = now
    link.updated_at = now
    answered = bool(link.connected_at)
    _upsert_voice_reporting_fact(
        session,
        link,
        status="closed" if answered else "abandoned",
        answered=answered,
        abandoned=not answered,
        wait_seconds=_seconds_between(link.started_at, link.connected_at) if answered else None,
        handle_seconds=_seconds_between(link.connected_at, link.ended_at) if answered else None,
        closed_at=link.ended_at,
    )
    if str(link.ai_state or "").strip() and str(link.ai_state or "").strip() != "human_owned":
        link.ai_state = "closed"
        link.ai_last_model_at = now
    logger.warning(
        "bridge.call_ended call_id=%s voice_session_id=%s hangup_cause=%s duration_seconds=%s ai_state=%s",
        row.call_id,
        link.voice_session_id,
        payload.get("HangupCause"),
        payload.get("DurationSeconds"),
        link.ai_state,
    )
    open_escalation = session.execute(
        select(EscalationRow).where(
            EscalationRow.call_id == row.call_id,
            EscalationRow.status.in_(["requested", "ringing", "connected"]),
        )
    ).scalar_one_or_none()
    was_talking = open_escalation is not None and open_escalation.status == "connected"
    if open_escalation is not None:
        open_escalation.status = "completed" if answered else "failed"
        open_escalation.completed_at = now
        if answered and not open_escalation.connected_at:
            open_escalation.connected_at = now
    if was_talking and open_escalation and open_escalation.real_agent_id:
        try:
            bridge._set_routing_agent_status(open_escalation.real_agent_id, "AFTER_CALL_WORK")
        except Exception:
            logger.warning("bridge.agent_acw_status_failed call_id=%s agent_id=%s", row.call_id, open_escalation.real_agent_id)
    else:
        try:
            bridge._release_routing_agent(row.call_id)
        except Exception:
            pass
    if open_escalation is not None:
        try:
            bridge._append_interaction_timeline(
                interaction_id=link.interaction_id,
                action="escalation.completed" if answered else "escalation.failed_client_disconnected",
                metadata={"call_id": row.call_id, "escalation_id": open_escalation.escalation_id, "agent_id": open_escalation.real_agent_id},
            )
        except Exception:
            pass
    try:
        bridge._notify_voice_ai_telephony_event(
            voice_session_id=link.voice_session_id,
            event_type="call.ended",
            payload={
                "call_id": row.call_id,
                "hangup_cause": payload.get("HangupCause"),
                "duration_seconds": payload.get("DurationSeconds"),
            },
        )
    except Exception:
        pass
    bridge._mark_log_forwarded(session, row, interaction_id=link.interaction_id)


def process_operator_connected(
    session,
    row: AsteriskEventLogRow,
    payload: dict[str, Any],
) -> None:
    bridge = _bridge_app()
    forwarded = bridge._find_forwarded_log(
        session,
        event_name=row.ami_event_name,
        call_id=row.call_id,
        exclude_bridge_event_id=row.bridge_event_id,
    )
    if forwarded:
        bridge._mark_log_forwarded(session, row, interaction_id=forwarded.interaction_id)
        return

    link = bridge._find_call_link(session, row.call_id)
    if not link:
        link = bridge._recover_call_link_from_started_event(
            session,
            call_id=row.call_id,
            exclude_bridge_event_id=row.bridge_event_id,
        )
    if not link:
        raise RuntimeError(f"Unknown CallID: {row.call_id}")

    operator_extension = bridge._first_non_empty(
        payload.get("OperatorExtension"),
        payload.get("OperatorExten"),
        payload.get("Exten"),
        payload.get("Extension"),
    )
    claimed_by = bridge._first_non_empty(
        payload.get("ClaimedBy"),
        payload.get("ClaimedByUser"),
        payload.get("OperatorUser"),
    )
    channel = bridge._resolve_channel_from_payload(payload)
    if operator_extension in {"", "s"}:
        operator_extension = bridge._extract_extension_from_channel(channel)
    if claimed_by in {"", "s"}:
        claimed_by = None
    if not claimed_by and str(link.claimed_by_user or "").strip():
        claimed_by = str(link.claimed_by_user or "").strip()
    if not claimed_by and operator_extension:
        claimed_by = bridge._user_for_operator_extension(operator_extension)
    now = utc_now_iso()

    if operator_extension:
        link.operator_extension = operator_extension
    if claimed_by:
        link.claimed_by_user = claimed_by
        link.claimed_at = link.claimed_at or now
    if channel:
        link.channel_name = channel
    link.telephony_status = "connected"
    link.connected_at = now
    link.updated_at = now

    open_escalation = session.execute(
        select(EscalationRow).where(
            EscalationRow.call_id == row.call_id,
            EscalationRow.status == "ringing",
        )
    ).scalar_one_or_none()
    if open_escalation is not None:
        open_escalation.status = "connected"
        open_escalation.connected_at = now
        if open_escalation.real_agent_id:
            try:
                bridge._set_routing_agent_status(open_escalation.real_agent_id, "TALKING")
            except Exception:
                logger.warning("bridge.agent_talking_status_failed call_id=%s agent_id=%s", row.call_id, open_escalation.real_agent_id)
        try:
            bridge._append_interaction_timeline(
                interaction_id=link.interaction_id,
                action="escalation.agent_connected",
                metadata={"call_id": row.call_id, "agent_id": open_escalation.real_agent_id, "escalation_id": open_escalation.escalation_id},
            )
        except Exception:
            pass
        try:
            bridge._emit_voice_event(
                event_type="AgentConnected",
                call_id=row.call_id,
                interaction_id=link.interaction_id,
                payload={"escalation_id": open_escalation.escalation_id, "agent_id": open_escalation.real_agent_id},
            )
        except Exception:
            pass
        try:
            bridge._emit_voice_event(
                event_type="TransferCompleted",
                call_id=row.call_id,
                interaction_id=link.interaction_id,
                payload={"escalation_id": open_escalation.escalation_id, "agent_id": open_escalation.real_agent_id},
            )
        except Exception:
            pass

    _upsert_voice_reporting_fact(
        session,
        link,
        status="in_progress",
        answered=True,
        abandoned=False,
        wait_seconds=_seconds_between(link.started_at, link.connected_at),
    )
    ai_event_type = "call.connected"
    if str(link.ai_state or "").strip() in {"handoff_requested", "handoff_required", "active", "speaking", "thinking", "listening", "greeting"}:
        link.ai_state = "human_owned"
        link.ai_last_model_at = now
        ai_event_type = "operator.connected"
        try:
            bridge._append_interaction_timeline(
                interaction_id=link.interaction_id,
                action="ai.handoff_completed",
                metadata={
                    "call_id": row.call_id,
                    "voice_session_id": link.voice_session_id,
                    "ai_session_id": link.ai_session_id,
                    "operator_extension": operator_extension,
                    "claimed_by_user": claimed_by,
                },
            )
        except Exception:
            pass

    bridge._emit_voice_event(
        event_type="call.connected",
        call_id=row.call_id,
        interaction_id=link.interaction_id,
        source_event_id=bridge._bridge_source_event_id(row, "call.connected"),
        payload={
            "source": "asterisk",
            "linked_id": row.linked_id,
            "operator_extension": operator_extension,
            "claimed_by_user": claimed_by,
            "channel_name": channel,
        },
    )
    try:
        bridge._notify_voice_ai_telephony_event(
            voice_session_id=link.voice_session_id,
            event_type=ai_event_type,
            payload={
                "call_id": row.call_id,
                "operator_extension": operator_extension,
                "claimed_by_user": claimed_by,
                "channel_name": channel,
            },
        )
    except Exception:
        pass
    logger.warning(
        "bridge.operator_connected call_id=%s voice_session_id=%s operator_extension=%s claimed_by=%s ai_event_type=%s",
        row.call_id,
        link.voice_session_id,
        operator_extension,
        claimed_by,
        ai_event_type,
    )
    bridge._mark_log_forwarded(session, row, interaction_id=link.interaction_id)


def process_recording_ready(
    session,
    row: AsteriskEventLogRow,
    payload: dict[str, Any],
) -> None:
    bridge = _bridge_app()
    forwarded = bridge._find_forwarded_log(
        session,
        event_name=row.ami_event_name,
        call_id=row.call_id,
        exclude_bridge_event_id=row.bridge_event_id,
    )
    if forwarded:
        bridge._mark_log_forwarded(
            session,
            row,
            interaction_id=forwarded.interaction_id,
            recording_id=forwarded.recording_id,
        )
        return

    link = bridge._find_call_link(session, row.call_id)
    if not link:
        link = bridge._recover_call_link_from_started_event(
            session,
            call_id=row.call_id,
            exclude_bridge_event_id=row.bridge_event_id,
        )
    if not link:
        raise RuntimeError(f"Unknown CallID: {row.call_id}")

    remote_path = str(payload.get("RemotePath") or "").strip()
    file_name = str(payload.get("FileName") or "").strip() or Path(remote_path).name or "call.wav"
    mime_type = str(payload.get("MimeType") or "").strip() or None
    duration_seconds = bridge._normalize_duration(payload.get("DurationSeconds"))
    if not remote_path:
        raise RuntimeError("RemotePath is required")

    ended_exists = bridge._has_voice_event(session, call_id=row.call_id, event_type="call.ended")
    if not ended_exists:
        bridge._emit_voice_event(
            event_type="call.ended",
            call_id=row.call_id,
            interaction_id=link.interaction_id,
            source_event_id=bridge._bridge_source_event_id(row, "call.ended"),
            payload={
                "source": "asterisk",
                "hangup_cause": "unknown",
                "duration_seconds": duration_seconds,
                "linked_id": row.linked_id,
                "reconciled": True,
                "reason": "recording.ready_without_call.ended",
            },
        )
        ended_exists = True

    if ended_exists and link.status != "ended":
        now = utc_now_iso()
        link.status = "ended"
        link.telephony_status = "ended"
        link.ended_at = link.ended_at or now
        link.updated_at = now
        session.flush()

    existing_record = bridge._latest_recording_for_call(session, call_id=row.call_id)
    if existing_record is not None:
        if not bridge._has_voice_event(session, call_id=row.call_id, event_type="recording.ready"):
            bridge._emit_voice_event(
                event_type="recording.ready",
                call_id=row.call_id,
                interaction_id=link.interaction_id,
                source_event_id=bridge._bridge_source_event_id(row, "recording.ready"),
                payload={
                    "source": "asterisk",
                    "remote_path": remote_path,
                    "file_name": file_name,
                    "mime_type": mime_type,
                    "duration_seconds": duration_seconds,
                    "linked_id": row.linked_id,
                },
            )
        bridge._mark_log_forwarded(
            session,
            row,
            interaction_id=link.interaction_id,
            recording_id=existing_record.recording_id,
        )
        return

    bridge._emit_voice_event(
        event_type="recording.ready",
        call_id=row.call_id,
        interaction_id=link.interaction_id,
        source_event_id=bridge._bridge_source_event_id(row, "recording.ready"),
        payload={
            "source": "asterisk",
            "remote_path": remote_path,
            "file_name": file_name,
            "mime_type": mime_type,
            "duration_seconds": duration_seconds,
            "linked_id": row.linked_id,
        },
    )
    try:
        bridge._notify_voice_ai_telephony_event(
            voice_session_id=link.voice_session_id,
            event_type="recording.ready",
            payload={
                "call_id": row.call_id,
                "file_name": file_name,
                "mime_type": mime_type,
                "duration_seconds": duration_seconds,
            },
        )
    except Exception:
        pass

    local_path, cleanup_required = bridge._fetch_recording_file(remote_path, file_name)
    try:
        created = bridge._upload_recording(
            local_path=local_path,
            call_id=row.call_id,
            interaction_id=link.interaction_id,
            source_event_id=bridge._bridge_source_event_id(row, "recording.import"),
            file_name=file_name,
            mime_type=mime_type,
            duration_seconds=duration_seconds,
        )
        bridge._mark_log_forwarded(
            session,
            row,
            interaction_id=link.interaction_id,
            recording_id=created["recording_id"],
        )
    finally:
        if cleanup_required and local_path.exists():
            local_path.unlink(missing_ok=True)


_NO_ANSWER_DIAL_STATUSES = {"NOANSWER", "BUSY", "CANCEL", "CHANUNAVAIL", "CONGESTION"}
_NO_ANSWER_HANGUP_CAUSES = {"17", "18", "19", "21", "34", "38"}


def process_agent_dial_outcome(session, row: AsteriskEventLogRow, payload: dict[str, Any]) -> None:
    bridge = _bridge_app()
    call_id = row.call_id
    if not call_id:
        bridge._mark_log_forwarded(session, row, interaction_id=None)
        return

    dial_status = str(payload.get("DialStatus") or "").strip().upper()
    hangup_cause = str(payload.get("Cause") or "").strip()

    is_no_answer_outcome = (
        (row.ami_event_name == "DialEnd" and dial_status in _NO_ANSWER_DIAL_STATUSES)
        or (row.ami_event_name == "Hangup" and hangup_cause in _NO_ANSWER_HANGUP_CAUSES)
    )
    if not is_no_answer_outcome:
        bridge._mark_log_forwarded(session, row, interaction_id=None)
        return

    escalation = session.execute(
        select(EscalationRow).where(
            EscalationRow.call_id == call_id,
            EscalationRow.status == "ringing",
        )
    ).scalar_one_or_none()
    if escalation is None:
        bridge._mark_log_forwarded(session, row, interaction_id=None)
        return

    dialed_channel = payload.get("DestChannel") if row.ami_event_name == "DialEnd" else payload.get("Channel")
    dialed_extension = bridge._extract_extension_from_channel(dialed_channel)
    if dialed_extension and escalation.assigned_agent_id and dialed_extension != escalation.assigned_agent_id:
        bridge._mark_log_forwarded(session, row, interaction_id=None)
        return

    try:
        bridge._retry_escalation_no_answer(
            session,
            call_id=call_id,
            dial_outcome=dial_status or f"hangup_cause_{hangup_cause}",
        )
    except Exception:
        logger.exception("bridge.agent_dial_outcome_retry_failed call_id=%s", call_id)
    bridge._mark_log_forwarded(session, row, interaction_id=None)


def process_bridge_row(session, row: AsteriskEventLogRow) -> AsteriskEventLogRow:
    bridge = _bridge_app()
    payload = json.loads(row.payload_json or "{}")
    logger.warning(
        "bridge.process_event event=%s call_id=%s status=%s",
        row.ami_event_name,
        row.call_id,
        row.forward_status,
    )
    if row.ami_event_name == f"{bridge._ami_prefix()}CallStarted":
        bridge._process_call_started(session, row, payload)
    elif row.ami_event_name == f"{bridge._ami_prefix()}AIAudioBridgeRequested":
        bridge._process_audio_bridge_requested(session, row, payload)
    elif row.ami_event_name == f"{bridge._ami_prefix()}AIAudioBridgeEnded":
        bridge._process_audio_bridge_ended(session, row, payload)
    elif row.ami_event_name in {
        f"{bridge._ami_prefix()}AIAudioSocketAttempt",
        f"{bridge._ami_prefix()}AIAudioSocketResult",
    }:
        logger.warning(
            "bridge.audio_socket_diag event=%s call_id=%s payload=%s",
            row.ami_event_name,
            row.call_id,
            row.payload_json,
        )
        interaction_id = None
        link = bridge._find_call_link(session, row.call_id)
        if link is not None:
            interaction_id = link.interaction_id
        bridge._mark_log_forwarded(session, row, interaction_id=interaction_id)
    elif row.ami_event_name == f"{bridge._ami_prefix()}OperatorConnected":
        bridge._process_operator_connected(session, row, payload)
    elif row.ami_event_name == f"{bridge._ami_prefix()}CallEnded":
        bridge._process_call_ended(session, row, payload)
    elif row.ami_event_name == f"{bridge._ami_prefix()}RecordingReady":
        bridge._process_recording_ready(session, row, payload)
    elif row.ami_event_name in {"DialEnd", "Hangup"}:
        bridge._process_agent_dial_outcome(session, row, payload)
    else:
        row.forward_status = "received"
        row.updated_at = utc_now_iso()
        session.flush()
    return row


def record_ami_payload(payload: dict[str, Any], *, event_name: str | None = None) -> None:
    bridge = _bridge_app()
    event_name = str(event_name or payload.get("UserEvent") or "").strip()
    call_id = bridge._extract_call_id(payload)
    linked_id = bridge._extract_linked_id(payload, call_id)
    if not event_name or not call_id:
        return

    session = get_session()
    try:
        row = bridge._create_bridge_log(
            session,
            ami_event_name=event_name,
            call_id=call_id,
            linked_id=linked_id,
            payload=payload,
        )
        session.commit()
        bridge_event_id = row.bridge_event_id
    finally:
        session.close()

    if not bridge._claim_bridge_event_for_processing(bridge_event_id, allowed_statuses=("received",)):
        return
    bridge._process_claimed_bridge_event(bridge_event_id)


def retry_failed_events_once() -> None:
    bridge = _bridge_app()
    session = get_session()
    try:
        rows = session.execute(
            select(AsteriskEventLogRow)
            .where(AsteriskEventLogRow.forward_status.in_(("failed", "received")))
            .order_by(AsteriskEventLogRow.id.asc())
        ).scalars().all()
        event_ids = [row.bridge_event_id for row in rows[: bridge._failed_retry_batch_size()]]
    finally:
        session.close()

    for bridge_event_id in event_ids:
        if not bridge._claim_bridge_event_for_processing(bridge_event_id):
            continue
        bridge._process_claimed_bridge_event(bridge_event_id)


def acw_duration_seconds() -> int:
    raw = str(os.environ.get("ACW_DURATION_SECONDS", "30")).strip()
    try:
        return max(int(raw), 1)
    except ValueError:
        return 30


def acw_sweep_interval_seconds() -> float:
    return max(min(float(acw_duration_seconds()) / 2, 15.0), 5.0)


def acw_sweep_once() -> None:
    from services.routing_service import engine as routing_engine

    session = get_session()
    try:
        swept = routing_engine.sweep_after_call_work(session, older_than_seconds=acw_duration_seconds())
        for agent in swept:
            logger.warning("bridge.agent_acw_swept agent_id=%s", agent.agent_id)
    finally:
        session.close()


def acw_sweep_loop(stop_event: threading.Event | None = None) -> None:
    bridge = _bridge_app()
    active_stop_event = stop_event or bridge._background_stop_event()
    while not active_stop_event.is_set():
        try:
            acw_sweep_once()
        except Exception:
            logger.exception("bridge.acw_sweep_failed")
        if active_stop_event.wait(acw_sweep_interval_seconds()):
            break


def failed_retry_loop(stop_event: threading.Event | None = None) -> None:
    bridge = _bridge_app()
    active_stop_event = stop_event or bridge._background_stop_event()
    while not active_stop_event.is_set():
        if bridge._bridge_enabled() and bridge._failed_retry_enabled():
            try:
                bridge._retry_failed_events_once()
            except Exception:
                pass
        if bridge._bridge_enabled() and bridge._reconcile_enabled():
            try:
                bridge._reconcile_stale_calls_once()
            except Exception:
                pass
        if active_stop_event.wait(bridge._failed_retry_interval_seconds()):
            break