┌─────────────────────────────────────────────────────────────────────────────┐
│                         VOICE PIPELINE ARCHITECTURE                         │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   ┌─────────────────────────────────────────────────────────────────────┐   │
│   │                         AUDIO INPUT STAGE                           │   │
│   │                                                                     │   │
│   │   Caller Audio ──▶ [LiveKit] ──▶ [Audio Buffer] ──▶ [VAD]          │   │
│   │                                        │              │             │   │
│   │                                        │              ▼             │   │
│   │                                        │         Voice Active?      │   │
│   │                                        │         ┌────┴────┐        │   │
│   │                                        │        Yes        No       │   │
│   │                                        │         │          │       │   │
│   │                                        ▼         ▼          ▼       │   │
│   │                                   [STT Engine]   │     (silence)    │   │
│   │                                        │         │                  │   │
│   └────────────────────────────────────────┼─────────┼──────────────────┘   │
│                                            │         │                      │
│   ┌────────────────────────────────────────▼─────────▼──────────────────┐   │
│   │                       PROCESSING STAGE                              │   │
│   │                                                                     │   │
│   │   Partial          Final              Endpointing                   │   │
│   │   Transcript ───▶  Transcript ───────▶ Detected?                   │   │
│   │       │                │               ┌────┴────┐                  │   │
│   │       │                │              Yes        No                 │   │
│   │       │                ▼               │          │                 │   │
│   │       │         [Context Assembly]     │     (continue              │   │
│   │       │                │               │      buffering)            │   │
│   │       ▼                ▼               ▼                            │   │
│   │   (display)     [LLM Processing] ◀────┘                            │   │
│   │                        │                                            │   │
│   │                        │ streaming tokens                           │   │
│   │                        ▼                                            │   │
│   │                 [Response Router]                                   │   │
│   │                   ┌────┴────┐                                       │   │
│   │              Speech      Tool Call                                  │   │
│   │                 │            │                                      │   │
│   └─────────────────┼────────────┼──────────────────────────────────────┘   │
│                     │            │                                          │
│   ┌─────────────────▼────────────▼──────────────────────────────────────┐   │
│   │                       OUTPUT STAGE                                  │   │
│   │                                                                     │   │
│   │   [Sentence Buffer] ──▶ [TTS Engine] ──▶ [Audio Queue] ──▶ LiveKit │   │
│   │         │                     │                │                    │   │
│   │         │                     │                │                    │   │
│   │         ▼                     ▼                ▼                    │   │
│   │   Natural break         Audio chunks      Playback to              │   │
│   │   detection             generated         caller                    │   │
│   │                                                                     │   │
│   │                                                                     │   │
│   │   [Tool Executor] ──▶ [Result Formatter] ──▶ (back to LLM)         │   │
│   │                                                                     │   │
│   └─────────────────────────────────────────────────────────────────────┘   │
│                                                                             │
│   ┌─────────────────────────────────────────────────────────────────────┐   │
│   │                      INTERRUPTION HANDLING                          │   │
│   │                                                                     │   │
│   │   VAD during playback ──▶ [Interrupt Detector] ──▶ Stop TTS        │   │
│   │                                   │                    │            │   │
│   │                                   ▼                    ▼            │   │
│   │                           Threshold met?          Clear queue       │   │
│   │                                   │                    │            │   │
│   │                                   ▼                    ▼            │   │
│   │                           Process new input      Resume listening   │   │
│   │                                                                     │   │
│   └─────────────────────────────────────────────────────────────────────┘   │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                           DATA FLOW SUMMARY                                 │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   STAGE 1: Audio Capture                                                    │
│   ───────────────────────                                                   │
│   Format: PCM 16-bit, 16kHz mono (STT) / 48kHz stereo (transport)          │
│   Chunk size: 20ms frames (320 samples at 16kHz)                           │
│   Buffer: 100ms sliding window for VAD                                      │
│                                                                             │
│   STAGE 2: Speech Detection                                                 │
│   ─────────────────────────                                                 │
│   VAD checks each 20ms frame                                                │
│   Speech threshold: 0.5 probability                                         │
│   Minimum speech duration: 250ms                                            │
│   Silence padding: 300ms before endpoint                                    │
│                                                                             │
│   STAGE 3: Transcription                                                    │
│   ─────────────────────────                                                 │
│   Streaming WebSocket to Deepgram                                           │
│   Interim results every ~100ms                                              │
│   Final transcript on endpoint                                              │
│   Output: JSON with text, confidence, timestamps                            │
│                                                                             │
│   STAGE 4: Context Assembly                                                 │
│   ─────────────────────────                                                 │
│   Retrieve conversation history (last 10 turns)                             │
│   Add system prompt + knowledge base context                                │
│   Include tool definitions                                                  │
│   Format for Claude API                                                     │
│                                                                             │
│   STAGE 5: LLM Processing                                                   │
│   ──────────────────────                                                    │
│   Streaming API call to Claude                                              │
│   Tokens arrive in ~50ms chunks                                             │
│   Parse for tool calls vs speech                                            │
│   Output: Text stream or tool call JSON                                     │
│                                                                             │
│   STAGE 6: Speech Synthesis                                                 │
│   ────────────────────────                                                  │
│   Buffer text until sentence boundary                                       │
│   Send to Chatterbox TTS                                                    │
│   Stream audio chunks back                                                  │
│   Output: PCM audio at 24kHz                                                │
│                                                                             │
│   STAGE 7: Audio Playback                                                   │
│   ───────────────────────                                                   │
│   Resample to 48kHz if needed                                               │
│   Queue chunks for smooth playback                                          │
│   Publish to LiveKit track                                                  │
│   Monitor for interruptions                                                 │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                         PIPELINE STATE MACHINE                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│                              ┌─────────┐                                    │
│                              │  IDLE   │                                    │
│                              └────┬────┘                                    │
│                                   │                                         │
│                          voice detected                                     │
│                                   │                                         │
│                                   ▼                                         │
│                           ┌───────────────┐                                 │
│                           │   LISTENING   │◀──────────────┐                │
│                           └───────┬───────┘               │                │
│                                   │                       │                │
│                          speech ended                     │                │
│                                   │                       │                │
│                                   ▼                       │                │
│                          ┌───────────────┐                │                │
│                          │  PROCESSING   │                │                │
│                          └───────┬───────┘                │                │
│                                  │                        │                │
│                    ┌─────────────┼─────────────┐          │                │
│                    │             │             │          │                │
│               tool call     response      error          │                │
│                    │         ready          │             │                │
│                    ▼             │           │             │                │
│           ┌───────────────┐     │           │             │                │
│           │EXECUTING_TOOL │     │           │             │                │
│           └───────┬───────┘     │           │             │                │
│                   │             │           │             │                │
│              result ready       │           │             │                │
│                   │             │           │             │                │
│                   └──────┬──────┘           │             │                │
│                          │                  │             │                │
│                          ▼                  │             │                │
│                   ┌───────────────┐         │             │                │
│                   │   SPEAKING    │─────────┼─────────────┘                │
│                   └───────┬───────┘         │       (on completion         │
│                           │                 │        or interruption)      │
│                    interrupted              │                              │
│                           │                 │                              │
│                           ▼                 │                              │
│                  ┌─────────────────┐        │                              │
│                  │  INTERRUPTED   │────────┘                               │
│                  └─────────────────┘   (resume listening)                  │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                         LATENCY BUDGET BREAKDOWN                            │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   COMPONENT                    TARGET      P50       P95       P99          │
│   ─────────────────────────────────────────────────────────────────────     │
│                                                                             │
│   1. Endpointing delay         200ms      150ms     250ms     350ms        │
│      (silence detection)                                                    │
│                                                                             │
│   2. STT finalization          100ms       80ms     150ms     200ms        │
│      (final transcript)                                                     │
│                                                                             │
│   3. Context assembly           20ms       15ms      30ms      50ms        │
│      (build prompt)                                                         │
│                                                                             │
│   4. Network to LLM             30ms       20ms      50ms      80ms        │
│      (API request)                                                          │
│                                                                             │
│   5. LLM TTFB                  200ms      150ms     300ms     500ms        │
│      (first token)                                                          │
│                                                                             │
│   6. Sentence accumulation     100ms       80ms     150ms     200ms        │
│      (buffer until boundary)                                                │
│                                                                             │
│   7. Network to TTS             20ms       15ms      30ms      50ms        │
│      (API request)                                                          │
│                                                                             │
│   8. TTS TTFB                  150ms      100ms     200ms     300ms        │
│      (first audio chunk)                                                    │
│                                                                             │
│   9. Audio transport            30ms       20ms      40ms      60ms        │
│      (to caller)                                                            │
│   ─────────────────────────────────────────────────────────────────────     │
│                                                                             │
│   TOTAL TTFB                   850ms      630ms    1200ms    1790ms        │
│                                                                             │
│   Note: Components 1-5 are on the critical path to TTFB                    │
│         Components 6-9 add to perceived response time                       │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

