┌─────────────────────────────────────────────────────────────────────────────┐
│                    PRODUCTION REAL-TIME VOICE AGENT                         │
├─────────────────────────────────────────────────────────────────────────────┤
│ Caller                                                                      │
│   │                                                                         │
│   ▼                                                                         │
│ SIP/WebRTC Gateway                                                          │
│   │                                                                         │
│   ▼                                                                         │
│ Streaming Orchestrator (WebSocket)                                          │
│   ├──► ASR Stream (partial + final transcripts)                             │
│   ├──► Dialogue Manager (state, policy, tools, fallbacks)                   │
│   └──► TTS Stream (chunked audio output)                                    │
│                         │                                                   │
│                         ▼                                                   │
│                     Tool Router                                             │
│              (CRM, Orders API, KB Retriever)                                │
│                         │                                                   │
│                         ▼                                                   │
│                 Policy and Guardrails                                       │
│         (PII masking, rate limits, escalation rules)                        │
│                                                                             │
│ Observability                                                               │
│   • Langfuse: traces, scores, prompt versions                               │
│   • Prometheus: latency/error metrics                                       │
│   • Logs: transcript + tool audit + call events                             │
└─────────────────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────────┐
│                        SINGLE TURN LIFECYCLE                         │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  Caller Audio ──► WebSocket ──► ASR Stream ──► Final Transcript      │
│  (PCM 16kHz)       (recv)       (Deepgram)       │                   │
│                                                   ▼                  │
│                                           ┌──────────────┐           │
│                                           │   Dialogue    │           │
│                                           │   Manager     │           │
│                                           │  ┌─────────┐  │           │
│                                           │  │ Policy  │  │           │
│                                           │  │ Check   │  │           │
│                                           │  └────┬────┘  │           │
│                                           │       ▼       │           │
│                                           │  ┌─────────┐  │           │
│                                           │  │ LLM +   │  │           │
│                                           │  │ Tools   │  │           │
│                                           │  └────┬────┘  │           │
│                                           └───────┼──────┘           │
│                                                   ▼                  │
│  Caller Speaker ◄── WebSocket ◄── TTS Stream ◄── Agent Text         │
│  (PCM chunks)        (send)       (OpenAI)                           │
│                                                                      │
│  ── Barge-in detected? ──► Cancel TTS ──► New ASR turn ──►          │
│                                                                      │
│  Latency budget:                                                     │
│    ASR: ~300ms │ LLM: ~800ms │ TTS first chunk: ~400ms              │
│    Total P95 target: <2500ms                                     │
└──────────────────────────────────────────────────────────────────────┘

voice-agent-platform/
├── src/
│   ├── config.py                  # Pydantic Settings + env vars
│   ├── models.py                  # Shared data models and enums
│   ├── app.py                     # FastAPI bootstrap + WebSocket endpoint
│   ├── streaming_gateway.py       # WebSocket call session handler
│   ├── asr_client.py              # ASR provider adapter (Deepgram)
│   ├── tts_client.py              # TTS provider adapter (OpenAI)
│   ├── dialogue_manager.py        # Stateful turn orchestration + LLM
│   ├── barge_in.py                # Interruption detection and handling
│   ├── tool_router.py             # Tool calls + retries + circuit breaker
│   ├── rag_pipeline.py            # Optional KB retrieval (ChromaDB)
│   ├── observability.py           # Langfuse + Prometheus instrumentation
│   ├── policy.py                  # Compliance, escalation, PII masking
│   └── evaluation/
│       ├── offline_eval.py        # Scenario and regression evaluation
│       ├── stress_test.py         # Concurrency, jitter, timeout tests
│       └── ragas_eval.py          # Optional RAGAS scoring
├── tests/
│   ├── test_barge_in.py
│   ├── test_tool_fallbacks.py
│   ├── test_dialogue_state.py
│   └── test_latency_budget.py
├── .env
├── Dockerfile
├── docker-compose.yml
├── prometheus.yml
└── requirements.txt

fastapi==0.115.0
uvicorn==0.32.0
websockets==13.1
openai==1.52.0
deepgram-sdk==3.7.0
chromadb==0.5.0
sentence-transformers==3.3.0
langfuse==2.51.0
prometheus-client==0.21.0
pydantic-settings==2.6.0
httpx==0.27.0
ragas==0.2.0
pandas==2.2.0
pytest==8.3.0
pytest-asyncio==0.24.0

# LLM
OPENAI_API_KEY=sk-your-key-here

# ASR (Deepgram)
DEEPGRAM_API_KEY=your-deepgram-key-here

# Observability
LANGFUSE_PUBLIC_KEY=pk-lf-your-key
LANGFUSE_SECRET_KEY=sk-lf-your-key
LANGFUSE_HOST=https://cloud.langfuse.com

# Feature Flags
ENABLE_RAG=true
ENABLE_TRACING=true

# Thresholds
CONFIDENCE_THRESHOLD=0.6
MAX_CONSECUTIVE_FAILURES=3
TURN_TIMEOUT_MS=5000

from functools import lru_cache

from pydantic_settings import BaseSettings
from pydantic import Field


class Settings(BaseSettings):
    """Voice agent configuration loaded from environment variables."""

    # LLM
    openai_api_key: str = Field(..., alias="OPENAI_API_KEY")
    llm_model: str = Field("gpt-4o", alias="LLM_MODEL")
    llm_temperature: float = Field(0.3, alias="LLM_TEMPERATURE")
    llm_max_tokens: int = Field(512, alias="LLM_MAX_TOKENS")

    # ASR
    deepgram_api_key: str = Field(..., alias="DEEPGRAM_API_KEY")
    asr_language: str = Field("en-US", alias="ASR_LANGUAGE")
    asr_model: str = Field("nova-2", alias="ASR_MODEL")

    # TTS
    tts_model: str = Field("tts-1", alias="TTS_MODEL")
    tts_voice: str = Field("alloy", alias="TTS_VOICE")
    tts_speed: float = Field(1.0, alias="TTS_SPEED")
    tts_response_format: str = Field("pcm", alias="TTS_RESPONSE_FORMAT")

    # Observability
    langfuse_public_key: str = Field("", alias="LANGFUSE_PUBLIC_KEY")
    langfuse_secret_key: str = Field("", alias="LANGFUSE_SECRET_KEY")
    langfuse_host: str = Field(
        "https://cloud.langfuse.com", alias="LANGFUSE_HOST"
    )
    enable_tracing: bool = Field(True, alias="ENABLE_TRACING")

    # RAG
    enable_rag: bool = Field(True, alias="ENABLE_RAG")
    chroma_collection: str = Field("voice_kb", alias="CHROMA_COLLECTION")
    embedding_model: str = Field(
        "all-MiniLM-L6-v2", alias="EMBEDDING_MODEL"
    )

    # Thresholds and Limits
    confidence_threshold: float = Field(0.6, alias="CONFIDENCE_THRESHOLD")
    max_consecutive_failures: int = Field(3, alias="MAX_CONSECUTIVE_FAILURES")
    turn_timeout_ms: int = Field(5000, alias="TURN_TIMEOUT_MS")
    max_turns_per_session: int = Field(50, alias="MAX_TURNS_PER_SESSION")
    session_ttl_seconds: int = Field(1800, alias="SESSION_TTL_SECONDS")

    # Latency Targets (used for monitoring alerts, not hard limits)
    target_first_token_ms: int = Field(900, alias="TARGET_FIRST_TOKEN_MS")
    target_turn_latency_ms: int = Field(2500, alias="TARGET_TURN_LATENCY_MS")

    # Circuit Breaker
    cb_failure_threshold: int = Field(5, alias="CB_FAILURE_THRESHOLD")
    cb_recovery_timeout: int = Field(30, alias="CB_RECOVERY_TIMEOUT")

    model_config = {"env_file": ".env", "extra": "ignore"}


