"""
khabeer_livekit — LiveKit Agents plugin for 5abear | خبير (Saudi STT + TTS).

Usage:
    pip install livekit-agents aiohttp
    export KHABEER_API_KEY=kb_live_xxx

    from khabeer_livekit import KhabeerSTT, KhabeerLLM, KhabeerTTS

    session = AgentSession(
        stt=KhabeerSTT(hints=["الرياض", "شاورما", "مضمون"]),
        llm=KhabeerLLM(agent_id="<agent uuid from the Khabeer dashboard>"),
        tts=KhabeerTTS(voice="noura", accent="najdi", speed=1.05),
    )

KhabeerLLM is optional: keep your own LLM if you prefer. Using it means the
reply comes from your Khabeer agent with its business info, knowledge base and
pronunciation dictionary already applied — and no second API key.

Low-latency tip: KhabeerTTS accepts streaming text. With a streaming LLM,
each sentence of the model's reply is synthesized the moment it is complete —
the caller hears the first words while the rest of the reply is still being
written. This cuts the perceived response delay by 2–4 seconds. Keep the LLM
prompt short (one or two sentences per reply) for the best result.

Both classes speak plain HTTPS to the Khabeer API, so no extra credentials
beyond your API key are required.
"""

from __future__ import annotations

import asyncio
import base64
import io
import json
import logging
import os
import re
import time
import wave
from dataclasses import dataclass

import aiohttp
from livekit import rtc
from livekit.agents import (
    DEFAULT_API_CONNECT_OPTIONS,
    NOT_GIVEN,
    APIConnectOptions,
    APIConnectionError,
    APIStatusError,
    llm,
    stt,
    tts,
    utils,
)

BASE_URL = os.getenv("KHABEER_BASE_URL", "https://ai.5abear.com/api/public/v1")
SAMPLE_RATE = 24000
NUM_CHANNELS = 1

# Every step of a turn is timed and logged so a slow call can be diagnosed from
# the journal: look for lines starting with "khabeer-timing".
_timing = logging.getLogger("khabeer-timing")
logger = logging.getLogger("khabeer-livekit")

# The recognizer sometimes returns its own instruction prompt instead of the
# caller's words. Any transcript built mostly from these instruction phrases is
# a provider artefact, never something the caller said.
_STT_PROMPT_MARKERS = (
    "فرّغ كلام المتصل",
    "فرغ كلام المتصل",
    "دون إضافة أمثلة",
    "إذا كان المقطع صامت",
    "ولا تخمّن كلمات",
    "ولا تخمن كلمات",
    "قد تظهر هذه الأسماء",
)


def _is_prompt_echo(text: str) -> bool:
    clean = re.sub(r"\s+", " ", text or "").strip()
    # Diagnostic instructions are distinctive enough to match as substrings.
    # Short Saudi courtesies are valid caller speech and are only artefacts
    # when the recognizer returns the exact prompt example by itself.
    return any(
        (marker in clean if len(marker) > 15 else clean == marker)
        for marker in _STT_PROMPT_MARKERS
    )


# Audio-codec tokens such as <|vq_hbr_audio_4937|> are recogniser artefacts, not
# speech. Strip them before anything else looks at the transcript.
_ARTEFACT_RE = re.compile(r"<\|[^|]*\|>")


def _strip_artifacts(text: str) -> str:
    return re.sub(r"\s+", " ", _ARTEFACT_RE.sub(" ", text or "")).strip()



def _log_timing(step: str, **fields) -> None:
    parts = " ".join(
        f"{k}={v:.2f}s" if isinstance(v, float) else f"{k}={v}" for k, v in fields.items()
    )
    _timing.info("khabeer-timing %s %s", step, parts)


# Optional call recorder (see khabeer_recorder.py). Set by khabeer_agent.py at
# the start of each job; None means nothing is recorded.
_recorder = None

# Called when the backend says the caller said goodbye and the line should be
# closed. khabeer_agent.py installs a handler that hangs up after the farewell.
_end_call_cb = None
_stt_rejection_cb = None


def set_recorder(recorder) -> None:
    global _recorder
    _recorder = recorder


def _record_generated_tts() -> bool:
    """Legacy path: write TTS audio when generated (ahead of playout).

    Disabled once a playout tap is installed, which records only the audio
    the room output actually played, trimmed at interruptions.
    """
    return _recorder is not None and not getattr(_recorder, "playout_tap", False)


def install_playout_recorder(session, recorder) -> bool:
    """Record agent audio at real playout time via LiveKit's own trim logic.

    Wraps session.output.audio with livekit-agents' RecorderAudioOutput, which
    buffers each segment and, on playback_finished, hands back only the
    `playback_position` seconds that were played (pauses as silence). Queued
    audio cancelled by an interruption is discarded. Server-side playout only;
    it cannot prove what the handset received.
    """
    try:
        from livekit.agents.voice.recorder_io.recorder_io import RecorderAudioOutput

        current = session.output.audio
        if current is None:
            return False

        class _Recording:
            recording = True

        def _write(frames) -> None:
            ended = time.monotonic()
            if not frames:
                return
            rate = frames[0].sample_rate
            pcm = b"".join(
                bytes(f.data) if f.num_channels == 1 else _downmix(f) for f in frames
            )
            recorder.add_agent_playout(pcm, rate, ended_at=ended)

        session.output.audio = RecorderAudioOutput(
            recording_io=_Recording(), audio_output=current, write_fnc=_write
        )
        recorder.playout_tap = True
        return True
    except Exception:
        logging.getLogger("khabeer").warning("playout recorder unavailable — recording generated TTS", exc_info=True)
        return False