# Pipeline latency instrumentation points

LATENCY_CHECKPOINTS = {
    "vad_speech_start": "Voice activity detected",
    "vad_speech_end": "Voice activity ended (endpoint)",
    "stt_interim_first": "First interim transcript",
    "stt_final": "Final transcript received",
    "context_assembled": "Prompt built",
    "llm_request_sent": "LLM API request sent",
    "llm_first_token": "First LLM token received",
    "llm_sentence_complete": "First sentence complete",
    "tts_request_sent": "TTS API request sent",
    "tts_first_chunk": "First audio chunk received",
    "audio_playback_start": "Audio playback began",
}

class LatencyTracker:
    """Track latency through the pipeline."""
    
    def __init__(self, call_id: str, turn_id: str):
        self.call_id = call_id
        self.turn_id = turn_id
        self.checkpoints: dict[str, float] = {}
        self.start_time: Optional[float] = None
    
    def mark(self, checkpoint: str) -> None:
        """Record a checkpoint timestamp."""
        now = time.perf_counter()
        
        if self.start_time is None:
            self.start_time = now
        
        self.checkpoints[checkpoint] = now
        
        # Log checkpoint
        elapsed = (now - self.start_time) * 1000
        logger.debug(
            f"[{self.call_id}:{self.turn_id}] "
            f"{checkpoint}: {elapsed:.1f}ms"
        )
    
    def get_metrics(self) -> dict:
        """Calculate latency metrics."""
        if not self.checkpoints:
            return {}
        
        metrics = {}
        
        # Calculate durations between checkpoints
        if "vad_speech_end" in self.checkpoints and "stt_final" in self.checkpoints:
            metrics["stt_latency"] = (
                self.checkpoints["stt_final"] - 
                self.checkpoints["vad_speech_end"]
            ) * 1000
        
        if "stt_final" in self.checkpoints and "llm_first_token" in self.checkpoints:
            metrics["llm_ttfb"] = (
                self.checkpoints["llm_first_token"] - 
                self.checkpoints["stt_final"]
            ) * 1000
        
        if "llm_sentence_complete" in self.checkpoints and "tts_first_chunk" in self.checkpoints:
            metrics["tts_ttfb"] = (
                self.checkpoints["tts_first_chunk"] - 
                self.checkpoints["llm_sentence_complete"]
            ) * 1000
        
        # End-to-end latency
        if "vad_speech_end" in self.checkpoints and "audio_playback_start" in self.checkpoints:
            metrics["e2e_latency"] = (
                self.checkpoints["audio_playback_start"] - 
                self.checkpoints["vad_speech_end"]
            ) * 1000
        
        return metrics
    
    def emit_metrics(self) -> None:
        """Emit metrics to monitoring system."""
        metrics = self.get_metrics()
        
        for name, value in metrics.items():
            prometheus_histogram(
                f"voice_pipeline_{name}_ms",
                value,
                labels={
                    "call_id": self.call_id,
                }
            )

# pipeline/vad/silero_vad.py

import torch
import numpy as np
from typing import Optional
from dataclasses import dataclass

@dataclass
class VADConfig:
    """Configuration for Silero VAD."""
    threshold: float = 0.5          # Speech probability threshold
    min_speech_ms: int = 250        # Minimum speech duration
    min_silence_ms: int = 300       # Silence before endpoint
    window_size_ms: int = 100       # Analysis window
    sample_rate: int = 16000        # Expected sample rate
    
class SileroVAD:
    """
    Silero VAD wrapper for real-time voice activity detection.
    
    Processes audio in 20ms frames and maintains state for
    speech detection and endpointing.
    """
    
    def __init__(self, config: VADConfig = None):
        self.config = config or VADConfig()
        
        # Load Silero VAD model
        self.model, self.utils = torch.hub.load(
            repo_or_dir='snakers4/silero-vad',
            model='silero_vad',
            force_reload=False
        )
        
        # State tracking
        self._is_speaking = False
        self._speech_start_time: Optional[float] = None
        self._silence_start_time: Optional[float] = None
        self._speech_frames: int = 0
        self._silence_frames: int = 0
        
        # Frame timing
        self._frame_duration_ms = 20  # Silero expects 20ms frames
        self._samples_per_frame = int(
            self.config.sample_rate * self._frame_duration_ms / 1000
        )
    
    def reset(self) -> None:
        """Reset VAD state for new utterance."""
        self.model.reset_states()
        self._is_speaking = False
        self._speech_start_time = None
        self._silence_start_time = None
        self._speech_frames = 0
        self._silence_frames = 0
    
    def process_frame(
        self,
        audio_frame: np.ndarray,
        timestamp: float
    ) -> dict:
        """
        Process a single audio frame.
        
        Args:
            audio_frame: PCM audio samples (16-bit, 16kHz)
            timestamp: Frame timestamp in seconds
        
        Returns:
            Dict with speech detection results
        """
        # Ensure correct frame size
        if len(audio_frame) != self._samples_per_frame:
            raise ValueError(
                f"Expected {self._samples_per_frame} samples, "
                f"got {len(audio_frame)}"
            )
        
        # Convert to float32 tensor
        audio_tensor = torch.from_numpy(audio_frame).float()
        audio_tensor = audio_tensor / 32768.0  # Normalize 16-bit
        
        # Get speech probability
        speech_prob = self.model(
            audio_tensor,
            self.config.sample_rate
        ).item()
        
        # Determine if this frame contains speech
        is_speech = speech_prob >= self.config.threshold
        
        # Update state machine
        result = self._update_state(is_speech, timestamp)
        result["speech_probability"] = speech_prob
        result["is_speech_frame"] = is_speech
        
        return result
    
    def _update_state(
        self,
        is_speech: bool,
        timestamp: float
    ) -> dict:
        """Update VAD state machine."""
        result = {
            "event": None,
            "is_speaking": self._is_speaking,
        }
        
        min_speech_frames = self.config.min_speech_ms // self._frame_duration_ms
        min_silence_frames = self.config.min_silence_ms // self._frame_duration_ms
        
        if is_speech:
            self._silence_frames = 0
            self._speech_frames += 1
            
            # Speech start detection
            if not self._is_speaking and self._speech_frames >= min_speech_frames:
                self._is_speaking = True
                self._speech_start_time = timestamp - (
                    self._speech_frames * self._frame_duration_ms / 1000
                )
                result["event"] = "speech_start"
                result["speech_start_time"] = self._speech_start_time
        else:
            self._speech_frames = 0
            
            if self._is_speaking:
                self._silence_frames += 1
                
                # Endpoint detection
                if self._silence_frames >= min_silence_frames:
                    self._is_speaking = False
                    speech_duration = timestamp - self._speech_start_time
                    result["event"] = "speech_end"
                    result["speech_duration"] = speech_duration
                    self._speech_start_time = None
        
        result["is_speaking"] = self._is_speaking
        return result
    
    @property
    def is_speaking(self) -> bool:
        """Current speech state."""
        return self._is_speaking