@lru_cache
def get_settings() -> Settings:
    return Settings()

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class CallEvent(str, Enum):
    """Events tracked throughout a call session."""
    CALL_START = "call_start"
    TURN_START = "turn_start"
    ASR_PARTIAL = "asr_partial"
    ASR_FINAL = "asr_final"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    LLM_START = "llm_start"
    LLM_DONE = "llm_done"
    TTS_START = "tts_start"
    TTS_DONE = "tts_done"
    BARGE_IN = "barge_in"
    ESCALATION = "escalation"
    CALL_END = "call_end"


@dataclass
class ToolResult:
    """Structured result from a tool invocation."""
    tool_name: str
    success: bool
    data: Optional[dict] = None
    error: Optional[str] = None
    latency_ms: int = 0


@dataclass
class PolicyDecision:
    """Decision from the policy engine for a given turn."""
    action: str  # "continue", "escalate", "block", "authenticate"
    reason: str = ""
    confidence: float = 1.0


@dataclass
class TurnState:
    """State for a single conversation turn."""
    turn_id: int
    user_text: str = ""
    agent_text: str = ""
    interrupted: bool = False
    tool_results: list[ToolResult] = field(default_factory=list)
    latency_ms: int = 0
    asr_latency_ms: int = 0
    llm_latency_ms: int = 0
    tts_latency_ms: int = 0
    events: list[CallEvent] = field(default_factory=list)


@dataclass
class SessionState:
    """Full state for a voice call session."""
    session_id: str
    caller_id: str
    locale: str = "en-US"
    authenticated: bool = False
    turns: list[TurnState] = field(default_factory=list)
    consecutive_failures: int = 0
    escalation_required: bool = False
    context: dict = field(default_factory=dict)
    messages: list[dict] = field(default_factory=list)  # LLM conversation history

import asyncio
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Optional

from deepgram import (
    DeepgramClient,
    DeepgramClientOptions,
    LiveOptions,
    LiveTranscriptionEvents,
)

from config import get_settings

logger = logging.getLogger(__name__)


@dataclass
class Transcript:
    """A single ASR transcript result."""
    text: str
    is_final: bool
    confidence: float = 0.0
    speech_final: bool = False  # end of utterance


class ASRClient(ABC):
    """Abstract base for speech-to-text providers."""

    @abstractmethod
    async def start(self, locale: str, hints: list[str] | None = None) -> None:
        ...

    @abstractmethod
    async def send_audio(self, chunk: bytes) -> None:
        ...

    @abstractmethod
    async def transcripts(self) -> AsyncIterator[Transcript]:
        ...

    @abstractmethod
    async def close(self) -> None:
        ...


class DeepgramASRClient(ASRClient):
    """Streaming ASR using Deepgram Nova-2 via WebSocket."""

    def __init__(self):
        settings = get_settings()
        config = DeepgramClientOptions(api_key=settings.deepgram_api_key)
        self._client = DeepgramClient(config=config)
        self._connection = None
        self._transcript_queue: asyncio.Queue[Transcript] = asyncio.Queue()
        self._closed = False

    async def start(self, locale: str, hints: list[str] | None = None) -> None:
        self._connection = self._client.listen.asyncwebsocket.v("1")

        # Register event handlers
        self._connection.on(
            LiveTranscriptionEvents.Transcript, self._on_transcript
        )
        self._connection.on(
            LiveTranscriptionEvents.Error, self._on_error
        )

        options = LiveOptions(
            model=get_settings().asr_model,
            language=locale[:2],  # "en-US" -> "en"
            encoding="linear16",
            sample_rate=16000,
            channels=1,
            interim_results=True,
            utterance_end_ms=1200,
            vad_events=True,
            smart_format=True,
        )
        if hints:
            options.keywords = hints

        started = await self._connection.start(options)
        if not started:
            raise RuntimeError("Failed to start Deepgram connection")
        logger.info("ASR stream started for locale=%s", locale)

    async def _on_transcript(self, _client, result, **_kwargs):
        """Handle incoming transcript from Deepgram."""
        alt = result.channel.alternatives[0] if result.channel.alternatives else None
        if alt is None or not alt.transcript:
            return

        transcript = Transcript(
            text=alt.transcript,
            is_final=result.is_final,
            confidence=alt.confidence,
            speech_final=result.speech_final,
        )
        await self._transcript_queue.put(transcript)

    async def _on_error(self, _client, error, **_kwargs):
        logger.error("Deepgram ASR error: %s", error)

    async def send_audio(self, chunk: bytes) -> None:
        if self._connection and not self._closed:
            await self._connection.send(chunk)

    async def transcripts(self) -> AsyncIterator[Transcript]:
        while not self._closed:
            try:
                transcript = await asyncio.wait_for(
                    self._transcript_queue.get(), timeout=0.1
                )
                yield transcript
            except asyncio.TimeoutError:
                continue

    async def close(self) -> None:
        self._closed = True
        if self._connection:
            await self._connection.finish()
        logger.info("ASR stream closed")

┌──────────────────────────────────────────────────────────┐
│                  DEEPGRAM ASR FLOW                        │
├──────────────────────────────────────────────────────────┤
│                                                          │
│  Audio chunks (PCM 16kHz) ──► Deepgram WebSocket         │
│  ~20ms per chunk                    │                    │
│                                     ▼                    │
│                          ┌──────────────────┐            │
│                          │  Partial results  │            │
│                          │  "I need to..."   │            │
│                          │  (is_final=false)  │            │
│                          └────────┬─────────┘            │
│                                   ▼                      │
│                          ┌──────────────────┐            │
│                          │  Final result     │            │
│                          │  "I need to check │            │
│                          │   my order"       │            │
│                          │  (is_final=true)   │            │
│                          └────────┬─────────┘            │
│                                   ▼                      │
│                          ┌──────────────────┐            │
│                          │  Speech final     │            │
│                          │  (utterance end)  │            │
│                          │  ──► Trigger turn  │            │
│                          └──────────────────┘            │
│                                                          │
│  Key: partial results enable early display;              │
│  speech_final triggers the dialogue manager              │
└──────────────────────────────────────────────────────────┘

import asyncio
import logging
from abc import ABC, abstractmethod
from typing import AsyncIterator

import openai

from config import get_settings

logger = logging.getLogger(__name__)


class TTSClient(ABC):
    """Abstract base for text-to-speech providers."""

    @abstractmethod
    async def synthesize(
        self, text: str, voice: str, cancel_event: asyncio.Event | None = None
    ) -> AsyncIterator[bytes]:
        ...


class OpenAITTSClient(TTSClient):
    """Streaming TTS using OpenAI's audio API."""

    def __init__(self):
        settings = get_settings()
        self._client = openai.AsyncOpenAI(api_key=settings.openai_api_key)
        self._settings = settings

    async def synthesize(
        self, text: str, voice: str | None = None, cancel_event: asyncio.Event | None = None
    ) -> AsyncIterator[bytes]:
        """Stream audio chunks for the given text.

        Args:
            text: The text to convert to speech.
            voice: TTS voice name. Falls back to config default.
            cancel_event: Set this event to stop mid-stream (barge-in).

        Yields:
            Raw PCM audio bytes in 16-bit 24kHz format.
        """
        if not text.strip():
            return

        voice = voice or self._settings.tts_voice

        async with self._client.audio.speech.with_streaming_response.create(
            model=self._settings.tts_model,
            voice=voice,
            input=text,
            response_format=self._settings.tts_response_format,
            speed=self._settings.tts_speed,
        ) as response:
            async for chunk in response.iter_bytes(chunk_size=4096):
                # Check for barge-in cancellation before yielding
                if cancel_event and cancel_event.is_set():
                    logger.info("TTS cancelled mid-stream (barge-in)")
                    return
                yield chunk

        logger.debug("TTS synthesis complete: %d chars", len(text))


