"""One agent-handled phone call: audio loop, barge-in, recording, metering (M4). The transport hands us paced 20 ms SLIN frames from Asterisk (AudioSocket) and accepts frames to play back. Flow per §6.2: greeting → collect utterance (VAD) → STT → agent turn (M2 loop, same tools as chat) → TTS → play, with barge-in (caller speech stops playback). Unintelligible audio reaches the agent as '[nerazumljivo]' so the two-attempts rule (§6.4/A.6) lives in the prompt. """ from __future__ import annotations import asyncio import logging import os import time import uuid from dataclasses import dataclass from sqlalchemy import select from gogo.agent.loop import AgentConversation from gogo.config import get_settings from gogo.db import get_sessionmaker from gogo.metering import record_agent_call from gogo.models import Call, CallRegistration, Tenant, utcnow from gogo.voice.audio import FRAME_BYTES, FRAME_MS, CallRecorder from gogo.voice.stt import get_stt from gogo.voice.tts import get_tts from gogo.voice.vad import UtteranceCollector log = logging.getLogger("gogo.voice.call") UNCLEAR_MARKER = "[nerazumljivo]" GOODBYE_TIMEOUT = ( "Izgleda da veza ne radi. Salon će vidjeti vaš broj i javiti vam se. Prijatno!" ) BARGE_IN_FRAMES = 6 # 120 ms of speech during playback stops it @dataclass class CallLimits: inactivity_s: float = 30.0 max_call_s: float = 600.0 max_turns: int = 30 class Transport: """What CallSession needs from the wire (implemented by AudioSocket + fakes).""" async def read_frame(self) -> bytes | None: """Next inbound 20 ms frame, or None on hangup.""" raise NotImplementedError async def send_audio(self, pcm: bytes) -> None: raise NotImplementedError async def hangup(self) -> None: raise NotImplementedError class CallSession: def __init__( self, transport: Transport, registration_id: uuid.UUID, *, limits: CallLimits | None = None, pace: float = FRAME_MS / 1000, ): self.transport = transport self.registration_id = registration_id self.limits = limits or CallLimits() self.pace = pace self.recorder = CallRecorder() self.collector = UtteranceCollector() self._started = time.monotonic() async def run(self) -> None: async with get_sessionmaker()() as session: reg = ( await session.execute( select(CallRegistration).where(CallRegistration.id == self.registration_id) ) ).scalar_one_or_none() if reg is None: log.error("no registration for call %s — hanging up", self.registration_id) await self.transport.hangup() return tenant = ( await session.execute(select(Tenant).where(Tenant.id == reg.tenant_id)) ).scalar_one() reg.handled_by_agent = True call = Call( tenant_id=tenant.id, caller_msisdn=reg.caller_msisdn, started_at=utcnow(), ) session.add(call) await session.flush() convo = AgentConversation( session, tenant, channel="voice", caller_phone=reg.caller_msisdn, call_id=call.id, ) tts, voice = get_tts(tenant.tts_voice) stt = get_stt() try: greeting = await convo.greeting() await session.commit() await self._speak(tts, voice, greeting) for _turn in range(self.limits.max_turns): if time.monotonic() - self._started > self.limits.max_call_s: log.warning("max call duration hit (tenant %s)", tenant.slug) break utterance = await self._collect_utterance() if utterance is None: # hangup break if utterance == b"": # inactivity await self._speak(tts, voice, GOODBYE_TIMEOUT) break t0 = time.monotonic() text = await stt.transcribe(utterance) stt_ms = int((time.monotonic() - t0) * 1000) t0 = time.monotonic() reply = await convo.user_turn(text or UNCLEAR_MARKER) llm_ms = int((time.monotonic() - t0) * 1000) await session.commit() t0 = time.monotonic() interrupted = await self._speak(tts, voice, reply) tts_ms = int((time.monotonic() - t0) * 1000) trace = call.trace or {} turns = trace.setdefault("turns", []) turns.append( {"stt_ms": stt_ms, "llm_ms": llm_ms, "speak_ms": tts_ms, "interrupted": interrupted, "chars": len(reply)} ) call.trace = dict(trace) if self._agent_closed(convo): break except Exception: # noqa: BLE001 — a crash must still finalize the call log.exception("call session error (tenant %s)", tenant.slug) finally: await self._finalize(session, tenant, call, convo, reg) await self.transport.hangup() # -- audio helpers ------------------------------------------------------ async def _speak(self, tts, voice: str, text: str) -> bool: """Synthesize and play; returns True if the caller barged in.""" if not text: return False try: pcm = await tts.synthesize(text, voice) except Exception: # noqa: BLE001 log.exception("TTS failed — continuing without audio") return False voiced_streak = 0 for i in range(0, len(pcm), FRAME_BYTES): frame = pcm[i : i + FRAME_BYTES] await self.transport.send_audio(frame) self.recorder.add_agent(frame) if self.pace: await asyncio.sleep(self.pace) # barge-in: watch inbound audio while we play inbound = await self._poll_frame() if inbound is not None: self.recorder.add_caller(inbound) if self.collector.is_speech(inbound): voiced_streak += 1 if voiced_streak >= BARGE_IN_FRAMES: self.collector.reset() self.collector.feed(inbound) return True else: voiced_streak = 0 return False async def _poll_frame(self) -> bytes | None: try: return await asyncio.wait_for(self.transport.read_frame(), timeout=0.001) except TimeoutError: return None async def _collect_utterance(self) -> bytes | None: """None = hangup, b'' = inactivity timeout, else utterance PCM.""" deadline = time.monotonic() + self.limits.inactivity_s while True: timeout = deadline - time.monotonic() if timeout <= 0: return b"" try: frame = await asyncio.wait_for(self.transport.read_frame(), timeout=timeout) except TimeoutError: return b"" if frame is None: return None self.recorder.add_caller(frame) utterance = self.collector.feed(frame) if utterance is not None: return utterance def _agent_closed(self, convo: AgentConversation) -> bool: """The closing line ends the call from our side (short calls §2.5).""" last = next((t.text for t in reversed(convo.turns) if t.role == "assistant"), "") lowered = last.lower() return ("prijatno" in lowered or "doviđenja" in lowered) and bool( convo.executor.created_request_id or convo.executor.took_message ) async def _finalize(self, session, tenant, call, convo, reg) -> None: try: call.duration_s = int(time.monotonic() - self._started) call.agent_seconds_charged = call.duration_s call.outcome = convo.outcome call.transcript = convo.transcript rec_dir = get_settings().recordings_dir os.makedirs(rec_dir, exist_ok=True) path = os.path.join(rec_dir, f"{call.id}.wav") with open(path, "wb") as f: f.write(self.recorder.to_wav()) call.recording_path = path reg.finalized = True await record_agent_call(session, tenant, call.agent_seconds_charged) await session.commit() log.info( "call %s finalized: %ss outcome=%s tenant=%s", call.id, call.duration_s, call.outcome, tenant.slug, ) except Exception: # noqa: BLE001 log.exception("failed to finalize call %s", call.id) await session.rollback()