python-services/voice_orchestrator.py

# voice_orchestrator.py
# Real-time state machine coordinating telephony socket framing, STT, LLM streaming, and TTS playout
#
# v5 — Issue 3 fix: instant audio stop on barge-in (was waiting 1.1-2.0s)
#
#   Previous (v4) behaviour: when a candidate interrupt was detected,
#   the orchestrator transitioned to INTERRUPT_PENDING but did NOT stop
#   audio playback — 'clear_audio' was only sent once the full 1.5s STT
#   confirmation window resolved as 'legitimate'. Combined with VAD
#   debounce + silence window + STT processing, this meant the agent
#   kept talking for 1.1-2.0 SECONDS after the caller started speaking
#   over them — defeating the entire purpose of barge-in.
#
#   Fix: 'clear_audio' is now sent IMMEDIATELY in _enter_interrupt_pending(),
#   the instant a candidate interrupt is detected (within the existing
#   0.2s VAD debounce). Confirmed via frontend scan (useVoiceWebSocket.ts)
#   that 'clear_audio' already does exactly what's needed — stops the
#   active Audio element and empties the playback queue — so NO frontend
#   changes were required. The STT confirmation window (Phase 2) still
#   runs underneath to decide whether to fully cancel the turn (resolved
#   legitimate) or resume playback (resolved accidental) — only the
#   AUDIO STOP MOMENT moved earlier, not the decision logic itself.
#
# v4 — Issue B correction handler + Issue C smart interrupt system
#
#   Issue B (same utterance answered twice):
#     STT now emits is_correction=True / corrects_pending=True instead of
#     a second is_final when Groq provides a same-WC spelling improvement.
#     The-brain handles this in run_stt_loop: if state==THINKING and a
#     correction arrives, the last user message in session history is
#     silently updated with the better text. No new turn is triggered.
#     This eliminates all double-reply occurrences from the last session.
#
#   Issue C (smart interrupt system):
#     Previous: any VAD blip passing 0.2s debounce → hard cancel → silence.
#     No recovery from accidental interrupts. No graceful handover.
#
#     New: three-phase interrupt system.
#
#     Phase 1 — Detection (unchanged):
#       VAD speech_start × 2 within 0.2s → candidate interrupt.
#       State transitions to INTERRUPT_PENDING (new sub-state of SPEAKING).
#       Current TTS sentence continues to completion — agent never cut off
#       mid-word. New TTS sentences are paused but queued in
#       _interrupt_pending_tts so they can be resumed if accidental.
#
#     Phase 2 — STT Confirmation Gate (1.5s window):
#       Wait for is_final from STT. Classify the confirmed speech:
#
#       LEGITIMATE (≥2 words, or 1 meaningful word in INTERRUPT_MEANINGFUL):
#         → Graceful stop after current sentence
#         → Emit bridge phrase "جی، بتائیں" / "جی فرمائیں" to fill silence
#         → Start new LLM turn with the caller's text
#
#       SHORT_MEANINGFUL (1 word in INTERRUPT_SHORT_MEANINGFUL like ہاں/نہیں/جی/ٹھیک):
#         → Same as LEGITIMATE but uses a shorter bridge "جی"
#         → Caller signalled agreement/disagreement, handle as normal turn
#
#       ACCIDENTAL (filler/noise word, or timeout with no is_final):
#         → Resume dispatching the paused TTS queue from where we stopped
#         → Agent continues speaking as if nothing happened
#         → Log "Interrupt resolved as accidental — resuming TTS"
#
#     Phase 3 — Graceful handover:
#       Bridge phrase is synthesised and emitted before the LLM turn fires.
#       This fills the ~1-2s gap while Groq generates the response.
#       Feels natural — like a human agent saying "yes, go ahead" before
#       answering.

import json
import logging
import asyncio
import threading
import time
from enum import Enum
from typing import Optional, List, Tuple

from conversation_manager import ConversationSession
from stt_client import SttClient
import voice_pipeline
import neon_client
import summary

logger = logging.getLogger("voice_orchestrator")