# Voice mapping per locale (used by streaming gateway)
LOCALE_VOICES = {
    "en-US": "alloy",
    "en-GB": "fable",
    "en-AU": "shimmer",
    "ar-SA": "nova",
    "es-ES": "onyx",
    "fr-FR": "echo",
}


def get_voice_for_locale(locale: str) -> str:
    """Return the best TTS voice for a locale, with fallback."""
    return LOCALE_VOICES.get(locale, get_settings().tts_voice)

Normal flow:          Barge-in flow:

Text ──► TTS API      Text ──► TTS API
          │                     │
    chunk 1 ──► WS        chunk 1 ──► WS
    chunk 2 ──► WS        chunk 2 ──► WS
    chunk 3 ──► WS        ── user speaks ──
    chunk 4 ──► WS        cancel_event.set()
    (done)                (TTS stops, no more chunks)
                          ──► new ASR turn starts

import logging
import re
from dataclasses import dataclass

from models import PolicyDecision, SessionState
from config import get_settings

logger = logging.getLogger(__name__)


# Dialect and locale configuration
DIALECT_CONFIG = {
    "en-US": {
        "asr_hints": ["refund", "order number", "tracking id", "cancel"],
        "tts_voice": "alloy",
    },
    "en-GB": {
        "asr_hints": ["postcode", "parcel", "returns policy"],
        "tts_voice": "fable",
    },
    "ar-SA": {
        "asr_hints": ["رقم الطلب", "الاسترجاع", "الشحنة"],
        "tts_voice": "nova",
    },
}