class VADProcessor:
    """
    High-level VAD processor with buffering and event emission.
    """
    
    def __init__(
        self,
        config: VADConfig = None,
        on_speech_start: callable = None,
        on_speech_end: callable = None
    ):
        self.vad = SileroVAD(config)
        self.on_speech_start = on_speech_start
        self.on_speech_end = on_speech_end
        
        self._audio_buffer = []
        self._frame_size = self.vad._samples_per_frame
    
    async def process_audio(
        self,
        audio_chunk: bytes,
        timestamp: float
    ) -> None:
        """
        Process incoming audio chunk.
        
        Audio may arrive in varying chunk sizes; this method
        buffers and processes in correct frame sizes.
        """
        # Convert bytes to numpy array
        samples = np.frombuffer(audio_chunk, dtype=np.int16)
        
        # Add to buffer
        self._audio_buffer.extend(samples)
        
        # Process complete frames
        while len(self._audio_buffer) >= self._frame_size:
            frame = np.array(self._audio_buffer[:self._frame_size])
            self._audio_buffer = self._audio_buffer[self._frame_size:]
            
            result = self.vad.process_frame(frame, timestamp)
            
            # Emit events
            if result["event"] == "speech_start" and self.on_speech_start:
                await self.on_speech_start(result)
            elif result["event"] == "speech_end" and self.on_speech_end:
                await self.on_speech_end(result)

# pipeline/vad/endpointing.py

from enum import Enum
from dataclasses import dataclass
from typing import Optional
import re

class EndpointReason(Enum):
    SILENCE = "silence"              # Silence duration exceeded
    PUNCTUATION = "punctuation"      # Sentence-ending punctuation detected
    TURN_TAKING = "turn_taking"      # Linguistic turn-taking cue
    TIMEOUT = "timeout"              # Maximum utterance duration
    FORCED = "forced"                # Manually forced endpoint

@dataclass
class EndpointConfig:
    """Configuration for endpoint detection."""
    silence_threshold_ms: int = 300       # Silence before endpoint
    min_utterance_ms: int = 500           # Minimum utterance length
    max_utterance_ms: int = 30000         # Maximum utterance length
    punctuation_endpoint: bool = True     # Use punctuation as endpoint hint
    aggressive_mode: bool = False         # Faster endpoints for quick responses

class EndpointDetector:
    """
    Multi-strategy endpoint detection.
    
    Combines VAD silence detection with linguistic cues
    from interim transcripts.
    """
    
    def __init__(self, config: EndpointConfig = None):
        self.config = config or EndpointConfig()
        
        # Sentence-ending patterns
        self._sentence_end_pattern = re.compile(
            r'[.!?][\s]*$'
        )
        
        # Turn-taking cue patterns (questions, trailing off)
        self._turn_taking_patterns = [
            re.compile(r'\?[\s]*$'),           # Questions
            re.compile(r'(?:right|okay|yeah|yes|no)[.?]?[\s]*$', re.I),
            re.compile(r'\.{3,}[\s]*$'),       # Ellipsis
        ]
        
        self._utterance_start: Optional[float] = None
        self._last_speech_time: Optional[float] = None
        self._current_transcript: str = ""
    
    def start_utterance(self, timestamp: float) -> None:
        """Mark the start of a new utterance."""
        self._utterance_start = timestamp
        self._last_speech_time = timestamp
        self._current_transcript = ""
    
    def update_transcript(self, transcript: str) -> None:
        """Update with latest transcript."""
        self._current_transcript = transcript
    
    def update_speech(self, timestamp: float) -> None:
        """Update last speech time."""
        self._last_speech_time = timestamp
    
    def check_endpoint(
        self,
        timestamp: float,
        vad_is_speaking: bool
    ) -> Optional[EndpointReason]:
        """
        Check if we should endpoint now.
        
        Returns EndpointReason if endpoint detected, None otherwise.
        """
        if self._utterance_start is None:
            return None
        
        utterance_duration = (timestamp - self._utterance_start) * 1000
        silence_duration = (timestamp - self._last_speech_time) * 1000
        
        # Check maximum duration
        if utterance_duration >= self.config.max_utterance_ms:
            return EndpointReason.TIMEOUT
        
        # Check minimum duration
        if utterance_duration < self.config.min_utterance_ms:
            return None
        
        # VAD-based silence endpoint
        if not vad_is_speaking:
            threshold = self.config.silence_threshold_ms
            
            # Reduce threshold if we have a complete sentence
            if self.config.punctuation_endpoint:
                if self._sentence_end_pattern.search(self._current_transcript):
                    threshold = threshold * 0.7  # 30% faster endpoint
            
            # Reduce threshold in aggressive mode
            if self.config.aggressive_mode:
                threshold = threshold * 0.6
            
            if silence_duration >= threshold:
                return EndpointReason.SILENCE
        
        # Linguistic turn-taking cues (even during speech)
        if self._current_transcript:
            for pattern in self._turn_taking_patterns:
                if pattern.search(self._current_transcript):
                    # Still need some silence
                    if silence_duration >= self.config.silence_threshold_ms * 0.5:
                        return EndpointReason.TURN_TAKING
        
        return None
    
    def force_endpoint(self) -> EndpointReason:
        """Force an immediate endpoint."""
        return EndpointReason.FORCED
    
    def reset(self) -> None:
        """Reset state for next utterance."""
        self._utterance_start = None
        self._last_speech_time = None
        self._current_transcript = ""

# pipeline/stt/deepgram_client.py

import asyncio
import json
from typing import AsyncIterator, Optional, Callable
from dataclasses import dataclass
import websockets

