from __future__ import annotations from datetime import datetime, timezone from sqlalchemy import select from services.shared.db import get_session from services.shared.sql_models import ( AsteriskCallLinkRow, CallRecordingRow, VoiceAISessionRow, VoiceEventRow, ) def _bridge_app(): import services.asterisk_bridge_service.app as bridge_app return bridge_app def parse_iso(value: str | None) -> datetime | None: text = str(value or "").strip() if not text: return None try: parsed = datetime.fromisoformat(text) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def has_voice_event(session, *, call_id: str, event_type: str) -> bool: row = session.execute( select(VoiceEventRow.id) .where(VoiceEventRow.call_id == call_id) .where(VoiceEventRow.event_type == event_type) .limit(1) ).scalar_one_or_none() return row is not None def has_recording(session, *, call_id: str) -> bool: row = session.execute( select(CallRecordingRow.id).where(CallRecordingRow.call_id == call_id).limit(1) ).scalar_one_or_none() return row is not None def latest_recording_for_call(session, *, call_id: str) -> CallRecordingRow | None: return session.execute( select(CallRecordingRow) .where(CallRecordingRow.call_id == call_id) .order_by(CallRecordingRow.id.desc()) .limit(1) ).scalar_one_or_none() def _voice_session_recently_active(session, *, link: AsteriskCallLinkRow, now: datetime) -> bool: bridge = _bridge_app() non_terminal_states = { "queued", "greeting", "active", "thinking", "speaking", "listening", "handoff_requested", "handoff_required", } link_ai_state = str(link.ai_state or "").strip() voice_session_id = str(link.voice_session_id or "").strip() if not voice_session_id and link_ai_state not in non_terminal_states: return False voice_session = session.execute( select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == voice_session_id) ).scalar_one_or_none() if voice_session is None: return link_ai_state in non_terminal_states effective_ai_state = link_ai_state or str(voice_session.status or "").strip() if effective_ai_state not in non_terminal_states: return False recent_points = [ bridge._parse_iso(voice_session.last_media_frame_at), bridge._parse_iso(voice_session.last_user_utterance_at), bridge._parse_iso(voice_session.last_ai_reply_at), bridge._parse_iso(voice_session.updated_at), ] recent_points = [point for point in recent_points if point is not None] if not recent_points: return False latest_point = max(recent_points) grace_seconds = max(bridge._reconcile_stale_seconds(), 12) + 8 return max((now - latest_point).total_seconds(), 0) <= grace_seconds def reconcile_stale_calls_once() -> None: bridge = _bridge_app() now = datetime.now(timezone.utc) session = get_session() try: unresolved_recording = ( select(CallRecordingRow.id) .where(CallRecordingRow.call_id == AsteriskCallLinkRow.call_id) .correlate(AsteriskCallLinkRow) .exists() ) active_ids = session.execute( select(AsteriskCallLinkRow.id) .where(AsteriskCallLinkRow.status == "active") .order_by(AsteriskCallLinkRow.id.asc()) .limit(bridge._reconcile_scan_limit()) ).scalars().all() ended_ids = session.execute( select(AsteriskCallLinkRow.id) .where(AsteriskCallLinkRow.status == "ended") .where(~unresolved_recording) .order_by(AsteriskCallLinkRow.id.asc()) .limit(bridge._reconcile_scan_limit()) ).scalars().all() finally: session.close() for link_id in active_ids: session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.id == link_id) ).scalar_one_or_none() if link is None: continue started_dt = bridge._parse_iso(link.started_at) or bridge._parse_iso(link.updated_at) or now age_seconds = max((now - started_dt).total_seconds(), 0) if age_seconds < bridge._reconcile_stale_seconds(): continue live_channels = bridge._list_channels_via_coreshowchannels( call_id=link.call_id, linked_id=link.linked_id, ) if live_channels is None: continue if live_channels: preferred_channel = ( bridge._operator_channel_for_extension(live_channels, link.operator_extension) or live_channels[0] ) if preferred_channel != link.channel_name: link.channel_name = preferred_channel link.updated_at = bridge.utc_now_iso() session.commit() continue if _voice_session_recently_active(session, link=link, now=now): link.updated_at = bridge.utc_now_iso() session.commit() continue ended_exists = bridge._has_voice_event(session, call_id=link.call_id, event_type="call.ended") if not ended_exists: bridge._emit_voice_event( event_type="call.ended", call_id=link.call_id, interaction_id=link.interaction_id, source_event_id=bridge._reconcile_source_event_id(link.call_id, "call.ended"), payload={ "source": "asterisk", "hangup_cause": "unknown", "duration_seconds": None, "linked_id": link.linked_id, "reconciled": True, }, ) now_iso = bridge.utc_now_iso() link.status = "ended" link.telephony_status = "ended" link.ended_at = link.ended_at or now_iso link.updated_at = now_iso session.commit() except Exception: session.rollback() continue finally: session.close() for link_id in ended_ids: session = get_session() try: link = session.execute( select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.id == link_id) ).scalar_one_or_none() if link is None: continue ready_exists = bridge._has_voice_event(session, call_id=link.call_id, event_type="recording.ready") recording_exists = bridge._has_recording(session, call_id=link.call_id) if recording_exists: continue candidate = bridge._find_remote_recording_for_call(link.call_id) if candidate is None: continue remote_path, file_name, mime_type, mtime_dt = candidate idle_seconds = max((now - mtime_dt).total_seconds(), 0) if idle_seconds < bridge._reconcile_settle_seconds(): continue if not ready_exists: bridge._emit_voice_event( event_type="recording.ready", call_id=link.call_id, interaction_id=link.interaction_id, source_event_id=bridge._reconcile_source_event_id(link.call_id, "recording.ready"), payload={ "source": "asterisk", "remote_path": remote_path, "file_name": file_name, "mime_type": mime_type, "duration_seconds": None, "linked_id": link.linked_id, "reconciled": True, }, ) local_path, cleanup_required = bridge._fetch_recording_file(remote_path, file_name) try: bridge._upload_recording( local_path=local_path, call_id=link.call_id, interaction_id=link.interaction_id, source_event_id=bridge._reconcile_source_event_id(link.call_id, "recording.import"), file_name=file_name, mime_type=mime_type, duration_seconds=None, ) finally: if cleanup_required and local_path.exists(): local_path.unlink(missing_ok=True) link.updated_at = bridge.utc_now_iso() session.commit() except Exception: session.rollback() continue finally: session.close()