# ── Echo guard (time-window only, content-overlap removed in v3) ──────────────
ECHO_GUARD_SECONDS = 1.2

# ── Smart interrupt constants ─────────────────────────────────────────────────
# How long to wait for STT confirmation after a candidate interrupt
INTERRUPT_CONFIRM_TIMEOUT = 1.5   # seconds

# Single meaningful words that constitute a SHORT_MEANINGFUL interrupt
# (caller said "yes"/"no"/"okay" etc. — still a real turn, just short)
INTERRUPT_SHORT_MEANINGFUL = {
    "ہاں", "نہیں", "جی", "ٹھیک", "ہاں جی", "نہیں جی",
    "رکو", "بس", "اچھا", "ٹھیک ہے", "سمجھ گیا", "سمجھ گئی",
    "han", "nahi", "ji", "theek", "ruko", "bas", "acha",
}

# Hard filler/noise words — if interrupt text is ONLY these, it's accidental
INTERRUPT_NOISE_WORDS = {
    "موسیقی", "میوزک", "ساز", "واہ", "ہم", "مرحبا",
}

# Bridge phrases emitted after a legitimate interrupt to fill the LLM gap
BRIDGE_PHRASES_FULL = [
    "جی، بتائیں۔",
    "جی فرمائیں۔",
    "ہاں، بتائیں۔",
]
BRIDGE_PHRASES_SHORT = [
    "جی۔",
    "ہاں۔",
]

# ── DISABLED (Issue 2 v3) — content-overlap echo guard ───────────────────────
# Preserved as comments. See v3 header for rationale.


class VoiceState(Enum):
    LISTENING          = "listening"
    PROCESSING         = "processing"
    THINKING           = "thinking"
    SPEAKING           = "speaking"
    INTERRUPT_PENDING  = "interrupt_pending"  # Issue C: new state


def _classify_interrupt_text(text: str) -> str:
    """
    Issue C: Classify STT text during interrupt confirmation window.
    Returns: 'legitimate', 'short_meaningful', or 'accidental'
    """
    if not text or not text.strip():
        return "accidental"

    words = text.strip().split()
    word_count = len(words)

    # Pure noise/artifact single word
    if word_count == 1 and words[0] in INTERRUPT_NOISE_WORDS:
        return "accidental"

    # 1 meaningful word
    if word_count == 1:
        if words[0] in INTERRUPT_SHORT_MEANINGFUL:
            return "short_meaningful"
        # Single content word (not in noise, not in short_meaningful)
        # Could be a genuine short command — treat as legitimate
        return "legitimate"

    # 2+ words → always legitimate
    return "legitimate"


def _pick_bridge_phrase(interrupt_class: str, turn_count: int) -> str:
    """Pick a natural bridge phrase based on interrupt type."""
    phrases = BRIDGE_PHRASES_SHORT if interrupt_class == "short_meaningful" else BRIDGE_PHRASES_FULL
    return phrases[turn_count % len(phrases)]