@dataclass
class DeepgramConfig:
    """Configuration for Deepgram STT."""
    api_key: str
    model: str = "nova-2"
    language: str = "en-US"
    punctuate: bool = True
    interim_results: bool = True
    utterance_end_ms: int = 1000
    vad_events: bool = True
    smart_format: bool = True
    diarize: bool = False
    sample_rate: int = 16000
    encoding: str = "linear16"
    channels: int = 1

@dataclass 
class TranscriptResult:
    """Transcription result from Deepgram."""
    text: str
    is_final: bool
    confidence: float
    words: list
    speech_final: bool
    start_time: float
    end_time: float

class DeepgramSTT:
    """
    Deepgram streaming STT client.
    
    Maintains a WebSocket connection for real-time transcription
    with interim and final results.
    """
    
    WS_URL = "wss://api.deepgram.com/v1/listen"
    
    def __init__(
        self,
        config: DeepgramConfig,
        on_transcript: Callable[[TranscriptResult], None] = None,
        on_utterance_end: Callable[[], None] = None
    ):
        self.config = config
        self.on_transcript = on_transcript
        self.on_utterance_end = on_utterance_end
        
        self._ws: Optional[websockets.WebSocketClientProtocol] = None
        self._running = False
        self._receive_task: Optional[asyncio.Task] = None
    
    def _build_url(self) -> str:
        """Build WebSocket URL with query parameters."""
        params = {
            "model": self.config.model,
            "language": self.config.language,
            "punctuate": str(self.config.punctuate).lower(),
            "interim_results": str(self.config.interim_results).lower(),
            "utterance_end_ms": str(self.config.utterance_end_ms),
            "vad_events": str(self.config.vad_events).lower(),
            "smart_format": str(self.config.smart_format).lower(),
            "diarize": str(self.config.diarize).lower(),
            "sample_rate": str(self.config.sample_rate),
            "encoding": self.config.encoding,
            "channels": str(self.config.channels),
        }
        
        query_string = "&".join(f"{k}={v}" for k, v in params.items())
        return f"{self.WS_URL}?{query_string}"
    
    async def connect(self) -> None:
        """Establish WebSocket connection to Deepgram."""
        url = self._build_url()
        
        self._ws = await websockets.connect(
            url,
            extra_headers={
                "Authorization": f"Token {self.config.api_key}",
            },
            ping_interval=20,
            ping_timeout=10,
        )
        
        self._running = True
        self._receive_task = asyncio.create_task(self._receive_loop())
        
        logger.info("Deepgram STT connected")
    
    async def disconnect(self) -> None:
        """Close the WebSocket connection."""
        self._running = False
        
        if self._ws:
            # Send close message
            try:
                await self._ws.send(json.dumps({"type": "CloseStream"}))
            except Exception:
                pass
            
            await self._ws.close()
            self._ws = None
        
        if self._receive_task:
            self._receive_task.cancel()
            try:
                await self._receive_task
            except asyncio.CancelledError:
                pass
        
        logger.info("Deepgram STT disconnected")
    
    async def send_audio(self, audio_chunk: bytes) -> None:
        """
        Send audio data to Deepgram.
        
        Args:
            audio_chunk: PCM audio bytes (16-bit, 16kHz)
        """
        if self._ws and self._running:
            try:
                await self._ws.send(audio_chunk)
            except websockets.ConnectionClosed:
                logger.warning("Deepgram connection closed while sending")
    
    async def _receive_loop(self) -> None:
        """Receive and process transcription results."""
        while self._running and self._ws:
            try:
                message = await self._ws.recv()
                data = json.loads(message)
                
                await self._handle_message(data)
                
            except websockets.ConnectionClosed:
                logger.warning("Deepgram connection closed")
                break
            except json.JSONDecodeError as e:
                logger.error(f"Invalid JSON from Deepgram: {e}")
            except Exception as e:
                logger.error(f"Error in Deepgram receive loop: {e}")
    
    async def _handle_message(self, data: dict) -> None:
        """Handle a message from Deepgram."""
        msg_type = data.get("type")
        
        if msg_type == "Results":
            await self._handle_results(data)
        elif msg_type == "UtteranceEnd":
            if self.on_utterance_end:
                await self.on_utterance_end()
        elif msg_type == "SpeechStarted":
            logger.debug("Deepgram: Speech started")
        elif msg_type == "Metadata":
            logger.debug(f"Deepgram metadata: {data}")
        elif msg_type == "Error":
            logger.error(f"Deepgram error: {data}")
    
    async def _handle_results(self, data: dict) -> None:
        """Handle transcription results."""
        channel = data.get("channel", {})
        alternatives = channel.get("alternatives", [])
        
        if not alternatives:
            return
        
        alt = alternatives[0]
        transcript = alt.get("transcript", "")
        
        if not transcript:
            return
        
        result = TranscriptResult(
            text=transcript,
            is_final=data.get("is_final", False),
            confidence=alt.get("confidence", 0.0),
            words=alt.get("words", []),
            speech_final=data.get("speech_final", False),
            start_time=data.get("start", 0.0),
            end_time=data.get("duration", 0.0) + data.get("start", 0.0),
        )
        
        if self.on_transcript:
            await self.on_transcript(result)


class STTManager:
    """
    High-level STT manager with connection lifecycle.
    """
    
    def __init__(self, config: DeepgramConfig):
        self.config = config
        self._client: Optional[DeepgramSTT] = None
        self._transcript_buffer: str = ""
        self._final_transcript: str = ""
    
    async def start_session(
        self,
        on_interim: Callable[[str], None] = None,
        on_final: Callable[[str], None] = None
    ) -> None:
        """Start a new transcription session."""
        
        async def handle_transcript(result: TranscriptResult):
            if result.is_final:
                self._final_transcript += result.text + " "
                self._transcript_buffer = ""
                if on_final:
                    await on_final(self._final_transcript.strip())
            else:
                self._transcript_buffer = result.text
                if on_interim:
                    full_transcript = self._final_transcript + result.text
                    await on_interim(full_transcript.strip())
        
        self._client = DeepgramSTT(
            config=self.config,
            on_transcript=handle_transcript,
        )
        
        await self._client.connect()
    
    async def send_audio(self, audio_chunk: bytes) -> None:
        """Send audio to the STT engine."""
        if self._client:
            await self._client.send_audio(audio_chunk)
    
    async def end_session(self) -> str:
        """End the session and return final transcript."""
        if self._client:
            await self._client.disconnect()
            self._client = None
        
        final = self._final_transcript.strip()
        self._final_transcript = ""
        self._transcript_buffer = ""
        
        return final
    
    def get_current_transcript(self) -> str:
        """Get the current transcript including buffer."""
        return (self._final_transcript + self._transcript_buffer).strip()

# pipeline/stt/transcript_processor.py

import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class ProcessedTranscript:
    """Processed and normalized transcript."""
    raw_text: str
    normalized_text: str
    detected_intent: Optional[str]
    contains_question: bool
    word_count: int