# PII patterns for masking
PII_PATTERNS = [
    (re.compile(r"\b\d{3}[-.]?\d{2}[-.]?\d{4}\b"), "[SSN_MASKED]"),           # SSN
    (re.compile(r"\b\d{13,19}\b"), "[CARD_MASKED]"),                            # Credit card
    (re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"), "[EMAIL_MASKED]"),
    (re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"), "[PHONE_MASKED]"),          # US phone
]

# Escalation trigger phrases
ESCALATION_PHRASES = [
    "speak to a human", "talk to someone", "real person",
    "transfer me", "manager", "supervisor", "complaint",
]


def mask_pii(text: str) -> str:
    """Replace PII patterns in text with masked placeholders."""
    masked = text
    for pattern, replacement in PII_PATTERNS:
        masked = pattern.sub(replacement, masked)
    return masked


class PolicyEngine:
    """Evaluate each turn against safety, compliance, and escalation rules."""

    def __init__(self):
        self._settings = get_settings()

    def decide(
        self, user_text: str, session: SessionState
    ) -> PolicyDecision:
        """Return a policy decision for the current turn.

        Decision priority:
        1. Authentication check (must complete before any tool access)
        2. Explicit escalation request from user
        3. Consecutive failure threshold
        4. Content safety
        5. Continue normally
        """
        text_lower = user_text.lower().strip()

        # 1. Authentication gate
        if not session.authenticated and self._requires_auth(text_lower):
            return PolicyDecision(
                action="authenticate",
                reason="Tool access requires caller authentication",
            )

        # 2. Explicit escalation request
        if any(phrase in text_lower for phrase in ESCALATION_PHRASES):
            return PolicyDecision(
                action="escalate",
                reason=f"User requested human: '{user_text[:50]}'",
            )

        # 3. Consecutive failure threshold
        if session.consecutive_failures >= self._settings.max_consecutive_failures:
            return PolicyDecision(
                action="escalate",
                reason=f"Consecutive failures: {session.consecutive_failures}",
            )

        # 4. Content safety (block abusive input)
        if self._is_unsafe(text_lower):
            return PolicyDecision(
                action="block",
                reason="Content policy violation",
            )

        # 5. Default: continue
        return PolicyDecision(action="continue")

    def _requires_auth(self, text: str) -> bool:
        """Check if the user's request needs authentication."""
        auth_triggers = ["order", "account", "payment", "refund", "cancel"]
        return any(trigger in text for trigger in auth_triggers)

    def _is_unsafe(self, text: str) -> bool:
        """Basic content safety check."""
        unsafe_patterns = [
            "ignore your instructions",
            "you are now",
            "pretend to be",
            "system prompt",
        ]
        return any(p in text for p in unsafe_patterns)

    def get_dialect_config(self, locale: str) -> dict:
        """Return ASR hints and TTS voice for a locale."""
        return DIALECT_CONFIG.get(locale, DIALECT_CONFIG["en-US"])

User Input
    │
    ▼
┌──────────────────────┐
│ Authenticated?       │──no──► requires_auth? ──yes──► "authenticate"
│                      │                       └──no──► continue below
└──────────┬───────────┘
           │ yes
           ▼
┌──────────────────────┐
│ Escalation phrase?   │──yes──► "escalate"
└──────────┬───────────┘
           │ no
           ▼
┌──────────────────────┐
│ Failures >= 3?       │──yes──► "escalate"
└──────────┬───────────┘
           │ no
           ▼
┌──────────────────────┐
│ Unsafe content?      │──yes──► "block"
└──────────┬───────────┘
           │ no
           ▼
       "continue"

import asyncio
import logging
import time
from dataclasses import dataclass, field
from enum import Enum

from models import ToolResult
from config import get_settings

logger = logging.getLogger(__name__)


class CircuitState(str, Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject calls immediately
    HALF_OPEN = "half_open"  # Testing recovery


@dataclass
class CircuitBreaker:
    """Track failure state for a single tool."""
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0.0
    failure_threshold: int = 5
    recovery_timeout: int = 30

    def record_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            logger.warning("Circuit breaker OPEN after %d failures", self.failure_count)

    def allow_request(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        # HALF_OPEN: allow one test request
        return True


class ToolRouter:
    """Route tool calls with retries, circuit breakers, and fallbacks."""

    def __init__(self, order_client=None, kb_client=None):
        self.order_client = order_client
        self.kb_client = kb_client
        settings = get_settings()
        self._breakers: dict[str, CircuitBreaker] = {}
        self._default_breaker_config = {
            "failure_threshold": settings.cb_failure_threshold,
            "recovery_timeout": settings.cb_recovery_timeout,
        }

    def _get_breaker(self, tool_name: str) -> CircuitBreaker:
        if tool_name not in self._breakers:
            self._breakers[tool_name] = CircuitBreaker(
                **self._default_breaker_config
            )
        return self._breakers[tool_name]

    async def _retry_async(self, fn, tool_name: str, retries: int = 2, delay: float = 0.3):
        """Retry an async function with circuit breaker protection."""
        breaker = self._get_breaker(tool_name)

        if not breaker.allow_request():
            raise RuntimeError(f"Circuit open for {tool_name}")

        last_error = None
        for attempt in range(retries + 1):
            try:
                result = await fn()
                breaker.record_success()
                return result
            except Exception as exc:
                last_error = exc
                logger.warning(
                    "Tool %s attempt %d failed: %s", tool_name, attempt + 1, exc
                )
                if attempt < retries:
                    await asyncio.sleep(delay * (attempt + 1))

        breaker.record_failure()
        raise RuntimeError(f"{tool_name} failed after {retries + 1} attempts: {last_error}")

    async def call_tool(self, tool_name: str, session, **kwargs) -> ToolResult:
        """Execute a tool call with full reliability wrapping."""
        start = time.time()

        try:
            if tool_name == "order_status":
                data = await self._retry_async(
                    lambda: self.order_client.get_status(session.caller_id),
                    tool_name="order_status",
                )
                return ToolResult(
                    tool_name="order_status",
                    success=True,
                    data=data,
                    latency_ms=int((time.time() - start) * 1000),
                )

            elif tool_name == "knowledge_base" and self.kb_client:
                query = kwargs.get("query", "")
                answer = await self._retry_async(
                    lambda: self.kb_client.answer(query),
                    tool_name="knowledge_base",
                )
                return ToolResult(
                    tool_name="knowledge_base",
                    success=True,
                    data={"answer": answer},
                    latency_ms=int((time.time() - start) * 1000),
                )

            else:
                return ToolResult(
                    tool_name=tool_name,
                    success=False,
                    error=f"Unknown tool: {tool_name}",
                )

        except RuntimeError as exc:
            latency = int((time.time() - start) * 1000)
            logger.error("Tool %s failed: %s", tool_name, exc)
            return ToolResult(
                tool_name=tool_name,
                success=False,
                error=str(exc),
                latency_ms=latency,
            )

    def get_fallback_response(self, tool_name: str) -> str:
        """Return a safe fallback message when a tool is unavailable."""
        fallbacks = {
            "order_status": "I cannot access order information right now. Let me connect you to a support agent who can help.",
            "knowledge_base": "I do not have enough information to answer that confidently. Would you like me to transfer you to a specialist?",
        }
        return fallbacks.get(tool_name, "I am having trouble with that request. Can I help with something else?")

┌─────────────────────────────────────────────────┐
│               CIRCUIT BREAKER STATES             │
├─────────────────────────────────────────────────┤
│                                                  │
│   CLOSED ──── failures >= 5 ────► OPEN           │
│     ▲                               │            │
│     │                        wait 30s            │
│   success                           │            │
│     │                               ▼            │
│     └──────────────────────── HALF_OPEN          │
│                              (test 1 request)    │
│                                │         │       │
│                            success    failure    │
│                                │         │       │
│                             CLOSED     OPEN      │
└─────────────────────────────────────────────────┘

import logging
from typing import Optional

import chromadb
from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

from config import get_settings

logger = logging.getLogger(__name__)


class KnowledgeBase:
    """ChromaDB-backed knowledge base with confidence scoring."""

    def __init__(self):
        settings = get_settings()
        self._client = chromadb.PersistentClient(path="./chroma_data")
        self._embedding_fn = SentenceTransformerEmbeddingFunction(
            model_name=settings.embedding_model
        )
        self._collection = self._client.get_or_create_collection(
            name=settings.chroma_collection,
            embedding_function=self._embedding_fn,
            metadata={"hnsw:space": "cosine"},
        )
        self._confidence_threshold = settings.confidence_threshold

    def load_documents(self, documents: list[dict]) -> int:
        """Load documents into the knowledge base.

        Args:
            documents: List of {"id": str, "text": str, "metadata": dict}

        Returns:
            Number of documents loaded.
        """
        if not documents:
            return 0

        self._collection.upsert(
            ids=[d["id"] for d in documents],
            documents=[d["text"] for d in documents],
            metadatas=[d.get("metadata", {}) for d in documents],
        )
        logger.info("Loaded %d documents into knowledge base", len(documents))
        return len(documents)

    async def answer(self, query: str, n_results: int = 3) -> dict:
        """Retrieve relevant documents and return an answer with confidence.

        Returns:
            {"answer": str, "confidence": float, "sources": list[str]}
        """
        results = self._collection.query(
            query_texts=[query],
            n_results=n_results,
        )

        if not results["documents"] or not results["documents"][0]:
            return {
                "answer": "",
                "confidence": 0.0,
                "sources": [],
            }

        documents = results["documents"][0]
        distances = results["distances"][0] if results["distances"] else []

        # Convert cosine distance to similarity score (0 to 1)
        similarities = [1 - d for d in distances] if distances else []
        avg_confidence = sum(similarities) / len(similarities) if similarities else 0.0

        # Combine retrieved chunks as context
        context = "\n\n".join(documents)

        if avg_confidence < self._confidence_threshold:
            return {
                "answer": "",
                "confidence": avg_confidence,
                "sources": documents,
            }

        return {
            "answer": context,
            "confidence": avg_confidence,
            "sources": documents,
        }

import logging
import time
from contextlib import asynccontextmanager
from typing import Optional

from langfuse import Langfuse
from prometheus_client import Counter, Gauge, Histogram

from config import get_settings

logger = logging.getLogger(__name__)

# ── Prometheus Metrics ─────────────────────────────────────────────────

TURN_LATENCY = Histogram(
    "voice_turn_latency_ms",
    "End-to-end turn latency in milliseconds",
    buckets=(200, 500, 800, 1200, 1800, 2500, 4000, 8000),
)
ASR_LATENCY = Histogram(
    "voice_asr_latency_ms",
    "ASR finalization latency",
    buckets=(100, 200, 400, 600, 1000, 2000),
)
LLM_LATENCY = Histogram(
    "voice_llm_latency_ms",
    "LLM response latency",
    buckets=(200, 400, 800, 1200, 2000, 4000),
)
TTS_LATENCY = Histogram(
    "voice_tts_first_chunk_ms",
    "TTS time to first audio chunk",
    buckets=(100, 200, 400, 600, 1000, 2000),
)
TOOL_FAILURES = Counter(
    "voice_tool_failures_total",
    "Count of tool execution failures",
    ["tool_name"],
)
ESCALATIONS = Counter(
    "voice_escalations_total",
    "Count of escalations to human agents",
    ["reason"],
)
BARGE_INS = Counter(
    "voice_barge_ins_total",
    "Count of user interruptions",
)
ACTIVE_SESSIONS = Gauge(
    "voice_active_sessions",
    "Number of currently active call sessions",
)
SESSION_SUCCESS = Counter(
    "voice_sessions_total",
    "Total sessions by outcome",
    ["outcome"],  # "completed", "escalated", "error"
)

# ── Langfuse Tracing ──────────────────────────────────────────────────

_langfuse: Optional[Langfuse] = None


def get_langfuse() -> Optional[Langfuse]:
    """Lazy-init Langfuse client if tracing is enabled."""
    global _langfuse
    settings = get_settings()
    if not settings.enable_tracing:
        return None
    if _langfuse is None:
        _langfuse = Langfuse(
            public_key=settings.langfuse_public_key,
            secret_key=settings.langfuse_secret_key,
            host=settings.langfuse_host,
        )
    return _langfuse


def start_session_trace(session_id: str, caller_id: str):
    """Create a Langfuse trace for a voice call session."""
    lf = get_langfuse()
    if lf is None:
        return None
    return lf.trace(
        name="voice_session",
        id=session_id,
        user_id=caller_id,
        metadata={"type": "voice_call"},
    )


@asynccontextmanager
async def trace_span(trace, name: str, metadata: dict | None = None):
    """Context manager that creates a Langfuse span and records duration."""
    start = time.time()
    span = None
    if trace:
        span = trace.span(name=name, metadata=metadata or {})
    try:
        yield span
    finally:
        duration_ms = int((time.time() - start) * 1000)
        if span:
            span.end(metadata={"duration_ms": duration_ms})


def flush_langfuse():
    """Flush pending Langfuse events. Call on shutdown."""
    lf = get_langfuse()
    if lf:
        lf.flush()

import asyncio
import logging
import struct

from models import TurnState, SessionState, CallEvent

logger = logging.getLogger(__name__)

# VAD thresholds for 16-bit PCM audio
DEFAULT_ENERGY_THRESHOLD = 500       # RMS energy above silence
MIN_SPEECH_FRAMES = 3                # Consecutive frames to confirm speech
FRAME_SIZE_SAMPLES = 320             # 20ms at 16kHz


def compute_rms_energy(audio_chunk: bytes) -> float:
    """Compute RMS energy of a PCM 16-bit audio chunk."""
    if len(audio_chunk) < 2:
        return 0.0
    n_samples = len(audio_chunk) // 2
    samples = struct.unpack(f"<{n_samples}h", audio_chunk[:n_samples * 2])
    if not samples:
        return 0.0
    mean_sq = sum(s * s for s in samples) / n_samples
    return mean_sq ** 0.5


class BargeInDetector:
    """Detect user speech during TTS playback and trigger cancellation."""

    def __init__(self, energy_threshold: float = DEFAULT_ENERGY_THRESHOLD):
        self.energy_threshold = energy_threshold
        self.consecutive_speech_frames = 0
        self.tts_active = False
        self._cancel_event: asyncio.Event | None = None

    def set_tts_active(self, active: bool, cancel_event: asyncio.Event | None = None):
        """Track whether TTS is currently playing."""
        self.tts_active = active
        self._cancel_event = cancel_event if active else None
        if not active:
            self.consecutive_speech_frames = 0

    def check_audio(self, audio_chunk: bytes) -> bool:
        """Check if the audio chunk contains speech during TTS playback.

        Returns True if barge-in is detected.
        """
        if not self.tts_active:
            return False

        energy = compute_rms_energy(audio_chunk)

        if energy > self.energy_threshold:
            self.consecutive_speech_frames += 1
        else:
            self.consecutive_speech_frames = 0

        if self.consecutive_speech_frames >= MIN_SPEECH_FRAMES:
            logger.info("Barge-in detected (energy=%.1f, frames=%d)",
                       energy, self.consecutive_speech_frames)
            self._trigger_cancellation()
            return True

        return False

    def _trigger_cancellation(self):
        """Signal TTS to stop."""
        if self._cancel_event and not self._cancel_event.is_set():
            self._cancel_event.set()
        self.consecutive_speech_frames = 0


def handle_barge_in(
    session: SessionState, active_turn: TurnState
) -> dict:
    """Process a confirmed barge-in event."""
    active_turn.interrupted = True
    active_turn.events.append(CallEvent.BARGE_IN)
    logger.info(
        "Turn %d interrupted by barge-in (session=%s)",
        active_turn.turn_id, session.session_id,
    )
    return {
        "action": "stop_tts",
        "reason": "user_interruption",
        "turn_id": active_turn.turn_id,
    }

┌─────────────────────────────────────────────────────────┐
│             BARGE-IN DETECTION TIMELINE                   │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  TTS playing:  ████████████████████                      │
│  User audio:   .....____|████|___....                    │
│                         ↑                                │
│                    energy > 500 RMS                       │
│                    for 3+ frames (60ms)                   │
│                         │                                │
│                         ▼                                │
│              cancel_event.set()                           │
│              TTS stops immediately                        │
│              New ASR turn begins                          │
│                                                          │
│  Why 3 frames? Single-frame spikes are often noise.      │
│  60ms is fast enough to feel instant to the caller.      │
└─────────────────────────────────────────────────────────┘

import json
import logging
import time
from typing import Optional

import openai

from config import get_settings
from models import (
    CallEvent,
    PolicyDecision,
    SessionState,
    ToolResult,
    TurnState,
)
from observability import (
    LLM_LATENCY,
    ESCALATIONS,
    TOOL_FAILURES,
    trace_span,
)
from policy import PolicyEngine, mask_pii
from tool_router import ToolRouter

logger = logging.getLogger(__name__)

SYSTEM_PROMPT = """You are a helpful customer support voice agent. Follow these rules:

1. Be concise. Callers are listening, not reading. Keep responses under 2 sentences.
2. When you need information from a tool, call the appropriate function.
3. If you are unsure, say so honestly and offer to transfer to a human.
4. Never make up order information. Only report data from tool results.
5. If the caller has not been authenticated, ask for their PIN before accessing account data.

Available tools:
- order_status: Look up order status by caller ID. Returns status and ETA.
- knowledge_base: Search the knowledge base for policy and product answers.
"""

TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "order_status",
            "description": "Look up the current status and ETA for a caller's order",
            "parameters": {
                "type": "object",
                "properties": {
                    "caller_id": {
                        "type": "string",
                        "description": "The authenticated caller ID",
                    }
                },
                "required": ["caller_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "knowledge_base",
            "description": "Search the knowledge base for policy, product, or FAQ answers",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The question to search for",
                    }
                },
                "required": ["query"],
            },
        },
    },
]


