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")

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)

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"])

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?")

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,
    }

[0, 512, 1024, 2048, 1024, 512, 0, -512, -1024, -2048, ...]
   ← silence →    ← getting louder →   ← wave peak →

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?"

Chat chatbot response:         Voice agent response:
"Here are the details         "Your order ships tomorrow
 about your order:             and arrives by Thursday."
 • Status: Shipped             (2 sentences max)
 • ETA: Thursday
 • Carrier: FedEx
 • Tracking: 123456789"

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