class TranscriptProcessor:
    """
    Post-process transcripts for LLM consumption.
    """
    
    def __init__(self):
        # Common speech recognition errors to fix
        self._corrections = {
            r'\bum+\b': '',
            r'\buh+\b': '',
            r'\blike,?\s+': '',
            r'\byou know,?\s+': '',
            r'\bI mean,?\s+': '',
            r'\s+': ' ',  # Multiple spaces
        }
        
        # Intent patterns
        self._intent_patterns = {
            "greeting": r'^(?:hi|hello|hey|good\s+(?:morning|afternoon|evening))',
            "goodbye": r'(?:bye|goodbye|see you|talk later|have a good)',
            "help": r'(?:help|assist|support|problem|issue)',
            "transfer": r'(?:speak to|talk to|transfer|human|agent|person)',
            "appointment": r'(?:schedule|book|appointment|meeting|calendar)',
            "cancel": r'(?:cancel|nevermind|forget it|stop)',
        }
    
    def process(self, transcript: str) -> ProcessedTranscript:
        """
        Process a raw transcript.
        
        Args:
            transcript: Raw transcript from STT
        
        Returns:
            Processed transcript with metadata
        """
        # Normalize
        normalized = transcript.strip()
        for pattern, replacement in self._corrections.items():
            normalized = re.sub(pattern, replacement, normalized, flags=re.I)
        normalized = normalized.strip()
        
        # Detect intent
        detected_intent = None
        for intent, pattern in self._intent_patterns.items():
            if re.search(pattern, normalized, re.I):
                detected_intent = intent
                break
        
        # Check for question
        contains_question = bool(re.search(r'\?|^(?:what|where|when|who|why|how|is|are|do|does|can|could|would|will)\b', normalized, re.I))
        
        return ProcessedTranscript(
            raw_text=transcript,
            normalized_text=normalized,
            detected_intent=detected_intent,
            contains_question=contains_question,
            word_count=len(normalized.split()),
        )

# pipeline/stt/fallback.py

class STTWithFallback:
    """
    STT with automatic fallback to backup provider.
    """
    
    def __init__(
        self,
        primary: DeepgramSTT,
        fallback: "WhisperSTT",  # Local Whisper as fallback
    ):
        self.primary = primary
        self.fallback = fallback
        self._using_fallback = False
        self._primary_failures = 0
        self._max_failures = 3
    
    async def transcribe(self, audio: bytes) -> str:
        """Transcribe with automatic fallback."""
        if self._using_fallback:
            return await self._fallback_transcribe(audio)
        
        try:
            result = await self.primary.transcribe(audio)
            self._primary_failures = 0
            return result
        except Exception as e:
            logger.warning(f"Primary STT failed: {e}")
            self._primary_failures += 1
            
            if self._primary_failures >= self._max_failures:
                logger.warning("Switching to fallback STT")
                self._using_fallback = True
            
            return await self._fallback_transcribe(audio)
    
    async def _fallback_transcribe(self, audio: bytes) -> str:
        """Use fallback transcription."""
        return await self.fallback.transcribe(audio)
    
    async def reset_to_primary(self) -> None:
        """Attempt to switch back to primary."""
        self._using_fallback = False
        self._primary_failures = 0

┌─────────────────────────────────────────────────────────────────────────────┐
│                        CONTEXT ASSEMBLY PIPELINE                            │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   ┌──────────────┐     ┌──────────────┐     ┌──────────────┐               │
│   │    Agent     │     │  Knowledge   │     │    Tool      │               │
│   │   Config     │     │    Base      │     │ Definitions  │               │
│   └──────┬───────┘     └──────┬───────┘     └──────┬───────┘               │
│          │                    │                    │                        │
│          ▼                    ▼                    ▼                        │
│   ┌─────────────────────────────────────────────────────────────────┐      │
│   │                      SYSTEM PROMPT                               │      │
│   │                                                                  │      │
│   │  • Agent identity and personality                               │      │
│   │  • Business context and rules                                   │      │
│   │  • Knowledge base excerpts (RAG)                                │      │
│   │  • Available tools and when to use them                         │      │
│   │  • Response guidelines (concise, natural)                       │      │
│   │                                                                  │      │
│   └─────────────────────────────────────────────────────────────────┘      │
│                                    │                                        │
│                                    ▼                                        │
│   ┌──────────────┐     ┌─────────────────────────────────────────────┐     │
│   │ Conversation │     │              MESSAGE HISTORY                 │     │
│   │   History    │────▶│                                              │     │
│   │   (Redis)    │     │  [user]: "Hi, I'd like to schedule..."      │     │
│   └──────────────┘     │  [assistant]: "Of course! What day..."      │     │
│                        │  [user]: "How about Tuesday?"                │     │
│                        │  [assistant]: "Tuesday works..."             │     │
│                        │                                              │     │
│                        └─────────────────────────────────────────────┘     │
│                                    │                                        │
│                                    ▼                                        │
│   ┌──────────────┐     ┌─────────────────────────────────────────────┐     │
│   │   Current    │     │              CURRENT INPUT                   │     │
│   │  Transcript  │────▶│                                              │     │
│   │              │     │  [user]: "Actually, can we do 3pm?"          │     │
│   └──────────────┘     │                                              │     │
│                        └─────────────────────────────────────────────┘     │
│                                    │                                        │
│                                    ▼                                        │
│                        ┌─────────────────────────────────────────────┐     │
│                        │           ASSEMBLED CONTEXT                  │     │
│                        │                                              │     │
│                        │  Ready for LLM API call                     │     │
│                        │  Token count: ~2,000-4,000                  │     │
│                        │                                              │     │
│                        └─────────────────────────────────────────────┘     │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

# pipeline/context/context_manager.py

from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class AgentConfig:
    """Agent configuration from database."""
    agent_id: str
    name: str
    personality: str
    business_context: str
    greeting: str
    voice_id: str
    tools: list[dict]
    knowledge_base_id: Optional[str]
    max_turns: int = 50
    response_style: str = "concise"  # concise, detailed, friendly

@dataclass
class AssembledContext:
    """Fully assembled context ready for LLM."""
    system_prompt: str
    messages: list[dict]
    tools: list[dict]
    token_estimate: int