class DialogueManager:
    """Orchestrate each conversation turn with LLM reasoning and tool execution."""

    def __init__(self, tool_router: ToolRouter, policy_engine: PolicyEngine):
        self.tool_router = tool_router
        self.policy_engine = policy_engine
        settings = get_settings()
        self._client = openai.AsyncOpenAI(api_key=settings.openai_api_key)
        self._settings = settings

    async def process_turn(
        self, session: SessionState, user_text: str, trace=None
    ) -> TurnState:
        """Process one conversation turn.

        Flow: policy check -> LLM call -> tool execution (if needed) -> response
        """
        turn = TurnState(
            turn_id=len(session.turns) + 1,
            user_text=user_text,
        )
        turn.events.append(CallEvent.TURN_START)

        # ── Policy Check ──
        decision = self.policy_engine.decide(user_text, session)

        if decision.action == "escalate":
            session.escalation_required = True
            turn.agent_text = "Let me connect you to a specialist who can help with that."
            turn.events.append(CallEvent.ESCALATION)
            ESCALATIONS.labels(reason=decision.reason).inc()
            session.turns.append(turn)
            return turn

        if decision.action == "authenticate":
            turn.agent_text = (
                "I can help with that. For security, could you please "
                "provide your four-digit PIN?"
            )
            session.turns.append(turn)
            return turn

        if decision.action == "block":
            turn.agent_text = "I can only help with customer support questions. How can I assist you today?"
            session.turns.append(turn)
            return turn

        # ── Build Messages ──
        if not session.messages:
            session.messages.append({"role": "system", "content": SYSTEM_PROMPT})

        session.messages.append({"role": "user", "content": user_text})

        # ── LLM Call ──
        turn.events.append(CallEvent.LLM_START)
        llm_start = time.time()

        async with trace_span(trace, "llm_call"):
            response = await self._client.chat.completions.create(
                model=self._settings.llm_model,
                messages=session.messages,
                tools=TOOL_SCHEMAS,
                tool_choice="auto",
                temperature=self._settings.llm_temperature,
                max_tokens=self._settings.llm_max_tokens,
            )

        turn.llm_latency_ms = int((time.time() - llm_start) * 1000)
        LLM_LATENCY.observe(turn.llm_latency_ms)
        turn.events.append(CallEvent.LLM_DONE)

        message = response.choices[0].message

        # ── Tool Execution ──
        if message.tool_calls:
            tool_results_text = []

            for tool_call in message.tool_calls:
                fn_name = tool_call.function.name
                fn_args = json.loads(tool_call.function.arguments)
                turn.events.append(CallEvent.TOOL_CALL)

                async with trace_span(trace, f"tool_{fn_name}"):
                    result = await self.tool_router.call_tool(
                        fn_name, session, **fn_args
                    )

                turn.tool_results.append(result)
                turn.events.append(CallEvent.TOOL_RESULT)

                if result.success:
                    session.consecutive_failures = 0
                    tool_results_text.append(json.dumps(result.data))
                else:
                    session.consecutive_failures += 1
                    TOOL_FAILURES.labels(tool_name=fn_name).inc()
                    fallback = self.tool_router.get_fallback_response(fn_name)
                    tool_results_text.append(json.dumps({"error": fallback}))

                # Append tool result to conversation for LLM
                session.messages.append(message.model_dump())
                session.messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_results_text[-1],
                })

            # Second LLM call to produce natural language from tool results
            async with trace_span(trace, "llm_tool_response"):
                follow_up = await self._client.chat.completions.create(
                    model=self._settings.llm_model,
                    messages=session.messages,
                    temperature=self._settings.llm_temperature,
                    max_tokens=self._settings.llm_max_tokens,
                )

            turn.agent_text = follow_up.choices[0].message.content or ""
            session.messages.append({
                "role": "assistant",
                "content": turn.agent_text,
            })

        else:
            # Direct response, no tool needed
            turn.agent_text = message.content or "Could you say that again?"
            session.messages.append({
                "role": "assistant",
                "content": turn.agent_text,
            })

        # PII-mask the turn text for logging
        logger.info(
            "Turn %d: user=%s agent=%s",
            turn.turn_id,
            mask_pii(user_text[:80]),
            mask_pii(turn.agent_text[:80]),
        )

        session.turns.append(turn)
        return turn

    def handle_pin_input(self, session: SessionState, pin: str) -> str:
        """Verify caller PIN and update session authentication state."""
        # In production, verify against a secure backend
        if len(pin) == 4 and pin.isdigit():
            session.authenticated = True
            return "Thank you, your identity has been verified. How can I help you?"
        return "That PIN was not recognized. Could you try again?"