def _downmix(frame) -> bytes:
    import array

    src = array.array("h", bytes(frame.data))
    ch = frame.num_channels
    out = array.array("h", (int(sum(src[i:i + ch]) / ch) for i in range(0, len(src), ch)))
    return out.tobytes()


def set_end_call_handler(cb) -> None:
    global _end_call_cb
    _end_call_cb = cb


def set_stt_rejection_handler(cb) -> None:
    global _stt_rejection_cb
    _stt_rejection_cb = cb


_NON_ARABIC_SCRIPT = set("پچژکیےٹڈڑںۀہ")


def is_mostly_arabic(text: str, minimum: float = 0.60) -> bool:
    """Accept Saudi Arabic text and reject foreign-script hallucinations."""
    letters = [char for char in text if char.isalpha()]
    if not letters:
        return False
    # Persian/Urdu-only letters (پ چ ژ ک ی ے ٹ ڈ ڑ ں ۀ ہ) are Arabic script
    # but never Saudi Arabic: Deepgram writes background voices this way.
    if any(char in _NON_ARABIC_SCRIPT for char in letters):
        return False
    arabic = sum(
        1
        for char in letters
        if "\u0600" <= char <= "\u06ff"
        or "\u0750" <= char <= "\u077f"
        or "\u08a0" <= char <= "\u08ff"
    )
    return arabic / len(letters) >= minimum


_EXPLICIT_GOODBYES = (
    "مع السلامة",
    "في أمان الله",
    "خلاص شكرا",
    "خلاص شكرًا",
    "الله يعطيك العافية باي",
    "باي باي",
)


def is_explicit_goodbye(text: str) -> bool:
    # Arabic diacritics are optional in STT output; normalize them so
    # "شكرًا" and "شكرا" have identical intent.
    clean = re.sub(r"[\u064b-\u065f\u0670]", "", text or "")
    clean = re.sub(r"[^\w\s]", " ", clean).strip()
    return any(phrase in clean for phrase in _EXPLICIT_GOODBYES)



def _api_key(explicit: str | None = None) -> str:
    key = explicit or os.getenv("KHABEER_API_KEY")
    if not key:
        raise ValueError("Set KHABEER_API_KEY or pass api_key=...")
    return key


def _pcm_to_wav(pcm: bytes, sample_rate: int) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(sample_rate)
        w.writeframes(pcm)
    return buf.getvalue()


def _downsample_16k(pcm: bytes, sample_rate: int) -> tuple[bytes, int]:
    """Cheap nearest-sample resample to 16 kHz (no speech quality loss)."""
    if sample_rate <= 16000 or len(pcm) < 4:
        return pcm, sample_rate
    try:
        import array

        src = array.array("h")
        src.frombytes(pcm[: len(pcm) - (len(pcm) % 2)])
        ratio = sample_rate / 16000.0
        count = int(len(src) / ratio)
        out = array.array("h", (src[min(int(i * ratio), len(src) - 1)] for i in range(count)))
        return out.tobytes(), 16000
    except Exception:  # pragma: no cover - never fail a turn over this
        return pcm, sample_rate


# --------------------------------------------------------------------------- STT


@dataclass
class _STTOptions:
    hints: list[str]