class ContextManager:
    """
    Manages conversation context assembly for LLM calls.
    """
    
    MAX_HISTORY_TURNS = 10
    MAX_CONTEXT_TOKENS = 8000  # Leave room for response
    
    def __init__(
        self,
        redis_client,
        knowledge_base_service,
    ):
        self.redis = redis_client
        self.kb_service = knowledge_base_service
    
    async def assemble_context(
        self,
        call_id: str,
        agent_config: AgentConfig,
        current_input: str,
    ) -> AssembledContext:
        """
        Assemble complete context for LLM call.
        
        Args:
            call_id: Current call ID
            agent_config: Agent configuration
            current_input: Current user transcript
        
        Returns:
            Assembled context ready for API call
        """
        # Build system prompt
        system_prompt = await self._build_system_prompt(
            agent_config,
            current_input,
        )
        
        # Get conversation history
        history = await self._get_conversation_history(call_id)
        
        # Build messages array
        messages = []
        for turn in history[-self.MAX_HISTORY_TURNS:]:
            messages.append({
                "role": turn["role"],
                "content": turn["content"],
            })
        
        # Add current input
        messages.append({
            "role": "user",
            "content": current_input,
        })
        
        # Format tools for Claude
        tools = self._format_tools(agent_config.tools)
        
        # Estimate tokens
        token_estimate = self._estimate_tokens(
            system_prompt,
            messages,
            tools,
        )
        
        # Trim history if needed
        while token_estimate > self.MAX_CONTEXT_TOKENS and len(messages) > 2:
            messages.pop(0)
            token_estimate = self._estimate_tokens(
                system_prompt,
                messages,
                tools,
            )
        
        return AssembledContext(
            system_prompt=system_prompt,
            messages=messages,
            tools=tools,
            token_estimate=token_estimate,
        )
    
    async def _build_system_prompt(
        self,
        config: AgentConfig,
        current_input: str,
    ) -> str:
        """Build the system prompt with all context."""
        
        # Base system prompt
        prompt_parts = [
            f"You are {config.name}, an AI voice assistant.",
            f"\n\n## Personality\n{config.personality}",
            f"\n\n## Business Context\n{config.business_context}",
        ]
        
        # Add knowledge base context if available
        if config.knowledge_base_id:
            kb_context = await self._get_knowledge_context(
                config.knowledge_base_id,
                current_input,
            )
            if kb_context:
                prompt_parts.append(
                    f"\n\n## Relevant Information\n{kb_context}"
                )
        
        # Add response guidelines
        style_guidelines = {
            "concise": """
## Response Guidelines
- Keep responses brief and conversational (1-2 sentences when possible)
- Speak naturally as if on a phone call
- Avoid lists or formatted text - this is a voice conversation
- Ask one question at a time
- Confirm understanding before taking actions
""",
            "detailed": """
## Response Guidelines
- Provide thorough explanations when needed
- Still maintain a conversational tone
- Break complex information into digestible parts
- Offer to elaborate on any points
""",
            "friendly": """
## Response Guidelines  
- Be warm, personable, and empathetic
- Use casual language appropriate for voice
- Show genuine interest in helping
- Add appropriate conversational acknowledgments
""",
        }
        
        prompt_parts.append(
            style_guidelines.get(config.response_style, style_guidelines["concise"])
        )
        
        # Add tool usage instructions if tools are available
        if config.tools:
            prompt_parts.append("""
## Tool Usage
- Use tools when you need to perform actions or look up information
- Always confirm with the user before executing consequential actions
- If a tool call fails, explain the issue and offer alternatives
""")
        
        return "\n".join(prompt_parts)
    
    async def _get_knowledge_context(
        self,
        knowledge_base_id: str,
        query: str,
    ) -> Optional[str]:
        """Retrieve relevant context from knowledge base."""
        try:
            results = await self.kb_service.search(
                knowledge_base_id=knowledge_base_id,
                query=query,
                top_k=3,
            )
            
            if not results:
                return None
            
            context_parts = []
            for result in results:
                context_parts.append(result["content"])
            
            return "\n\n".join(context_parts)
            
        except Exception as e:
            logger.warning(f"Knowledge base lookup failed: {e}")
            return None
    
    async def _get_conversation_history(
        self,
        call_id: str,
    ) -> list[dict]:
        """Get conversation history from Redis."""
        history_key = f"call:{call_id}:history"
        history_json = await self.redis.get(history_key)
        
        if history_json:
            return json.loads(history_json)
        return []
    
    async def add_to_history(
        self,
        call_id: str,
        role: str,
        content: str,
    ) -> None:
        """Add a turn to conversation history."""
        history_key = f"call:{call_id}:history"
        
        history = await self._get_conversation_history(call_id)
        history.append({
            "role": role,
            "content": content,
            "timestamp": time.time(),
        })
        
        # Keep last N turns
        history = history[-self.MAX_HISTORY_TURNS * 2:]
        
        await self.redis.set(
            history_key,
            json.dumps(history),
            ex=3600,  # 1 hour TTL
        )
    
    def _format_tools(self, tools: list[dict]) -> list[dict]:
        """Format tools for Claude API."""
        formatted = []
        
        for tool in tools:
            formatted.append({
                "name": tool["name"],
                "description": tool["description"],
                "input_schema": tool.get("parameters", {}),
            })
        
        return formatted
    
    def _estimate_tokens(
        self,
        system_prompt: str,
        messages: list[dict],
        tools: list[dict],
    ) -> int:
        """Rough token estimation (4 chars per token)."""
        total_chars = len(system_prompt)
        
        for msg in messages:
            total_chars += len(msg.get("content", ""))
        
        for tool in tools:
            total_chars += len(json.dumps(tool))
        
        return total_chars // 4

# pipeline/context/prompts.py

VOICE_AGENT_BASE_PROMPT = """You are {agent_name}, an AI voice assistant for {company_name}.

## Your Role
{role_description}

## Communication Style
- Speak naturally and conversationally - you're on a phone call
- Keep responses concise (1-3 sentences typically)
- Use contractions and casual language
- Never use markdown, bullet points, or formatting
- Don't say "I'm an AI" unless directly asked
- Acknowledge what the caller says before responding

## Current Context
- Caller phone number: {caller_phone}
- Time: {current_time}
- Previous calls with this number: {call_history_summary}

{additional_context}
"""

TOOL_USAGE_PROMPT = """
## Available Actions
You can perform these actions using the tools provided:
{tool_descriptions}

When using tools:
1. Only use tools when necessary to help the caller
2. Confirm consequential actions before executing
3. If a tool fails, apologize and offer alternatives
4. Don't mention "tools" or "functions" to the caller
"""

ESCALATION_PROMPT = """
## When to Transfer
Transfer to a human agent if:
- The caller explicitly requests to speak with a person
- You cannot resolve their issue after 2-3 attempts
- The situation involves sensitive matters you cannot handle
- The caller becomes frustrated or upset

To transfer, use the transfer_to_agent tool.
"""

# pipeline/llm/claude_client.py

import anthropic
from typing import AsyncIterator, Optional
from dataclasses import dataclass
import json

@dataclass
class LLMConfig:
    """Configuration for Claude LLM."""
    api_key: str
    model: str = "claude-sonnet-4-20250514"
    max_tokens: int = 1024
    temperature: float = 0.7
    timeout: float = 30.0

@dataclass
class LLMResponse:
    """Streamed response from LLM."""
    text: str
    is_complete: bool
    tool_calls: list[dict]
    stop_reason: Optional[str]
    usage: Optional[dict]