import asyncio
import logging
import time
import uuid
from typing import Optional

from fastapi import WebSocket, WebSocketDisconnect

from asr_client import DeepgramASRClient
from barge_in import BargeInDetector, handle_barge_in
from config import get_settings
from dialogue_manager import DialogueManager
from models import CallEvent, SessionState, TurnState
from observability import (
    ACTIVE_SESSIONS,
    ASR_LATENCY,
    BARGE_INS,
    SESSION_SUCCESS,
    TURN_LATENCY,
    TTS_LATENCY,
    flush_langfuse,
    start_session_trace,
    trace_span,
)
from policy import PolicyEngine
from tts_client import OpenAITTSClient, get_voice_for_locale

logger = logging.getLogger(__name__)


class StreamingGateway:
    """Manage a single voice call session over WebSocket."""

    def __init__(self, dialogue_manager: DialogueManager):
        self.dialogue_manager = dialogue_manager
        self._settings = get_settings()

    async def handle_session(self, websocket: WebSocket, caller_id: str, locale: str = "en-US"):
        """Run the full voice session lifecycle."""
        session_id = str(uuid.uuid4())
        session = SessionState(
            session_id=session_id,
            caller_id=caller_id,
            locale=locale,
        )

        # Initialize components
        asr = DeepgramASRClient()
        tts = OpenAITTSClient()
        barge_in_detector = BargeInDetector()
        trace = start_session_trace(session_id, caller_id)

        ACTIVE_SESSIONS.inc()
        logger.info("Session started: %s (caller=%s, locale=%s)", session_id, caller_id, locale)

        try:
            await websocket.accept()

            # Send greeting
            greeting = "Hello, thank you for calling. How can I help you today?"
            await self._stream_tts(websocket, tts, greeting, locale, barge_in_detector, trace)

            # Start ASR
            dialect_config = PolicyEngine().get_dialect_config(locale)
            await asr.start(locale, hints=dialect_config.get("asr_hints"))

            # Main session loop
            await self._session_loop(
                websocket, session, asr, tts, barge_in_detector, trace
            )

            SESSION_SUCCESS.labels(outcome="completed").inc()

        except WebSocketDisconnect:
            logger.info("Session %s: caller disconnected", session_id)
            SESSION_SUCCESS.labels(outcome="completed").inc()

        except Exception as exc:
            logger.error("Session %s error: %s", session_id, exc, exc_info=True)
            SESSION_SUCCESS.labels(outcome="error").inc()

        finally:
            await asr.close()
            ACTIVE_SESSIONS.dec()
            logger.info("Session ended: %s (%d turns)", session_id, len(session.turns))

    async def _session_loop(
        self,
        websocket: WebSocket,
        session: SessionState,
        asr,
        tts,
        barge_in_detector: BargeInDetector,
        trace,
    ):
        """Process turns until session ends or escalation is triggered."""
        transcript_buffer = ""

        # Launch concurrent audio receiver
        audio_queue: asyncio.Queue[bytes] = asyncio.Queue()
        receiver_task = asyncio.create_task(
            self._receive_audio(websocket, audio_queue)
        )

        try:
            while not session.escalation_required:
                # Check turn limit
                if len(session.turns) >= self._settings.max_turns_per_session:
                    await self._stream_tts(
                        websocket, tts,
                        "We have reached our session limit. Let me transfer you to an agent.",
                        session.locale, barge_in_detector, trace,
                    )
                    session.escalation_required = True
                    break

                # Read audio and process ASR
                try:
                    audio_chunk = await asyncio.wait_for(
                        audio_queue.get(), timeout=0.05
                    )
                except asyncio.TimeoutError:
                    # Check for ASR transcripts even without new audio
                    async for transcript in asr.transcripts():
                        if transcript.is_final and transcript.text.strip():
                            transcript_buffer = transcript.text.strip()
                        if transcript.speech_final and transcript_buffer:
                            # Process the completed utterance
                            await self._process_utterance(
                                websocket, session, tts, barge_in_detector,
                                transcript_buffer, trace,
                            )
                            transcript_buffer = ""
                    continue

                # Check for barge-in during TTS
                if barge_in_detector.check_audio(audio_chunk):
                    BARGE_INS.inc()
                    if session.turns:
                        handle_barge_in(session, session.turns[-1])
                    transcript_buffer = ""

                # Feed audio to ASR
                await asr.send_audio(audio_chunk)

                # Process any ready transcripts
                async for transcript in asr.transcripts():
                    if transcript.is_final and transcript.text.strip():
                        transcript_buffer = transcript.text.strip()
                    if transcript.speech_final and transcript_buffer:
                        await self._process_utterance(
                            websocket, session, tts, barge_in_detector,
                            transcript_buffer, trace,
                        )
                        transcript_buffer = ""

        finally:
            receiver_task.cancel()
            try:
                await receiver_task
            except asyncio.CancelledError:
                pass

    async def _receive_audio(self, websocket: WebSocket, queue: asyncio.Queue):
        """Continuously receive audio bytes from the WebSocket."""
        try:
            while True:
                data = await websocket.receive_bytes()
                await queue.put(data)
        except (WebSocketDisconnect, asyncio.CancelledError):
            pass

    async def _process_utterance(
        self, websocket, session, tts, barge_in_detector, user_text, trace
    ):
        """Process a complete user utterance through the dialogue pipeline."""
        turn_start = time.time()

        async with trace_span(trace, "turn", {"user_text": user_text}):
            # Run dialogue manager
            turn = await self.dialogue_manager.process_turn(session, user_text, trace)

            # Stream TTS response
            if turn.agent_text:
                await self._stream_tts(
                    websocket, tts, turn.agent_text,
                    session.locale, barge_in_detector, trace,
                )

        turn.latency_ms = int((time.time() - turn_start) * 1000)
        TURN_LATENCY.observe(turn.latency_ms)

    async def _stream_tts(
        self, websocket, tts, text, locale, barge_in_detector, trace
    ):
        """Stream TTS audio to the caller with barge-in support."""
        cancel_event = asyncio.Event()
        barge_in_detector.set_tts_active(True, cancel_event)
        voice = get_voice_for_locale(locale)

        tts_start = time.time()
        first_chunk = True

        try:
            async with trace_span(trace, "tts", {"text": text[:100]}):
                async for chunk in tts.synthesize(text, voice, cancel_event):
                    if first_chunk:
                        TTS_LATENCY.observe(int((time.time() - tts_start) * 1000))
                        first_chunk = False
                    try:
                        await websocket.send_bytes(chunk)
                    except Exception:
                        break
        finally:
            barge_in_detector.set_tts_active(False)