class KhabeerSTT(stt.STT):
    """Saudi Arabic speech recognition backed by POST /v1/stt."""

    def __init__(
        self,
        *,
        api_key: str | None = None,
        hints: list[str] | None = None,
        base_url: str = BASE_URL,
        http_session: aiohttp.ClientSession | None = None,
    ) -> None:
        super().__init__(capabilities=stt.STTCapabilities(streaming=False, interim_results=False))
        self._api_key = _api_key(api_key)
        self._base_url = base_url.rstrip("/")
        self._opts = _STTOptions(hints=hints or [])
        self._session = http_session

    def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None:
            self._session = utils.http_context.http_session()
        return self._session

    async def _recognize_impl(
        self,
        buffer: utils.AudioBuffer,
        *,
        language=NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> stt.SpeechEvent:
        started = time.monotonic()
        frame = utils.merge_frames(buffer)
        pcm = frame.data.tobytes()
        audio_seconds = len(pcm) / (2 * max(frame.sample_rate, 1))
        # Speech recognition gains nothing above 16 kHz, but a 24 kHz upload is
        # 50% more bytes leaving the phone server on every single turn. Send the
        # smaller file; the transcript is identical and the upload is faster.
        upload_pcm, upload_rate = _downsample_16k(pcm, frame.sample_rate)
        wav = _pcm_to_wav(upload_pcm, upload_rate)

        form = aiohttp.FormData()
        form.add_field("file", wav, filename="audio.wav", content_type="audio/wav")
        if self._opts.hints:
            form.add_field("hints", ",".join(self._opts.hints))

        try:
            async with self._ensure_session().post(
                f"{self._base_url}/stt",
                data=form,
                headers={"Authorization": f"Bearer {self._api_key}"},
            ) as resp:
                if resp.status != 200:
                    raise APIStatusError(await resp.text(), status_code=resp.status)
                payload = await resp.json()
        except aiohttp.ClientError as err:  # network-level failure
            raise APIConnectionError() from err

        text = _strip_artifacts(payload.get("text", ""))
        accepted = payload.get("accepted")
        rejection_reason = payload.get("rejection_reason")
        # Keep valid short Saudi answers such as "لا", "أبي", "أبغى" and
        # "تمام". The previous character-count guard erased them, leaving the
        # agent with an empty turn and causing repeated "ما فهمتك" replies.
        clean_text = text.strip()
        if (
            not clean_text
            # A quick "لا" can be under 0.35 s; only drop sub-word clicks.
            or audio_seconds < 0.15
            or _is_prompt_echo(clean_text)
            # Callers speak Arabic. Latin/Cyrillic/Japanese output is a
            # misrecognition; answering it as if it made sense derails the call.
            or accepted is False
            or not is_mostly_arabic(clean_text)
        ):
            if clean_text and not is_mostly_arabic(clean_text):
                _timing.warning("non_arabic_stt dropped: %r", clean_text[:80])
            if rejection_reason == "wrong_language" and _stt_rejection_cb is not None:
                try:
                    _stt_rejection_cb("wrong_language")
                except Exception:
                    _timing.warning("stt rejection callback failed", exc_info=True)
            text = ""

        # Caller audio is recorded by the room-track tap in khabeer_agent.
        _log_timing(
            "stt",
            took=time.monotonic() - started,
            audio=audio_seconds,
            kb=round(len(wav) / 1024),
            chars=len(text),
        )

        return stt.SpeechEvent(
            type=stt.SpeechEventType.FINAL_TRANSCRIPT,
            alternatives=[
                stt.SpeechData(text=text, language=language if isinstance(language, str) else "ar-SA")
            ],
        )


# --------------------------------------------------------------------------- TTS


class KhabeerTTS(tts.TTS):
    """Saudi Arabic speech synthesis backed by POST /v1/tts."""

    def __init__(
        self,
        *,
        voice: str = "noura",
        # None = use the voice's own Saudi accent profile. Override with
        # "najdi" (Riyadh), "hijazi" (Jeddah), "sharqawi" or "aam".
        accent: str | None = None,
        speed: float = 1.0,
        api_key: str | None = None,
        base_url: str = BASE_URL,
        http_session: aiohttp.ClientSession | None = None,
    ) -> None:
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=True),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
        )
        self._api_key = _api_key(api_key)
        self._base_url = base_url.rstrip("/")
        self._voice = voice
        self._accent = accent
        self._speed = speed
        self._session = http_session
        self._greeting_completed = False
        self._replayed_greeting_recovery_spoken = False

    def mark_greeting_completed(self) -> None:
        self._greeting_completed = True

    def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None:
            self._session = utils.http_context.http_session()
        return self._session

    def synthesize(
        self,
        text: str,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> "ChunkedStream":
        return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    def stream(
        self,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> "SynthesizeStream":
        return SynthesizeStream(tts=self, conn_options=conn_options)


# Split streamed LLM tokens into speakable chunks. Arabic punctuation marks
# (؟ ، ؛) count as boundaries, and very long fragments are cut at a comma or
# space so synthesis never waits on a huge paragraph.
_SENTENCE_END = re.compile(r"^(.*?[.!?؟…؛\n]+)\s*", re.S)
_MAX_FRAGMENT = 220
# Start with a short, complete phrase rather than one word. Synthesizing
# "تمام" alone made the caller hear a word followed by a fresh TTS request,
# which sounded exactly like the voice disappeared as the reply began.
_FIRST_FRAGMENT = 32
_FIRST_MIN = 12
# A complete two-word Saudi opener («هلا والله،» / «حياك الله،») may be sent
# as soon as its comma arrives. One-word openers («أبشر،») still wait so the
# caller never hears an isolated word followed by a fresh TTS request.
_OPENER_MIN_CHARS = 7


def _first_opener_cut(buf: str) -> int:
    cut = buf.find("،")
    if cut == -1 or cut >= _FIRST_MIN:
        return -1
    head = buf[:cut].strip()
    if len(head) >= _OPENER_MIN_CHARS and len(head.split()) >= 2:
        return cut
    return -1


# Phone turn-taking tuning, kept pure so tests can build a real AgentSession
# from exactly what the worker passes (LiveKit 1.6.10 maps these legacy kwargs
# onto turn_handling.interruption / endpointing).
def phone_session_tuning() -> dict:
    return {
        "min_endpointing_delay": 0.35,
        "max_endpointing_delay": 0.70,
        "preemptive_generation": False,
        "allow_interruptions": True,
        # 0.65 s of sustained VAD speech is the noise/echo protection. One
        # recognized (Arabic, guard-passed) word is enough after that: with two,
        # short Saudi barge-ins such as «لحظة» or «لا» never stopped the reply.
        "min_interruption_duration": 0.65,
        "min_interruption_words": 1,
        # Word-less noise never passes the word gate; if it somehow pauses the
        # reply, LiveKit resumes it when no transcript follows.
        "resume_false_interruption": True,
        "false_interruption_timeout": 1.0,
    }


def apply_stt_endpointing(session_kwargs: dict) -> dict:
    """Streaming STT (Deepgram endpointing_ms=300) owns end-of-turn.

    LiveKit 1.6.10 applies endpointing.min_delay (default 0.5 s when unset)
    measured from the VAD speech end, on top of the STT end-of-speech event.
    Set it to 0 so Deepgram's 300 ms endpoint is the only wait.
    """
    session_kwargs["turn_detection"] = "stt"
    session_kwargs["min_endpointing_delay"] = 0.0
    session_kwargs.pop("max_endpointing_delay", None)
    return session_kwargs


class TurnClock:
    """Correlated per-turn timings on ONE monotonic clock.

    A turn starts when the caller starts speaking while the agent is idle.
    Speech that starts while the agent is speaking is recorded as an
    interruption of the current turn instead. Every value is milliseconds
    from the turn's VAD speech end (or from interruption start for barge-in
    marks), and marks reset on each new turn so a late event from the
    previous reply can never be attached to the next one. No text or audio
    is stored — only times and word counts.
    """

    def __init__(self, now=time.monotonic) -> None:
        self._now = now
        self.turn_id = 0
        self.marks: dict[str, float] = {}
        self.counts: dict[str, int] = {}
        self.agent_speaking = False
        self.speech_id: str | None = None

    def user_started(self) -> str:
        t = self._now()
        if self.agent_speaking:
            self.marks.setdefault("barge_in_start", t)
            return "barge_in"
        self.turn_id += 1
        self.marks = {"user_speech_start": t}
        self.counts = {}
        self.speech_id = None
        return "turn"

    def user_stopped(self) -> None:
        """Caller went quiet. After an interruption that stopped the agent,
        the caller's words are a NEW turn timed from this speech end; marking
        them on the interrupted turn produced bogus 15.9 s "final" figures."""
        if self.agent_speaking:
            return
        if "barge_in_stop" in self.marks:
            start = self.marks.get("barge_in_start", self._now())
            self.turn_id += 1
            self.marks = {"user_speech_start": start}
            self.counts = {}
            self.speech_id = None
        self.mark("user_speech_end")

    def mark(self, name: str) -> bool:
        if name in self.marks:
            return False
        self.marks[name] = self._now()
        return True

    def interim(self, words: int) -> None:
        if "barge_in_start" in self.marks and "barge_in_stop" not in self.marks:
            if self.mark("barge_in_first_interim"):
                self.counts["barge_in_first_interim_words"] = words
        elif "user_speech_end" not in self.marks or "final_transcript" not in self.marks:
            self.mark("first_interim")

    def agent_state(self, state: str) -> None:
        if state == "thinking":
            self.mark("thinking")
        if state == "speaking":
            self.agent_speaking = True
            self.mark("playout_started")
        elif self.agent_speaking:
            self.agent_speaking = False
            if "barge_in_start" in self.marks:
                self.mark("barge_in_stop")
            else:
                self.mark("agent_done")

    def accept_metric(self, speech_id: str | None) -> bool:
        """Bind the first reply's speech_id to this turn; reject others."""
        if not speech_id:
            return True
        if self.speech_id is None:
            self.speech_id = speech_id
        return speech_id == self.speech_id

    def snapshot(self) -> dict:
        out: dict = {"turn_id": self.turn_id}
        end = self.marks.get("user_speech_end")
        barge = self.marks.get("barge_in_start")
        for name, t in self.marks.items():
            if name.startswith("barge_in"):
                if barge is not None and name != "barge_in_start":
                    out[f"{name}_ms"] = round((t - barge) * 1000)
            elif end is not None and name not in ("user_speech_start", "user_speech_end"):
                out[f"{name}_ms"] = round((t - end) * 1000)
        if end is not None and "user_speech_start" in self.marks:
            out["user_speech_ms"] = round((end - self.marks["user_speech_start"]) * 1000)
        out.update(self.counts)
        return out


def should_reprompt(
    now: float,
    marks: dict,
    *,
    agent_state: str | None,
    speech_active: bool,
    user_speaking: bool,
    already_for: float,
    idle_s: float = 8.0,
) -> bool:
    """Watchdog «عفوًا، وش قلت؟» only for a genuinely unanswered caller turn.

    Never while the caller talks, while the agent thinks/speaks/runs a tool,
    or when ANY agent activity (thinking, speaking, finished reply) happened
    after the caller stopped — e.g. a reply resumed after a false
    interruption. The old check compared only the speaking START, so a
    resumed reply was followed by a reprompt the moment it ended.
    """
    stopped = marks.get("turn_start") or 0.0
    if stopped <= 0 or already_for == stopped:
        return False
    if user_speaking or speech_active or agent_state in ("thinking", "speaking"):
        return False
    after = max(marks.get("speaking") or 0.0, marks.get("thinking") or 0.0, marks.get("agent_done") or 0.0)
    if after >= stopped:
        return False
    return now - stopped > idle_s


def may_say_noise_prompt(*, agent_state: str | None, speech_active: bool, user_speaking: bool) -> bool:
    """A rejected (noise/foreign) transcript may trigger a recovery line only
    when nobody is talking and no reply or tool call is under way."""
    return not (speech_active or user_speaking or agent_state in ("thinking", "speaking"))


# --- Tool engagement phrases -------------------------------------------------
# Platform-level acknowledgments spoken while a customer-system action runs,
# for EVERY phone agent (existing and newly created), so the caller never
# hears silence. They state INTENTION only — never claim the action succeeded
# before the tool returns. Saudi Arabic is the default product language; an
# agent whose dispatch metadata sets another language gets that language's
# phrases. No agent, client or business name is hardcoded.

_TOOL_PHRASES: dict[str, dict[str, str]] = {
    "ar": {
        "whatsapp": "أبشر، أرسلها لك الحين.",
        "lookup": "لحظة، أشيّك لك على المعلومة.",
        "generic": "لحظة واحدة.",
        "progress": "ما زلت أشيّك عليها، ثواني.",
    },
    "en": {
        "whatsapp": "Sure, sending it to you now.",
        "lookup": "One moment, let me check that for you.",
        "generic": "One moment please.",
        "progress": "Still checking that for you.",
    },
}

TOOL_ACK_WHATSAPP = _TOOL_PHRASES["ar"]["whatsapp"]
TOOL_ACK_LOOKUP = _TOOL_PHRASES["ar"]["lookup"]
TOOL_ACK_GENERIC = _TOOL_PHRASES["ar"]["generic"]
TOOL_PROGRESS = _TOOL_PHRASES["ar"]["progress"]

# One progress phrase at most, only if the tool is genuinely still running.
TOOL_PROGRESS_DELAY_S = 4.0

_LOOKUP_HINTS = ("get", "fetch", "search", "lookup", "find", "status", "order", "track", "check", "query", "list")


def _phrase_set(language: str | None) -> dict[str, str]:
    """Saudi Arabic unless the agent is explicitly configured otherwise."""
    lang = (language or "ar").strip().lower()
    return _TOOL_PHRASES.get("ar" if lang.startswith("ar") else lang, _TOOL_PHRASES["en"])


def tool_ack_phrase(key: str, method: str | None = None, language: str | None = None) -> str:
    """Pick the one acknowledgment for a tool call. WhatsApp sends get the
    send-intention phrase; data lookups the checking phrase; anything else a
    generic safe fallback that exposes no implementation detail."""
    k = (key or "").lower()
    phrases = _phrase_set(language)
    if "whatsapp" in k or "whats" in k:
        return phrases["whatsapp"]
    if (method or "").upper() == "GET" or any(h in k for h in _LOOKUP_HINTS):
        return phrases["lookup"]
    return phrases["generic"]


def tool_progress_phrase(key: str, language: str | None = None) -> str:
    """The single progress phrase allowed while a slow tool is still running."""
    return _phrase_set(language)["progress"]


async def speak_tool_ack(
    session,
    key: str,
    done: "asyncio.Event",
    method: str | None = None,
    language: str | None = None,
    can_speak=None,
) -> None:
    """Speak the acknowledgment, then at most ONE progress phrase if the tool
    is still running after TOOL_PROGRESS_DELAY_S. Fire-and-forget from the
    tool call so the tool itself is never delayed.

    Lifecycle guards:
    - The session is passed per call by the tool closure, never read from a
      module global, so concurrent calls on one worker process can never
      address each other's sessions.
    - If the tool already finished (done set) before this task first runs,
      nothing is said — no stale ack after the result.
    - `can_speak` (when given) is checked before every phrase; while the
      caller is speaking (barge-in) no ack/progress starts, so the agent
      never talks over the caller.
    - The pending progress phrase is cancelled when the tool finishes or the
      task is cancelled (session close); a phrase already handed to the
      session keeps playing and the result reply queues after it.
    Barge-in interrupts the speech because allow_interruptions=True; phrases
    stay out of the chat context so the model never treats them as part of
    the reply."""
    if session is None:
        return

    def _may_speak() -> bool:
        if done.is_set():
            return False
        if can_speak is not None:
            try:
                if not can_speak():
                    return False
            except Exception:
                return False
        return True

    try:
        await asyncio.sleep(0)  # let a synchronously-finished tool win
        if not _may_speak():
            return
        session.say(tool_ack_phrase(key, method, language), allow_interruptions=True, add_to_chat_ctx=False)
    except asyncio.CancelledError:
        return
    except Exception:
        logger.debug("tool ack speech failed", exc_info=True)
        return
    try:
        await asyncio.sleep(TOOL_PROGRESS_DELAY_S)
        if _may_speak():
            try:
                session.say(tool_progress_phrase(key, language), allow_interruptions=True, add_to_chat_ctx=False)
            except Exception:
                logger.debug("tool progress speech failed", exc_info=True)
    except asyncio.CancelledError:
        pass


_turn_clock: "TurnClock | None" = None


def set_turn_clock(clock: "TurnClock | None") -> None:
    global _turn_clock
    _turn_clock = clock


def _pop_sentence(buf: str, first: bool = False) -> tuple[str | None, str]:
    # Farewell replies stay one smooth TTS request («العفو، حياك الله. في أمان الله»).
    if len(buf) < 80 and ("أمان" in buf or "امان" in buf or "السلامة" in buf):
        return None, buf
    m = _SENTENCE_END.match(buf)
    if m:
        return m.group(1).strip() or None, buf[m.end() :]
    if first:
        # Cut the first phrase only at a comma — never mid-phrase at a space,
        # which split «حياك الله، في | أمان الله» into choppy audio.
        # Farewells stay whole so they are spoken as one smooth phrase.
        if "الله" in buf and ("أمان" in buf or "امان" in buf or "السلامة" in buf):
            return None, buf
        cut = _first_opener_cut(buf)
        if cut == -1:
            cut = buf.find("،", _FIRST_MIN)
        if cut == -1 and len(buf) > 120:
            cut = buf.rfind(" ", 0, 100)
        if cut >= _OPENER_MIN_CHARS:
            return buf[:cut + 1].strip(), buf[cut + 1 :]
    if len(buf) > _MAX_FRAGMENT:
        cut = buf.rfind("،", 0, _MAX_FRAGMENT)
        if cut == -1:
            cut = buf.rfind(" ", 0, _MAX_FRAGMENT)
        if cut > 40:
            return buf[:cut].strip(), buf[cut + 1 :]
    return None, buf


# Saudi text fixer: GPT sometimes writes MSA words; Nouf must say the Saudi one.
_SAUDI_SWAPS = [
    (r"\bالآن\b", "الحين"),
    (r"\bماذا\b", "وش"),
    (r"\bلماذا\b", "ليش"),
    (r"\bأريد\b", "أبغى"),
    (r"\bتريد\b", "تبغى"),
    (r"\bتريدين\b", "تبغين"),
    (r"\bحسنًا\b|\bحسنا\b", "زين"),
    (r"\bنعم\b", "إيه"),
    (r"\bيوجد\b", "فيه"),
    (r"\bلا يوجد\b", "ما فيه"),
    (r"\bليس\b", "مو"),
    (r"\bتأكد لي\b", "أكّد لي"),
    (r"\bلحظات\b", "ثواني"),
    (r"\bكيف يمكنني\b", "كيف أقدر"),
    (r"\bهل\s+", ""),
    (r"\bشيئًا آخر\b|\bشيئا آخر\b|\bشيء آخر\b", "شي ثاني"),
    # Pronunciation: brand split, smooth Saudi farewell, زِي (kasra) as Saudis say it.
    (r"ديرصن|Dear\s*Sun", "دير صن"),
    (r"(?<!\S)فمان\s+الله|في\s+أمان\s+الله|في\s+امان\s+الله", "في أمانِ الله"),
    (r"(?<!\S)(ال)?زي(?!\S)", r"\1زِي"),
]
_SAUDI_NUM = {
    "0": "صفر", "1": "واحد", "2": "اثنين", "3": "ثلاثة", "4": "أربعة", "5": "خمسة",
    "6": "ستة", "7": "سبعة", "8": "ثمانية", "9": "تسعة", "10": "عشرة", "11": "حدعش",
    "12": "اطنعش", "13": "ثلطعش", "14": "أربعطعش", "15": "خمسطعش", "16": "سطعش",
    "17": "سبعطعش", "18": "ثمنطعش", "19": "تسعطعش", "20": "عشرين", "25": "خمسة وعشرين",
    "30": "ثلاثين", "40": "أربعين", "50": "خمسين", "60": "ستين", "100": "مية",
}


def saudi_fix(text: str) -> str:
    out = text
    for pat, rep in _SAUDI_SWAPS:
        out = re.sub(pat, rep, out)
    out = re.sub(r"\b(\d{1,3})\b", lambda m: _SAUDI_NUM.get(m.group(1), m.group(1)), out)
    return out


# Self-introduction markers only. «هلا والله» is an ordinary Saudi opener
# inside answers; treating it as a greeting replaced valid replies with a
# canned «تشرفنا…» line that sounded like a second introduction.
_GREETING_MARKERS = (
    "معك نورة",
    "من خبير",
    "مستشارة المساعدات الصوتية",
    "مع مين أتشرف",
)


def remove_replayed_greeting(text: str) -> str:
    """Drop a model-generated introduction after the fixed greeting was played.

    The conversation model occasionally treats a caller's name as a fresh call
    and emits the configured greeting again. This output boundary makes that
    impossible even if a prompt or model regresses.
    """
    clean = re.sub(r"\s+", " ", text or "").strip()
    if not clean or not any(marker in clean for marker in _GREETING_MARKERS):
        return clean
    sentences = [part.strip() for part in re.split(r"(?<=[.!؟?])\s+", clean) if part.strip()]
    kept = [part for part in sentences if not any(marker in part for marker in _GREETING_MARKERS)]
    # Never inject a canned line mid-reply; an all-introduction fragment is skipped.
    return " ".join(kept).strip()


# Short fixed phrases (greeting, أبشر, goodbye) are synthesized once per call
# worker and replayed instantly afterwards.
_PHRASE_CACHE: dict[tuple, list[bytes]] = {}
_CACHE_MAX_CHARS = 90


def _cache_key(khabeer: "KhabeerTTS", text: str) -> tuple:
    return (getattr(khabeer, "_provider", None), khabeer._api_key, khabeer._voice, khabeer._accent, khabeer._speed, text.strip())


async def prewarm_phrases(khabeer: "KhabeerTTS", phrases: list[str]) -> None:
    """Render common phrases in the background so they play with no wait."""
    async def one(p: str) -> None:
        try:
            await _fetch_pcm(khabeer, p, _discard)
        except Exception:
            logger.debug("phrase prewarm failed: %s", p, exc_info=True)

    await asyncio.gather(*(one(p) for p in phrases if p))


async def _discard(_chunk: bytes) -> None:
    return None


def _decode_phone_audio(payload: dict) -> bytes:
    """Honor provider PCM metadata before handing frames to the 24 kHz sink."""
    mime = payload.get("mime_type") or "audio/pcm;rate=24000;encoding=s16le;channels=1"
    params = dict(re.findall(r"(?:^|;)\s*(\w+)=([^;]+)", mime))
    rate = int(params.get("rate", "24000"))
    channels = int(params.get("channels", "1"))
    if params.get("encoding", "s16le") != "s16le" or channels != 1 or not 8000 <= rate <= 96000:
        raise APIStatusError("Unsupported phone PCM format", status_code=502)
    chunk = base64.b64decode(payload["audio"], validate=True)
    if not chunk or len(chunk) % 2:
        raise APIStatusError("Invalid phone PCM frame", status_code=502)
    if rate == SAMPLE_RATE:
        return chunk
    resampler = rtc.AudioResampler(rate, SAMPLE_RATE, num_channels=1)
    frame = rtc.AudioFrame(chunk, rate, 1, len(chunk) // 2)
    frames = resampler.push(frame) + resampler.flush()
    return b"".join(bytes(frame.data) for frame in frames)


async def _fetch_pcm(khabeer: "KhabeerTTS", text: str, sink) -> None:
    """POST /v1/tts-stream and hand raw PCM frames to `sink` as they arrive."""
    text = saudi_fix(text)
    if khabeer._greeting_completed and any(marker in text for marker in _GREETING_MARKERS):
        text = remove_replayed_greeting(text)
        _log_timing("tts.self_intro_suppressed", took=0.0, kept_chars=len(text))
        if not text:
            return
    key = _cache_key(khabeer, text)
    cached = _PHRASE_CACHE.get(key)
    if cached:
        _log_timing("tts.cache_hit", took=0.0, chars=len(text))
        for chunk in cached:
            if _recorder is not None and sink is not _discard:
                _recorder.add_audio(chunk, SAMPLE_RATE)
            await sink(chunk)
            if sink is not _discard:
                khabeer._last_audio_at = time.monotonic()
        return
    record = sink is not _discard
    collected: list[bytes] | None = [] if len(text) <= _CACHE_MAX_CHARS else None
    if collected is not None:
        inner_sink = sink

        async def sink(chunk: bytes) -> None:  # type: ignore[no-redef]
            collected.append(chunk)
            await inner_sink(chunk)

    body = {
        "text": text,
        "voice": khabeer._voice,
        "speed": khabeer._speed,
        # LiveKit already sends short speakable fragments. Splitting them again
        # at the API created extra synthesis calls, gaps and uneven delivery.
        "chunking": False,
    }
    if khabeer._accent:
        body["accent"] = khabeer._accent
    if getattr(khabeer, "_provider", None):
        body["provider"] = khabeer._provider

    started = time.monotonic()
    first_audio: float | None = None
    async with khabeer._ensure_session().post(
        f"{khabeer._base_url}/tts-stream",
        json=body,
        headers={"Authorization": f"Bearer {khabeer._api_key}"},
    ) as resp:
        if resp.status != 200:
            raise APIStatusError(await resp.text(), status_code=resp.status)
        _log_timing("tts.headers", took=time.monotonic() - started, chars=len(text))
        event = ""
        buffer = ""

        async def handle_line(line: str) -> bool:
            nonlocal event, first_audio
            line = line.rstrip("\r")
            if line.startswith("event:"):
                event = line[6:].strip()
                return False
            if not line.startswith("data:"):
                return False
            data = line[5:].strip()
            current_event = event
            event = ""
            if not data or data == "[DONE]":
                return data == "[DONE]"
            try:
                payload = json.loads(data)
            except Exception:
                return False
            if current_event == "error":
                raise APIStatusError("Khabeer TTS stream failed", status_code=402 if payload.get("code") == 40201 else 502)
            if current_event == "audio_delta" and payload.get("audio"):
                chunk = _decode_phone_audio(payload)
                if first_audio is None:
                    first_audio = time.monotonic() - started
                    _log_timing("tts.first_audio", took=first_audio, chars=len(text))
                if _recorder is not None and record:
                    _recorder.add_audio(chunk, SAMPLE_RATE)
                await sink(chunk)
                if record:
                    khabeer._last_audio_at = time.monotonic()
            return current_event == "done"

        done = False
        # HTTP chunks do not preserve SSE line boundaries. The old parser
        # treated each network chunk as one line, so a chunk containing both
        # `event:` and `data:` silently discarded that audio and sounded cut.
        async for raw in resp.content.iter_any():
            buffer += raw.decode("utf-8", "ignore")
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                if await handle_line(line):
                    done = True
                    break
            if done:
                break
        if buffer and not done:
            await handle_line(buffer)
    if first_audio is None:
        raise APIStatusError("Khabeer TTS stream produced no audio", status_code=502)
    if collected:
        _PHRASE_CACHE[key] = collected
    _log_timing(
        "tts.done",
        took=time.monotonic() - started,
        first=first_audio if first_audio is not None else -1.0,
        chars=len(text),
    )


async def _stream_pcm(khabeer: "KhabeerTTS", text: str, emitter: "tts.AudioEmitter") -> None:
    async def sink(chunk: bytes) -> None:
        emitter.push(chunk)

    await _fetch_pcm(khabeer, text, sink)


class ChunkedStream(tts.ChunkedStream):
    def __init__(
        self,
        *,
        tts: KhabeerTTS,
        input_text: str,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
        self._tts = tts

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        output_emitter.initialize(
            request_id=utils.shortuuid(),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
            mime_type="audio/pcm",
        )
        recorded = False

        async def sink(chunk: bytes) -> None:
            nonlocal recorded
            if not recorded and _recorder is not None:
                _recorder.add_turn("agent", self._input_text)
                recorded = True
            output_emitter.push(chunk)

        try:
            await _fetch_pcm(self._tts, self._input_text, sink)
        except aiohttp.ClientError as err:
            raise APIConnectionError() from err
        output_emitter.flush()


class SynthesizeStream(tts.SynthesizeStream):
    """Starts the first phrase immediately, then synthesizes sequentially.

    Waiting for the complete model answer added several seconds before speech.
    Sequential requests preserve one voice and avoid the provider throttling
    previously caused by parallel synthesis.
    """

    def __init__(
        self,
        *,
        tts: KhabeerTTS,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        super().__init__(tts=tts, conn_options=conn_options)
        self._tts = tts

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        output_emitter.initialize(
            request_id=utils.shortuuid(),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
            mime_type="audio/pcm",
        )

        buffer = ""
        first = True

        async def speak(fragment: str) -> None:
            if not fragment:
                return
            recorded = False

            async def sink(chunk: bytes) -> None:
                nonlocal recorded
                if _turn_clock is not None:
                    _turn_clock.mark("first_tts_pcm")
                if not recorded and _recorder is not None:
                    _recorder.add_turn("agent", fragment)
                    recorded = True
                output_emitter.push(chunk)

            await _fetch_pcm(self._tts, fragment, sink)

        try:
            async for item in self._input_ch:
                if not isinstance(item, str):
                    continue
                buffer += item
                while True:
                    fragment, buffer = _pop_sentence(buffer, first=first)
                    if not fragment:
                        break
                    await speak(fragment)
                    first = False
            if buffer.strip():
                await speak(buffer.strip())
        except asyncio.CancelledError:
            # Barge-in: LiveKit cancels this task. Leaving `async with` above
            # closes the TTS HTTP response, so the backend stops synthesizing
            # and no queued audio is pushed after the interruption.
            _log_timing("tts.cancelled_by_interruption", took=0.0)
            raise
        except aiohttp.ClientError as err:
            raise APIConnectionError() from err
        finally:
            output_emitter.flush()


# --------------------------------------------------------------------------- LLM


class KhabeerLLM(llm.LLM):
    """The reply itself comes from your Khabeer agent (POST /v1/chat).

    Uses the agent's business info, knowledge base, pronunciation dictionary and
    Saudi prompt, so the caller hears an answer grounded in your business —
    no OpenAI key and no second prompt to maintain.

        llm=KhabeerLLM(agent_id="<agent uuid from the Khabeer dashboard>")
    """

    def __init__(
        self,
        *,
        agent_id: str,
        api_key: str | None = None,
        base_url: str = BASE_URL,
        # Shorter history = smaller prompt = faster first token on every turn.
        history_turns: int = 6,
        http_session: aiohttp.ClientSession | None = None,
    ) -> None:
        super().__init__()
        self._api_key = _api_key(api_key)
        self._base_url = base_url.rstrip("/")
        self._agent_id = agent_id
        self._history_turns = history_turns
        self._session = http_session

    def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None:
            self._session = utils.http_context.http_session()
        return self._session

    def chat(
        self,
        *,
        chat_ctx: "llm.ChatContext",
        tools=None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls=None,
        tool_choice=None,
        extra_kwargs=None,
    ) -> "KhabeerLLMStream":
        return KhabeerLLMStream(
            self, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options
        )


def _plain_turns(chat_ctx: "llm.ChatContext") -> list[dict[str, str]]:
    """Flatten a LiveKit chat context into {role, content} turns."""
    turns: list[dict[str, str]] = []
    items = getattr(chat_ctx, "items", None) or getattr(chat_ctx, "messages", []) or []
    for item in items:
        role = getattr(item, "role", None)
        if role not in ("user", "assistant"):
            continue
        text = getattr(item, "text_content", None)
        if callable(text):
            text = text()
        if text is None:
            content = getattr(item, "content", None)
            if isinstance(content, str):
                text = content
            elif isinstance(content, list):
                text = " ".join(c for c in content if isinstance(c, str))
        if text and text.strip():
            turns.append({"role": role, "content": text.strip()})
    return turns


class KhabeerLLMStream(llm.LLMStream):
    def __init__(
        self,
        khabeer: KhabeerLLM,
        *,
        chat_ctx: "llm.ChatContext",
        tools,
        conn_options: APIConnectOptions,
    ) -> None:
        super().__init__(khabeer, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
        self._khabeer = khabeer

    async def _run(self) -> None:
        turns = _plain_turns(self._chat_ctx)
        if not turns or turns[-1]["role"] != "user":
            return
        message = turns[-1]["content"]
        history = turns[:-1][-self._khabeer._history_turns :]

        request_id = utils.shortuuid()

        def emit(text: str) -> None:
            if text:
                self._event_ch.send_nowait(
                    llm.ChatChunk(
                        id=request_id,
                        delta=llm.ChoiceDelta(role="assistant", content=text),
                    )
                )

        started = time.monotonic()
        first_delta: float | None = None
        chars = 0
        try:
            async with self._khabeer._ensure_session().post(
                f"{self._khabeer._base_url}/chat",
                json={
                    "agent_id": self._khabeer._agent_id,
                    "message": message,
                    "history": history,
                    # Speech is produced by KhabeerTTS so it can start the moment
                    # the first sentence closes — skip audio on the reply itself.
                    "audio": False,
                    # Stream text deltas: the first sentence reaches TTS while the
                    # model is still writing the rest of the answer.
                    "stream": True,
                },
                headers={"Authorization": f"Bearer {self._khabeer._api_key}"},
            ) as resp:
                if resp.status != 200:
                    raise APIStatusError(await resp.text(), status_code=resp.status)
                _log_timing("llm.headers", took=time.monotonic() - started)

                buffer = ""

                def handle_line(line: str) -> bool:
                    nonlocal first_delta, chars
                    line = line.strip()
                    if not line.startswith("data:"):
                        return False
                    data = line[5:].strip()
                    if data == "[DONE]":
                        return True
                    if not data:
                        return False
                    try:
                        event = json.loads(data)
                    except Exception:
                        return False
                    if event.get("end_call") and is_explicit_goodbye(message) and _end_call_cb is not None:
                        try:
                            _end_call_cb()
                        except Exception:  # pragma: no cover
                            pass
                    delta = event.get("delta")
                    if not delta:
                        return False

                    if first_delta is None:
                        first_delta = time.monotonic() - started
                        _log_timing("llm.first_token", took=first_delta, ask=len(message))
                    chars += len(delta)
                    # Forward every token the moment it arrives; the TTS stream
                    # decides where to cut, so nothing is buffered twice.
                    emit(delta)
                    return False

                done = False
                # SSE records and HTTP chunks have unrelated boundaries. Buffer
                # complete lines so no token is lost when several events share
                # one packet or one JSON event spans two packets.
                async for raw in resp.content.iter_any():
                    buffer += raw.decode("utf-8", "ignore")
                    while "\n" in buffer:
                        line, buffer = buffer.split("\n", 1)
                        if handle_line(line):
                            done = True
                            break
                    if done:
                        break
                if buffer and not done:
                    handle_line(buffer)
            _log_timing(
                "llm.done",
                took=time.monotonic() - started,
                first=first_delta if first_delta is not None else -1.0,
                chars=chars,
            )
        except aiohttp.ClientError as err:
            raise APIConnectionError() from err


__all__ = [
    "KhabeerSTT",
    "KhabeerTTS",
    "KhabeerLLM",
    "is_explicit_goodbye",
    "is_mostly_arabic",
    "set_stt_rejection_handler",
    "rtc",
]