class ClaudeLLM:
    """
    Claude LLM client with streaming support.
    
    Handles streaming responses and tool calling
    in voice conversation context.
    """
    
    def __init__(self, config: LLMConfig):
        self.config = config
        self.client = anthropic.AsyncAnthropic(
            api_key=config.api_key,
            timeout=config.timeout,
        )
    
    async def generate_stream(
        self,
        system_prompt: str,
        messages: list[dict],
        tools: list[dict] = None,
    ) -> AsyncIterator[LLMResponse]:
        """
        Generate a streaming response.
        
        Args:
            system_prompt: System prompt with context
            messages: Conversation history + current input
            tools: Available tools for the agent
        
        Yields:
            LLMResponse objects with incremental text
        """
        request_params = {
            "model": self.config.model,
            "max_tokens": self.config.max_tokens,
            "temperature": self.config.temperature,
            "system": system_prompt,
            "messages": messages,
        }
        
        if tools:
            request_params["tools"] = tools
        
        accumulated_text = ""
        tool_calls = []
        
        async with self.client.messages.stream(**request_params) as stream:
            async for event in stream:
                if event.type == "content_block_delta":
                    if hasattr(event.delta, "text"):
                        accumulated_text += event.delta.text
                        
                        yield LLMResponse(
                            text=accumulated_text,
                            is_complete=False,
                            tool_calls=[],
                            stop_reason=None,
                            usage=None,
                        )
                    
                    elif hasattr(event.delta, "partial_json"):
                        # Tool call being streamed
                        pass
                
                elif event.type == "content_block_stop":
                    pass
                
                elif event.type == "message_delta":
                    stop_reason = event.delta.stop_reason
                    usage = {
                        "output_tokens": event.usage.output_tokens,
                    }
            
            # Get final message for tool calls
            final_message = await stream.get_final_message()
            
            for block in final_message.content:
                if block.type == "tool_use":
                    tool_calls.append({
                        "id": block.id,
                        "name": block.name,
                        "input": block.input,
                    })
            
            yield LLMResponse(
                text=accumulated_text,
                is_complete=True,
                tool_calls=tool_calls,
                stop_reason=final_message.stop_reason,
                usage={
                    "input_tokens": final_message.usage.input_tokens,
                    "output_tokens": final_message.usage.output_tokens,
                },
            )
    
    async def generate_with_tools(
        self,
        system_prompt: str,
        messages: list[dict],
        tools: list[dict],
        tool_results: list[dict] = None,
    ) -> AsyncIterator[LLMResponse]:
        """
        Generate response with tool calling support.
        
        If tool_results are provided, continues conversation
        after tool execution.
        """
        # If we have tool results, add them to messages
        if tool_results:
            for result in tool_results:
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": result["tool_use_id"],
                            "content": result["content"],
                        }
                    ],
                })
        
        async for response in self.generate_stream(
            system_prompt=system_prompt,
            messages=messages,
            tools=tools,
        ):
            yield response


class LLMResponseProcessor:
    """
    Process streaming LLM responses for voice output.
    
    Buffers text until natural speech boundaries
    and handles tool calls appropriately.
    """
    
    def __init__(self):
        self._buffer = ""
        self._sentence_endings = re.compile(r'[.!?]\s+')
        self._clause_endings = re.compile(r'[,;:]\s+')
    
    async def process_stream(
        self,
        response_stream: AsyncIterator[LLMResponse],
        on_sentence: callable,
        on_tool_call: callable,
    ) -> None:
        """
        Process a streaming response.
        
        Args:
            response_stream: Stream of LLM responses
            on_sentence: Called with each complete sentence
            on_tool_call: Called when tool calls are detected
        """
        full_text = ""
        last_sent_position = 0
        
        async for response in response_stream:
            full_text = response.text
            
            # Find sentence boundaries in new text
            new_text = full_text[last_sent_position:]
            
            # Look for sentence endings
            matches = list(self._sentence_endings.finditer(new_text))
            
            for match in matches:
                sentence_end = last_sent_position + match.end()
                sentence = full_text[last_sent_position:sentence_end].strip()
                
                if sentence:
                    await on_sentence(sentence)
                
                last_sent_position = sentence_end
            
            # Handle tool calls when response is complete
            if response.is_complete:
                # Send any remaining text
                remaining = full_text[last_sent_position:].strip()
                if remaining:
                    await on_sentence(remaining)
                
                # Process tool calls
                for tool_call in response.tool_calls:
                    await on_tool_call(tool_call)
    
    def extract_speakable_chunk(
        self,
        text: str,
        min_length: int = 10,
    ) -> tuple[str, str]:
        """
        Extract a chunk suitable for TTS.
        
        Returns:
            Tuple of (chunk_to_speak, remaining_text)
        """
        if len(text) < min_length:
            return "", text
        
        # Try to find sentence boundary
        match = self._sentence_endings.search(text)
        if match:
            return text[:match.end()].strip(), text[match.end():]
        
        # Try clause boundary if text is long enough
        if len(text) > 50:
            match = self._clause_endings.search(text)
            if match and match.start() > min_length:
                return text[:match.end()].strip(), text[match.end():]
        
        return "", text

# pipeline/llm/response_router.py

from enum import Enum
from dataclasses import dataclass
from typing import Union

class ResponseType(Enum):
    SPEECH = "speech"
    TOOL_CALL = "tool_call"
    MIXED = "mixed"  # Contains both speech and tool calls

@dataclass
class RoutedResponse:
    """Response after routing decision."""
    response_type: ResponseType
    speech_text: Optional[str]
    tool_calls: list[dict]
    should_wait_for_tool: bool

class ResponseRouter:
    """
    Route LLM responses to appropriate handlers.
    """
    
    def route(self, response: LLMResponse) -> RoutedResponse:
        """
        Determine how to handle the response.
        
        Args:
            response: Complete LLM response
        
        Returns:
            Routing decision with extracted components
        """
        has_text = bool(response.text.strip())
        has_tools = bool(response.tool_calls)
        
        if has_tools and not has_text:
            return RoutedResponse(
                response_type=ResponseType.TOOL_CALL,
                speech_text=None,
                tool_calls=response.tool_calls,
                should_wait_for_tool=True,
            )
        
        elif has_text and not has_tools:
            return RoutedResponse(
                response_type=ResponseType.SPEECH,
                speech_text=response.text,
                tool_calls=[],
                should_wait_for_tool=False,
            )
        
        elif has_text and has_tools:
            # Text before tool call - speak first, then execute
            return RoutedResponse(
                response_type=ResponseType.MIXED,
                speech_text=response.text,
                tool_calls=response.tool_calls,
                should_wait_for_tool=True,
            )
        
        else:
            # Empty response
            return RoutedResponse(
                response_type=ResponseType.SPEECH,
                speech_text="I'm sorry, I didn't catch that. Could you please repeat?",
                tool_calls=[],
                should_wait_for_tool=False,
            )

# pipeline/llm/fallback.py