import asyncio
import logging
import time
from contextlib import asynccontextmanager

from fastapi import FastAPI, WebSocket, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST

from config import get_settings
from dialogue_manager import DialogueManager
from observability import ACTIVE_SESSIONS, flush_langfuse
from policy import PolicyEngine
from rag_pipeline import KnowledgeBase
from streaming_gateway import StreamingGateway
from tool_router import ToolRouter

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup and shutdown lifecycle."""
    logger.info("Voice agent starting up...")
    settings = get_settings()

    # Initialize RAG if enabled
    kb = None
    if settings.enable_rag:
        kb = KnowledgeBase()
        logger.info("Knowledge base initialized (collection=%s)", settings.chroma_collection)

    # Build dependency chain
    tool_router = ToolRouter(order_client=None, kb_client=kb)
    policy_engine = PolicyEngine()
    dialogue_manager = DialogueManager(tool_router, policy_engine)
    gateway = StreamingGateway(dialogue_manager)

    # Store in app state
    app.state.gateway = gateway
    app.state.start_time = time.time()

    logger.info("Voice agent ready")
    yield

    # Shutdown
    logger.info("Shutting down...")
    flush_langfuse()


app = FastAPI(
    title="Production Voice Agent",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.websocket("/ws/call")
async def websocket_call(
    websocket: WebSocket,
    caller_id: str = Query(...),
    locale: str = Query("en-US"),
):
    """WebSocket endpoint for voice call sessions.

    Connect with: ws://localhost:8000/ws/call?caller_id=user123&locale=en-US
    Send: raw PCM 16-bit 16kHz audio bytes
    Receive: raw PCM audio bytes (TTS output)
    """
    gateway: StreamingGateway = app.state.gateway
    await gateway.handle_session(websocket, caller_id, locale)


@app.get("/health")
async def health():
    """Liveness probe."""
    return {"status": "ok"}


@app.get("/ready")
async def ready():
    """Readiness probe. Reports not ready if too many active sessions."""
    active = ACTIVE_SESSIONS._value.get()
    return {
        "status": "ready" if active < 100 else "overloaded",
        "active_sessions": active,
        "uptime_seconds": int(time.time() - app.state.start_time),
    }


@app.get("/metrics", response_class=PlainTextResponse)
async def metrics():
    """Prometheus metrics endpoint."""
    return PlainTextResponse(
        content=generate_latest().decode("utf-8"),
        media_type=CONTENT_TYPE_LATEST,
    )

import asyncio
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path

from dialogue_manager import DialogueManager
from models import SessionState
from policy import PolicyEngine
from tool_router import ToolRouter

logger = logging.getLogger(__name__)


@dataclass
class Scenario:
    """A test scenario with expected outcomes."""
    name: str
    caller_id: str
    turns: list[dict]  # [{"user": str, "expect_tool": str|None, "expect_escalation": bool}]
    locale: str = "en-US"


@dataclass
class EvalResult:
    """Results from evaluating a single scenario."""
    scenario_name: str
    passed: bool
    turn_results: list[dict] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


@dataclass
class EvalReport:
    """Aggregated evaluation report."""
    total: int = 0
    passed: int = 0
    failed: int = 0
    results: list[EvalResult] = field(default_factory=list)

    @property
    def success_rate(self) -> float:
        return self.passed / self.total if self.total > 0 else 0.0


def load_scenarios(path: str) -> list[Scenario]:
    """Load test scenarios from a JSON file."""
    data = json.loads(Path(path).read_text())
    return [
        Scenario(
            name=s["name"],
            caller_id=s.get("caller_id", "test_user"),
            turns=s["turns"],
            locale=s.get("locale", "en-US"),
        )
        for s in data
    ]


async def evaluate_scenario(
    dm: DialogueManager, scenario: Scenario
) -> EvalResult:
    """Run a single scenario and check expectations."""
    session = SessionState(
        session_id=f"eval_{scenario.name}",
        caller_id=scenario.caller_id,
        locale=scenario.locale,
        authenticated=True,  # Skip auth for eval
    )

    result = EvalResult(scenario_name=scenario.name, passed=True)

    for i, turn_spec in enumerate(scenario.turns):
        user_text = turn_spec["user"]
        expect_tool = turn_spec.get("expect_tool")
        expect_escalation = turn_spec.get("expect_escalation", False)

        turn = await dm.process_turn(session, user_text)

        turn_result = {
            "turn": i + 1,
            "user": user_text,
            "agent": turn.agent_text[:100],
            "tools_called": [r.tool_name for r in turn.tool_results],
            "escalated": session.escalation_required,
        }

        # Check tool expectation
        if expect_tool:
            tools_called = [r.tool_name for r in turn.tool_results]
            if expect_tool not in tools_called:
                result.passed = False
                result.errors.append(
                    f"Turn {i+1}: expected tool '{expect_tool}', got {tools_called}"
                )

        # Check escalation expectation
        if expect_escalation and not session.escalation_required:
            result.passed = False
            result.errors.append(f"Turn {i+1}: expected escalation, did not happen")

        result.turn_results.append(turn_result)

    return result


async def run_evaluation(scenarios_path: str) -> EvalReport:
    """Run all scenarios and produce a report."""
    scenarios = load_scenarios(scenarios_path)
    tool_router = ToolRouter(order_client=None, kb_client=None)
    policy_engine = PolicyEngine()
    dm = DialogueManager(tool_router, policy_engine)

    report = EvalReport(total=len(scenarios))

    for scenario in scenarios:
        result = await evaluate_scenario(dm, scenario)
        report.results.append(result)
        if result.passed:
            report.passed += 1
        else:
            report.failed += 1
            logger.warning("FAILED: %s - %s", result.scenario_name, result.errors)

    logger.info(
        "Evaluation complete: %d/%d passed (%.0f%%)",
        report.passed, report.total, report.success_rate * 100,
    )
    return report

import asyncio
import logging
import random
import time
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


@dataclass
class CallResult:
    """Result of a single simulated call."""
    call_id: int
    success: bool
    turn_latencies_ms: list[int] = field(default_factory=list)
    error: str | None = None
    barge_in_count: int = 0


@dataclass
class StressReport:
    """Aggregated stress test results."""
    total_calls: int
    successful: int
    failed: int
    p50_latency_ms: float
    p95_latency_ms: float
    p99_latency_ms: float
    avg_latency_ms: float
    error_rate: float
    barge_in_total: int


def percentile(values: list[float], p: float) -> float:
    """Compute the p-th percentile of a sorted list."""
    if not values:
        return 0.0
    sorted_v = sorted(values)
    k = (len(sorted_v) - 1) * (p / 100)
    f = int(k)
    c = f + 1 if f + 1 < len(sorted_v) else f
    return sorted_v[f] + (k - f) * (sorted_v[c] - sorted_v[f])


async def simulate_call(session_runner, call_id: int) -> CallResult:
    """Simulate a single call with jitter and optional failures."""
    result = CallResult(call_id=call_id, success=True)

    # Network jitter before connecting
    await asyncio.sleep(random.uniform(0.01, 0.2))

    test_utterances = [
        "I want to check my order status",
        "When will it arrive?",
        "What is your return policy?",
    ]

    # Simulate random barge-in
    if random.random() < 0.15:
        result.barge_in_count += 1

    try:
        for utterance in test_utterances:
            start = time.time()

            # Simulate tool latency spikes (10% chance)
            if random.random() < 0.10:
                await asyncio.sleep(random.uniform(1.0, 3.0))
            else:
                await asyncio.sleep(random.uniform(0.1, 0.5))

            # Simulate random tool failure (5% chance)
            if random.random() < 0.05:
                raise RuntimeError(f"Simulated tool timeout for call {call_id}")

            latency = int((time.time() - start) * 1000)
            result.turn_latencies_ms.append(latency)

    except Exception as exc:
        result.success = False
        result.error = str(exc)

    return result


async def run_stress_test(
    session_runner=None, concurrent_calls: int = 100
) -> StressReport:
    """Run concurrent simulated calls and produce a report."""
    logger.info("Starting stress test with %d concurrent calls", concurrent_calls)

    tasks = [simulate_call(session_runner, i) for i in range(concurrent_calls)]
    results: list[CallResult] = await asyncio.gather(*tasks, return_exceptions=False)

    # Aggregate metrics
    all_latencies = []
    successful = 0
    failed = 0
    barge_ins = 0

    for r in results:
        if r.success:
            successful += 1
            all_latencies.extend(r.turn_latencies_ms)
        else:
            failed += 1
        barge_ins += r.barge_in_count

    report = StressReport(
        total_calls=concurrent_calls,
        successful=successful,
        failed=failed,
        p50_latency_ms=percentile(all_latencies, 50),
        p95_latency_ms=percentile(all_latencies, 95),
        p99_latency_ms=percentile(all_latencies, 99),
        avg_latency_ms=sum(all_latencies) / len(all_latencies) if all_latencies else 0,
        error_rate=failed / concurrent_calls if concurrent_calls > 0 else 0,
        barge_in_total=barge_ins,
    )

    logger.info(
        "Stress test complete: %d/%d passed, P95=%.0fms, error_rate=%.1f%%",
        successful, concurrent_calls, report.p95_latency_ms, report.error_rate * 100,
    )
    return report

from ragas import evaluate
from ragas.metrics import answer_relevancy, faithfulness, context_precision


def run_ragas(dataset):
    """Evaluate RAG quality using RAGAS metrics.

    Args:
        dataset: A RAGAS-compatible dataset with columns:
            question, answer, contexts, ground_truth

    Returns:
        RAGAS evaluation result with per-metric scores.
    """
    return evaluate(
        dataset,
        metrics=[answer_relevancy, faithfulness, context_precision],
    )

import asyncio
import struct

import pytest

from barge_in import BargeInDetector, compute_rms_energy, handle_barge_in
from models import SessionState, TurnState


def _make_audio(amplitude: int, n_samples: int = 320) -> bytes:
    """Generate a PCM audio chunk with constant amplitude."""
    return struct.pack(f"<{n_samples}h", *([amplitude] * n_samples))


class TestRMSEnergy:
    def test_silence(self):
        audio = _make_audio(0)
        assert compute_rms_energy(audio) == 0.0

    def test_loud_audio(self):
        audio = _make_audio(5000)
        assert compute_rms_energy(audio) > 1000

    def test_empty_chunk(self):
        assert compute_rms_energy(b"") == 0.0


class TestBargeInDetector:
    def test_no_detection_when_tts_inactive(self):
        detector = BargeInDetector()
        loud = _make_audio(5000)
        assert detector.check_audio(loud) is False

    def test_detection_after_consecutive_frames(self):
        detector = BargeInDetector(energy_threshold=500)
        cancel = asyncio.Event()
        detector.set_tts_active(True, cancel)

        loud = _make_audio(5000)
        # Need MIN_SPEECH_FRAMES consecutive loud frames
        detector.check_audio(loud)
        detector.check_audio(loud)
        assert detector.check_audio(loud) is True
        assert cancel.is_set()

    def test_no_detection_on_quiet_audio(self):
        detector = BargeInDetector(energy_threshold=500)
        detector.set_tts_active(True, asyncio.Event())

        quiet = _make_audio(10)
        for _ in range(10):
            assert detector.check_audio(quiet) is False

    def test_reset_on_silence_gap(self):
        detector = BargeInDetector(energy_threshold=500)
        detector.set_tts_active(True, asyncio.Event())

        loud = _make_audio(5000)
        quiet = _make_audio(10)

        detector.check_audio(loud)
        detector.check_audio(loud)
        detector.check_audio(quiet)  # Reset consecutive count
        assert detector.check_audio(loud) is False  # Back to 1


class TestHandleBargeIn:
    def test_marks_turn_interrupted(self):
        session = SessionState(session_id="s1", caller_id="c1")
        turn = TurnState(turn_id=1, agent_text="Speaking...")

        result = handle_barge_in(session, turn)

        assert turn.interrupted is True
        assert result["action"] == "stop_tts"

import asyncio
from unittest.mock import AsyncMock, MagicMock

import pytest

from models import SessionState
from tool_router import CircuitBreaker, CircuitState, ToolRouter


class TestCircuitBreaker:
    def test_starts_closed(self):
        cb = CircuitBreaker()
        assert cb.state == CircuitState.CLOSED
        assert cb.allow_request() is True

    def test_opens_after_threshold(self):
        cb = CircuitBreaker(failure_threshold=3)
        for _ in range(3):
            cb.record_failure()
        assert cb.state == CircuitState.OPEN
        assert cb.allow_request() is False

    def test_resets_on_success(self):
        cb = CircuitBreaker(failure_threshold=3)
        cb.record_failure()
        cb.record_failure()
        cb.record_success()
        assert cb.failure_count == 0
        assert cb.state == CircuitState.CLOSED

    def test_half_open_after_recovery(self):
        cb = CircuitBreaker(failure_threshold=2, recovery_timeout=0)
        cb.record_failure()
        cb.record_failure()
        assert cb.state == CircuitState.OPEN
        # With recovery_timeout=0, should immediately go to HALF_OPEN
        assert cb.allow_request() is True
        assert cb.state == CircuitState.HALF_OPEN


class TestToolRouter:
    @pytest.mark.asyncio
    async def test_successful_tool_call(self):
        mock_client = MagicMock()
        mock_client.get_status = AsyncMock(
            return_value={"status": "shipped", "eta": "2 days"}
        )
        router = ToolRouter(order_client=mock_client)
        session = SessionState(session_id="s1", caller_id="c1")

        result = await router.call_tool("order_status", session)
        assert result.success is True
        assert result.data["status"] == "shipped"

    @pytest.mark.asyncio
    async def test_fallback_on_failure(self):
        mock_client = MagicMock()
        mock_client.get_status = AsyncMock(side_effect=Exception("timeout"))
        router = ToolRouter(order_client=mock_client)
        session = SessionState(session_id="s1", caller_id="c1")

        result = await router.call_tool("order_status", session)
        assert result.success is False

        fallback = router.get_fallback_response("order_status")
        assert "connect you" in fallback.lower()

    @pytest.mark.asyncio
    async def test_unknown_tool(self):
        router = ToolRouter()
        session = SessionState(session_id="s1", caller_id="c1")

        result = await router.call_tool("nonexistent", session)
        assert result.success is False
        assert "Unknown tool" in result.error