class VoiceOrchestratorSession:
    def __init__(self, websocket, session: ConversationSession, stt_client: SttClient):
        self.websocket   = websocket
        self.session     = session
        self.stt_client  = stt_client
        self.state       = VoiceState.LISTENING
        self.cancel_event            = threading.Event()
        self.active_pipeline_task    = None
        self.stt_loop_task           = None
        self.lock                    = asyncio.Lock()
        self.current_summary         = ""
        self.is_configured           = True
        self._pending_turns          = asyncio.Queue()
        self.keep_alive_task         = None
        self.last_speech_start_time  = 0.0
        self.last_listening_start_time = 0.0

        # Issue 6 (v3): finals queued during THINKING/SPEAKING
        self._pending_finals_during_turn: asyncio.Queue = asyncio.Queue()

        # Issue C: TTS sentences queued but not yet dispatched during
        # INTERRUPT_PENDING — resumed if interrupt is accidental
        self._interrupt_pending_tts: List[Tuple[str, bytes, str]] = []
        # asyncio.Event set when interrupt resolution is known
        self._interrupt_resolved_event: Optional[asyncio.Event] = None
        # Resolution: 'legitimate', 'short_meaningful', 'accidental', or None
        self._interrupt_resolution: Optional[str] = None
        # Text from the caller that caused the interrupt
        self._interrupt_caller_text: Optional[str] = None

    async def set_state(self, new_state: VoiceState):
        async with self.lock:
            self.state = new_state
            logger.info(f"[VoiceSession {self.session.session_id}] State changed to: {new_state.value}")
            if new_state == VoiceState.LISTENING:
                self.last_listening_start_time = time.time()
            await self.send_json({"type": "state", "state": new_state.value})

    async def send_json(self, data: dict):
        try:
            await self.websocket.send_json(data)
        except Exception as e:
            logger.warning(f"Error sending json to client: {e}")

    async def send_binary(self, data: bytes):
        try:
            await self.websocket.send_bytes(data)
        except Exception as e:
            logger.warning(f"Error sending binary to client: {e}")

    async def start(self):
        self.stt_client.start()
        self.stt_loop_task   = asyncio.create_task(self.run_stt_loop())
        self.keep_alive_task = asyncio.create_task(self.run_keep_alive_loop())
        await self.send_json({"type": "state", "state": self.state.value})

    async def run_keep_alive_loop(self):
        try:
            while True:
                await asyncio.sleep(30.0)
                await self.send_json({"type": "ping"})
        except asyncio.CancelledError:
            pass

    async def run_ingress_loop(self):
        try:
            while True:
                message = await self.websocket.receive()
                if "bytes" in message:
                    pcm_data = message["bytes"]
                    if self.is_configured and self.stt_client.connected:
                        self.stt_client.send_pcm(pcm_data)
                elif "text" in message:
                    try:
                        data    = json.loads(message["text"])
                        msg_type = data.get("type")
                        if msg_type == "session_end":
                            logger.info(f"Client requested session end: {self.session.session_id}")
                            break
                        elif msg_type == "text_message":
                            text = data.get("message", "")
                            if text.strip():
                                await self.trigger_assistant_turn(text)
                    except Exception as parse_err:
                        logger.warning(f"Failed to parse text message from client: {parse_err}")
                else:
                    break
        except Exception as e:
            logger.info(f"Ingress loop exception: {e}")
        finally:
            await self.cleanup()

    def _is_echo(self, text: str) -> bool:
        elapsed = time.time() - self.last_listening_start_time
        if self.last_listening_start_time > 0 and elapsed < ECHO_GUARD_SECONDS:
            logger.info(
                f"[VoiceSession {self.session.session_id}] Echo guard (time window, "
                f"{elapsed:.2f}s < {ECHO_GUARD_SECONDS}s) — discarding: '{text[:60]}'"
            )
            return True
        return False

    def _get_last_assistant_message(self) -> str:
        for msg in reversed(self.session.messages):
            if msg.get("role") == "assistant":
                return msg.get("content", "")
        return ""

    async def run_stt_loop(self):
        """Reads events from upstream STT space and drives state machine."""
        try:
            while True:
                event = await self.stt_client.get_event()

                is_vad_start  = (event.get("type") == "vad" and event.get("event") == "speech_start")
                is_partial    = (not event.get("is_final") and event.get("text") and not event.get("is_correction"))
                is_final      = event.get("is_final") and not event.get("is_correction")
                is_correction = event.get("is_correction") and event.get("corrects_pending")

                # ── Issue B: handle STT correction event ─────────────────────
                # Update the last user message text silently — no new turn.
                if is_correction:
                    corrected_text = event.get("text", "").strip()
                    if corrected_text and self.state == VoiceState.THINKING:
                        # Find and update the last user message in session history
                        for i in range(len(self.session.messages) - 1, -1, -1):
                            if self.session.messages[i].get("role") == "user":
                                old_text = self.session.messages[i]["content"]
                                self.session.messages[i]["content"] = corrected_text
                                logger.info(
                                    f"[VoiceSession {self.session.session_id}] "
                                    f"Correction applied: '{old_text}' → '{corrected_text}'"
                                )
                                break
                    continue  # corrections never trigger new turns

                # ── INTERRUPT_PENDING state: waiting for STT confirmation ─────
                if self.state == VoiceState.INTERRUPT_PENDING:
                    if is_final:
                        text = event.get("text", "").strip()
                        if text:
                            resolution = _classify_interrupt_text(text)
                            logger.info(
                                f"[VoiceSession {self.session.session_id}] "
                                f"Interrupt confirmed as '{resolution}': '{text[:60]}'"
                            )
                            self._interrupt_caller_text = text
                            self._interrupt_resolution  = resolution
                            if self._interrupt_resolved_event:
                                self._interrupt_resolved_event.set()
                        # If text is empty keep waiting until timeout
                    continue  # don't fall through to THINKING/SPEAKING/LISTENING

                # ── THINKING / SPEAKING states ────────────────────────────────
                if self.state in (VoiceState.THINKING, VoiceState.SPEAKING):
                    if is_vad_start:
                        now = time.time()
                        if now - self.last_speech_start_time <= 0.2:
                            logger.info(
                                f"[VoiceSession {self.session.session_id}] "
                                "Barge-in candidate — entering INTERRUPT_PENDING"
                            )
                            await self._enter_interrupt_pending()
                        else:
                            logger.info("First VAD speech_start detected. Waiting for confirmation.")
                            self.last_speech_start_time = now
                    elif is_partial:
                        # Partial text during speaking is a strong signal.
                        # Treat directly as legitimate interrupt candidate —
                        # enter INTERRUPT_PENDING so we still gate it.
                        if self.state == VoiceState.SPEAKING:
                            logger.info(
                                f"[VoiceSession {self.session.session_id}] "
                                "Partial text during SPEAKING — entering INTERRUPT_PENDING"
                            )
                            await self._enter_interrupt_pending()
                        else:
                            # THINKING — old behaviour: immediate interrupt
                            logger.info("Barge-in text during THINKING — halting turn.")
                            await self.interrupt_turn()
                    elif is_final:
                        # Issue 6 (v3): queue instead of dropping
                        text = event.get("text", "").strip()
                        if text:
                            logger.info(
                                f"[VoiceSession {self.session.session_id}] Final received "
                                f"while {self.state.value} — queueing: '{text[:60]}'"
                            )
                            await self._pending_finals_during_turn.put(text)

                # ── LISTENING state ───────────────────────────────────────────
                elif self.state == VoiceState.LISTENING:
                    if is_partial:
                        await self.send_json({"type": "partial", "text": event["text"]})
                    elif is_final:
                        text = event.get("text", "").strip()
                        if text:
                            if self._is_echo(text):
                                continue
                            await self.send_json({"type": "user_final", "text": text})
                            await self.trigger_assistant_turn(text)

        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.error(f"STT loop error: {e}")

    async def _enter_interrupt_pending(self):
        """
        Issue C Phase 1: transition to INTERRUPT_PENDING.

        Issue 3 fix (v5): Previously, audio playback only stopped when
        the FULL interrupt was resolved as 'legitimate' — after the
        1.5s STT confirmation window (debounce + silence window + STT
        processing = 1.1s-2.0s of the agent continuing to talk over the
        caller). This contradicted the entire purpose of barge-in.

        Now: 'clear_audio' is sent IMMEDIATELY here, the instant a
        candidate interrupt is detected (VAD debounce, <250ms from the
        caller's first sound). The frontend's flushAudioQueue() stops
        the currently-playing chunk and empties the queue within
        milliseconds — true "instant" barge-in, matching user
        expectation. We do NOT send 'interrupt' yet (that signals a
        full pipeline cancel) — only 'clear_audio' (audio playback
        stop). The LLM pipeline keeps running underneath; on_audio()
        already checks `self.state == INTERRUPT_PENDING` and queues
        any further synthesised sentences into _interrupt_pending_tts
        instead of sending them, so nothing more reaches the frontend
        until resolution.

        If the interrupt resolves as ACCIDENTAL, _resolve_interrupt()
        simply re-sends the queued sentences as fresh audio_header +
        binary messages — the frontend's existing message handling
        naturally re-queues and plays them, no special "resume"
        protocol needed (confirmed via useVoiceWebSocket.ts: every
        audio_header+binary pair is queued via the same code path
        regardless of when it arrives).

        If resolved as LEGITIMATE/SHORT_MEANINGFUL, the full pipeline
        cancel + bridge-phrase flow proceeds exactly as before — the
        only change is that perceived audio silence now starts
        immediately instead of after the full confirmation window.
        """
        async with self.lock:
            self.state = VoiceState.INTERRUPT_PENDING
        logger.info(f"[VoiceSession {self.session.session_id}] State changed to: interrupt_pending")
        await self.send_json({"type": "state", "state": "interrupt_pending"})

        # Issue 3 fix: stop audio playback NOW, not after the 1.5s
        # confirmation window. This is the actual "barge-in" moment —
        # the caller should hear silence within milliseconds, not
        # seconds, regardless of how the interrupt eventually resolves.
        await self.send_json({"type": "clear_audio"})

        self._interrupt_resolved_event = asyncio.Event()
        self._interrupt_resolution     = None
        self._interrupt_caller_text    = None

        # Start timeout task — if no is_final arrives in time → accidental
        asyncio.create_task(self._interrupt_confirmation_timeout())

    async def _interrupt_confirmation_timeout(self):
        """
        Issue C Phase 2: wait for STT confirmation.
        Times out after INTERRUPT_CONFIRM_TIMEOUT seconds → accidental.
        """
        try:
            await asyncio.wait_for(
                self._interrupt_resolved_event.wait(),
                timeout=INTERRUPT_CONFIRM_TIMEOUT,
            )
        except asyncio.TimeoutError:
            logger.info(
                f"[VoiceSession {self.session.session_id}] "
                f"Interrupt timeout ({INTERRUPT_CONFIRM_TIMEOUT}s) — resolving as accidental"
            )
            self._interrupt_resolution = "accidental"
            if self._interrupt_resolved_event:
                self._interrupt_resolved_event.set()

        await self._resolve_interrupt()

    async def _resolve_interrupt(self):
        """
        Issue C Phase 3: act on the confirmed interrupt resolution.

        Issue 3 fix (v5): audio playback was ALREADY stopped immediately
        in _enter_interrupt_pending() (clear_audio sent there). This
        method now only decides what happens NEXT:
          - accidental: resume by re-sending the paused sentences
          - legitimate/short_meaningful: cancel pipeline + bridge phrase
        The 'interrupt'/'clear_audio' sends below for the legitimate
        path are now redundant-but-harmless (frontend queue is already
        empty) — kept for protocol clarity and to handle any sentence
        that slipped through between detection and this resolution.
        """
        resolution  = self._interrupt_resolution or "accidental"
        caller_text = self._interrupt_caller_text

        if resolution == "accidental":
            logger.info(
                f"[VoiceSession {self.session.session_id}] "
                "Interrupt resolved as accidental — resuming TTS"
            )
            # Audio was already stopped in _enter_interrupt_pending().
            # Resume by replaying the paused queue from where we stopped.
            async with self.lock:
                self.state = VoiceState.SPEAKING
            await self.send_json({"type": "state", "state": "speaking"})
            await self.send_json({"type": "interrupt_cancelled"})

            # Replay any TTS sentences that were queued during INTERRUPT_PENDING.
            # These re-enter the frontend's normal audio_header+binary handling
            # path (useVoiceWebSocket.ts) and queue/play exactly like any other
            # turn's audio — no special resume protocol required.
            for sentence, audio_bytes, tts_lang in self._interrupt_pending_tts:
                if not self.cancel_event.is_set():
                    await self.send_json({
                        "type": "audio_header",
                        "sentence": sentence,
                        "lang": tts_lang,
                    })
                    await self.send_binary(audio_bytes)
            self._interrupt_pending_tts.clear()

        else:
            # LEGITIMATE or SHORT_MEANINGFUL — graceful stop
            # (audio already stopped in _enter_interrupt_pending(); this
            # just cancels the underlying pipeline and starts the new turn)
            logger.info(
                f"[VoiceSession {self.session.session_id}] "
                f"Interrupt resolved as '{resolution}' — graceful handover"
            )
            self._interrupt_pending_tts.clear()

            # Hard cancel the pipeline
            self.cancel_event.set()
            if self.active_pipeline_task and not self.active_pipeline_task.done():
                self.active_pipeline_task.cancel()
            self.cancel_event = threading.Event()

            # Clear pending finals from the old turn
            while not self._pending_finals_during_turn.empty():
                self._pending_finals_during_turn.get_nowait()

            await self.send_json({"type": "interrupt"})
            await self.send_json({"type": "clear_audio"})

            # Emit bridge phrase to fill the LLM gap
            if caller_text:
                bridge = _pick_bridge_phrase(resolution, self.session.turn_count)
                logger.info(
                    f"[VoiceSession {self.session.session_id}] "
                    f"Emitting bridge phrase: '{bridge}'"
                )
                await self.send_json({"type": "state", "state": "speaking"})
                bridge_audio, bridge_lang = await voice_pipeline.fire_tts(
                    bridge, "ur", self.session.agent_gender
                )
                if bridge_audio:
                    await self.send_json({
                        "type": "audio_header",
                        "sentence": bridge,
                        "lang": bridge_lang,
                    })
                    await self.send_binary(bridge_audio)

                # Now start the real turn with caller's text
                await self.set_state(VoiceState.THINKING)
                self.cancel_event = threading.Event()
                self.active_pipeline_task = asyncio.create_task(
                    self._execute_turn(caller_text)
                )
            else:
                await self.set_state(VoiceState.LISTENING)

        # Clean up interrupt state
        self._interrupt_resolved_event = None
        self._interrupt_resolution     = None
        self._interrupt_caller_text    = None

    async def _maybe_process_pending_final(self):
        """Issue 6 (v3): drain STT finals queued during THINKING/SPEAKING."""
        if self._pending_finals_during_turn.empty():
            return

        text: Optional[str] = None
        while not self._pending_finals_during_turn.empty():
            text = await self._pending_finals_during_turn.get()

        if not text:
            return

        logger.info(
            f"[VoiceSession {self.session.session_id}] Processing queued final "
            f"(arrived during prior turn): '{text[:60]}'"
        )
        await self.send_json({"type": "user_final", "text": text})
        await self.trigger_assistant_turn(text)

    async def interrupt_turn(self):
        """Hard interrupt — used during THINKING state (pre-speech)."""
        self.cancel_event.set()
        if self.active_pipeline_task and not self.active_pipeline_task.done():
            self.active_pipeline_task.cancel()

        self.cancel_event = threading.Event()

        while not self._pending_finals_during_turn.empty():
            self._pending_finals_during_turn.get_nowait()
        self._interrupt_pending_tts.clear()

        await self.send_json({"type": "interrupt"})
        await self.send_json({"type": "clear_audio"})
        await self.set_state(VoiceState.LISTENING)

    async def trigger_assistant_turn(self, text: str):
        if not self.is_configured:
            logger.info(f"Session not yet configured. Queueing turn: {text}")
            await self._pending_turns.put(text)
            asyncio.create_task(self._drain_pending_turns())
            return

        if self.state in (VoiceState.THINKING, VoiceState.SPEAKING):
            await self.interrupt_turn()

        await self.set_state(VoiceState.THINKING)
        self.cancel_event = threading.Event()
        self.active_pipeline_task = asyncio.create_task(self._execute_turn(text))

    async def _drain_pending_turns(self):
        while not self.is_configured:
            await asyncio.sleep(0.05)
        while not self._pending_turns.empty():
            text = await self._pending_turns.get()
            if self.state in (VoiceState.THINKING, VoiceState.SPEAKING):
                await self.interrupt_turn()
            await self.set_state(VoiceState.THINKING)
            self.cancel_event = threading.Event()
            self.active_pipeline_task = asyncio.create_task(self._execute_turn(text))

    async def _execute_turn(self, text: str):
        try:
            async def on_token(token: str):
                await self.send_json({"type": "token", "text": token})

            async def on_audio(sentence: str, audio_bytes: bytes, tts_lang: str):
                # Issue C: if INTERRUPT_PENDING, queue audio instead of sending
                if self.state == VoiceState.INTERRUPT_PENDING:
                    logger.debug(
                        f"[VoiceSession {self.session.session_id}] "
                        f"TTS queued during INTERRUPT_PENDING: '{sentence[:40]}'"
                    )
                    self._interrupt_pending_tts.append((sentence, audio_bytes, tts_lang))
                    return

                if self.state == VoiceState.THINKING:
                    await self.set_state(VoiceState.SPEAKING)

                await self.send_json({
                    "type": "audio_header",
                    "sentence": sentence,
                    "lang": tts_lang,
                })
                await self.send_binary(audio_bytes)

            async def on_done(full_text: str, language: str, active_summary: str, elapsed_ms: int):
                self.current_summary = active_summary
                await self.send_json({
                    "type": "done",
                    "session_id": self.session.session_id,
                    "full_text": full_text,
                    "language": language,
                    "language_locked": self.session.language_detector.is_locked(),
                    "turn_count": self.session.turn_count,
                    "caller_info": self.session.caller_info.to_dict(),
                    "activeSummary": active_summary,
                    "elapsed_ms": elapsed_ms,
                })
                await self.set_state(VoiceState.LISTENING)
                await self._maybe_process_pending_final()

            await voice_pipeline.run_turn(
                session=self.session,
                message=text,
                cancel_event=self.cancel_event,
                on_token=on_token,
                on_audio=on_audio,
                on_done=on_done,
                agent_gender=self.session.agent_gender,
                current_summary=self.current_summary,
                session_start=int(self.session.created_at * 1000),
            )
        except Exception as err:
            logger.error(f"Error during assistant turn execution: {err}")
            await self.send_json({"type": "error", "detail": str(err)})
            await self.set_state(VoiceState.LISTENING)

    async def cleanup(self):
        self.cancel_event.set()
        if self.active_pipeline_task and not self.active_pipeline_task.done():
            self.active_pipeline_task.cancel()
        if self.stt_loop_task and not self.stt_loop_task.done():
            self.stt_loop_task.cancel()
        if self.keep_alive_task and not self.keep_alive_task.done():
            self.keep_alive_task.cancel()
        await self.stt_client.close()

        transcript_turns = [
            f"{'Caller' if m['role'] == 'user' else 'Agent'}: {m['content']}"
            for m in self.session.messages
        ]
        full_transcript = "\n".join(transcript_turns)
        fields_spec     = self.session.caller_collection_fields
        extracted_data  = self.session.caller_info.to_dict()

        if fields_spec and full_transcript.strip():
            llm_extracted = await summary.extract_caller_data_from_transcript(
                full_transcript, fields_spec
            )
            self.session.caller_info.update_from_llm_extraction(llm_extracted)
            extracted_data = self.session.caller_info.to_dict()

        import time
        duration = int(time.time() - self.session.created_at)

        if getattr(self.session, "company_id", None):
            await neon_client.insert_call_log(
                call_id=self.session.session_id,
                company_id=self.session.company_id,
                phone_line_id=getattr(self.session, "phone_line_id", None),
                full_transcript=full_transcript,
                extracted_caller_data=extracted_data,
                duration_seconds=duration,
                status="completed",
            )

        try:
            await self.send_json({
                "type": "call_summary",
                "session_id": self.session.session_id,
                "extracted_caller_data": extracted_data,
                "duration_seconds": duration,
                "transcript_turns": len(self.session.messages),
            })
        except Exception:
            pass

        logger.info(f"Orchestration session closed and cleaned up: {self.session.session_id}")