class LLMWithFallback:
    """
    LLM client with automatic fallback.
    
    Falls back to Haiku for faster responses if Sonnet is slow,
    or to cached responses if API is unavailable.
    """
    
    def __init__(
        self,
        primary: ClaudeLLM,       # Sonnet
        fast_fallback: ClaudeLLM,  # Haiku
        cache: ResponseCache,
    ):
        self.primary = primary
        self.fast_fallback = fast_fallback
        self.cache = cache
        
        self._primary_timeout = 5.0  # Switch to fallback if TTFB > 5s
        self._failure_count = 0
        self._max_failures = 3
    
    async def generate_stream(
        self,
        system_prompt: str,
        messages: list[dict],
        tools: list[dict] = None,
    ) -> AsyncIterator[LLMResponse]:
        """Generate with automatic fallback."""
        
        # Check if we should use fallback due to recent failures
        if self._failure_count >= self._max_failures:
            async for response in self._use_fast_fallback(
                system_prompt, messages, tools
            ):
                yield response
            return
        
        # Try primary with timeout monitoring
        try:
            first_token_received = False
            start_time = time.time()
            
            async for response in self.primary.generate_stream(
                system_prompt, messages, tools
            ):
                if not first_token_received:
                    first_token_received = True
                    ttfb = time.time() - start_time
                    
                    # If TTFB is too slow, switch to fallback for next request
                    if ttfb > self._primary_timeout:
                        logger.warning(f"Primary LLM slow ({ttfb:.1f}s), consider fallback")
                
                yield response
            
            # Success - reset failure count
            self._failure_count = 0
            
        except Exception as e:
            logger.error(f"Primary LLM failed: {e}")
            self._failure_count += 1
            
            # Try fast fallback
            async for response in self._use_fast_fallback(
                system_prompt, messages, tools
            ):
                yield response
    
    async def _use_fast_fallback(
        self,
        system_prompt: str,
        messages: list[dict],
        tools: list[dict] = None,
    ) -> AsyncIterator[LLMResponse]:
        """Use the fast fallback model."""
        try:
            async for response in self.fast_fallback.generate_stream(
                system_prompt, messages, tools
            ):
                yield response
        except Exception as e:
            logger.error(f"Fallback LLM also failed: {e}")
            
            # Last resort - cached response
            cached = await self.cache.get_fallback_response(messages)
            if cached:
                yield LLMResponse(
                    text=cached,
                    is_complete=True,
                    tool_calls=[],
                    stop_reason="cache",
                    usage=None,
                )
            else:
                yield LLMResponse(
                    text="I apologize, I'm having technical difficulties. Please try again in a moment.",
                    is_complete=True,
                    tool_calls=[],
                    stop_reason="error",
                    usage=None,
                )

# pipeline/tts/chatterbox_client.py

import httpx
import numpy as np
from typing import AsyncIterator, Optional
from dataclasses import dataclass

@dataclass
class TTSConfig:
    """Configuration for Chatterbox TTS."""
    api_url: str  # RunPod endpoint
    api_key: str
    voice_id: str = "default"
    sample_rate: int = 24000
    speed: float = 1.0
    pitch: float = 1.0
    exaggeration: float = 0.5  # Emotion intensity
    cfg_weight: float = 0.5   # Adherence to reference
    timeout: float = 30.0

@dataclass
class AudioChunk:
    """Chunk of synthesized audio."""
    audio_data: bytes
    sample_rate: int
    duration_ms: float
    is_final: bool

class ChatterboxTTS:
    """
    Chatterbox TTS client with streaming support.
    
    Connects to self-hosted Chatterbox on RunPod
    for high-quality, cost-effective speech synthesis.
    """
    
    def __init__(self, config: TTSConfig):
        self.config = config
        self._client = httpx.AsyncClient(timeout=config.timeout)
    
    async def synthesize_stream(
        self,
        text: str,
        voice_id: str = None,
    ) -> AsyncIterator[AudioChunk]:
        """
        Synthesize text to speech with streaming output.
        
        Args:
            text: Text to synthesize
            voice_id: Voice to use (overrides config)
        
        Yields:
            AudioChunk objects with PCM audio data
        """
        voice = voice_id or self.config.voice_id
        
        request_data = {
            "text": text,
            "voice_id": voice,
            "sample_rate": self.config.sample_rate,
            "speed": self.config.speed,
            "pitch": self.config.pitch,
            "exaggeration": self.config.exaggeration,
            "cfg_weight": self.config.cfg_weight,
            "stream": True,
        }
        
        async with self._client.stream(
            "POST",
            f"{self.config.api_url}/synthesize",
            json=request_data,
            headers={
                "Authorization": f"Bearer {self.config.api_key}",
                "Accept": "audio/pcm",
            },
        ) as response:
            response.raise_for_status()
            
            chunk_index = 0
            total_samples = 0
            
            async for chunk in response.aiter_bytes(chunk_size=4800):  # 100ms chunks
                if chunk:
                    samples = len(chunk) // 2  # 16-bit audio
                    duration_ms = (samples / self.config.sample_rate) * 1000
                    total_samples += samples
                    
                    yield AudioChunk(
                        audio_data=chunk,
                        sample_rate=self.config.sample_rate,
                        duration_ms=duration_ms,
                        is_final=False,
                    )
                    
                    chunk_index += 1
            
            # Final chunk marker
            yield AudioChunk(
                audio_data=b"",
                sample_rate=self.config.sample_rate,
                duration_ms=0,
                is_final=True,
            )
    
    async def synthesize(
        self,
        text: str,
        voice_id: str = None,
    ) -> bytes:
        """
        Synthesize text to speech (non-streaming).
        
        Returns complete audio as bytes.
        """
        chunks = []
        async for chunk in self.synthesize_stream(text, voice_id):
            if chunk.audio_data:
                chunks.append(chunk.audio_data)
        
        return b"".join(chunks)
    
    async def get_voices(self) -> list[dict]:
        """Get available voices."""
        response = await self._client.get(
            f"{self.config.api_url}/voices",
            headers={"Authorization": f"Bearer {self.config.api_key}"},
        )
        response.raise_for_status()
        return response.json()
    
    async def clone_voice(
        self,
        name: str,
        audio_samples: list[bytes],
    ) -> str:
        """
        Clone a voice from audio samples.
        
        Returns the new voice ID.
        """
        # Implementation depends on Chatterbox voice cloning API
        pass
    
    async def close(self) -> None:
        """Close the client."""
        await self._client.aclose()


class TTSManager:
    """
    High-level TTS manager with sentence queuing.
    """
    
    def __init__(self, config: TTSConfig):
        self.config = config
        self._client = ChatterboxTTS(config)
        self._sentence_queue: asyncio.Queue = asyncio.Queue()
        self._audio_queue: asyncio.Queue = asyncio.Queue()
        self._running = False
    
    async def start(self) -> None:
        """Start the TTS processing loop."""
        self._running = True
        asyncio.create_task(self._process_queue())
    
    async def stop(self) -> None:
        """Stop the TTS processing loop."""
        self._running = False
    
    async def queue_sentence(self, text: str) -> None:
        """Add a sentence to the synthesis queue."""
        await self._sentence_queue.put(text)
    
    async def get_audio(self) -> Optional[AudioChunk]:
        """Get the next audio chunk."""
        try:
            return await asyncio.wait_for(
                self._audio_queue.get(),
                timeout=0.1,
            )
        except asyncio.TimeoutError:
            return None
    
    async def clear_queue(self) -> None:
        """Clear pending sentences (for interruption)."""
        while not self._sentence_queue.empty():
            try:
                self._sentence_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
        
        while not self._audio_queue.empty():
            try:
                self._audio_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
    
    async def _process_queue(self) -> None:
        """Process sentences from the queue."""
        while self._running:
            try:
                text = await asyncio.wait_for(
                    self._sentence_queue.get(),
                    timeout=0.5,
                )
                
                async for chunk in self._client.synthesize_stream(text):
                    await self._audio_queue.put(chunk)
                    
            except asyncio.TimeoutError:
                continue
            except Exception as e:
                logger.error(f"TTS error: {e}")