hope-voice-api/agent/agent.py

"""
Voice Agent — real-time voice assistant.

Pipeline: Azure STT → LLM (OpenWebUI, OpenAI-compatible) → Azure TTS
Runs inside the single Docker container alongside LiveKit server and web frontend.
"""

import asyncio
import json
import logging
import os
import re
import textwrap
import uuid

from dotenv import load_dotenv
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    TurnHandlingOptions,
    cli,
    inference,
    llm as lk_llm,
    mcp,
    room_io,
)
from livekit.plugins import azure, openai

logger = logging.getLogger("voice-agent")
load_dotenv()  # picks up /app/.env in the container

# ── Configuration from environment ──────────────────────────────────────────
AZURE_KEY = os.environ.get("AZURE_SPEECH_KEY", "")
AZURE_REGION = os.environ.get("AZURE_SPEECH_REGION", "eastus")
DEFAULT_VOICE = os.environ.get(
    "AZURE_TTS_VOICE", "en-US-Ava:DragonHDOmniLatestNeural"
)
LLM_BASE_URL = (
    os.environ.get("LLM_BASE_URL")
    or os.environ.get("GEMMA_BASE_URL")
    or "http://192.168.86.2:3005/api"
)
LLM_MODEL = (
    os.environ.get("LLM_MODEL")
    or os.environ.get("GEMMA_MODEL")
    or "qwen2.5-vl-3b"
)
LLM_API_KEY = (
    os.environ.get("LLM_API_KEY")
    or os.environ.get("GEMMA_API_KEY")
    or "not-needed"
)
# Model ids that cannot hold a conversation (OpenWebUI hosts these too).
NON_CHAT_MODEL_ID = re.compile(r"embed|rerank|whisper|tts-|stable-?diffusion|flux", re.I)
WEB_MCP_ENABLED = os.environ.get("WEB_MCP_ENABLED", "true").lower() in ("1", "true", "yes")
FIRECRAWL_BASE = os.environ.get("FIRECRAWL_BASE", "http://192.168.86.2:3002")
SKILLS_DIR = os.environ.get("SKILLS_DIR", "/skills")
VOICE_USERS_FILE = os.environ.get("VOICE_USERS_FILE", "/etc/hope/users.json")


def load_user_profile(user_id: str) -> dict:
    """Load one server-side user profile without exposing its credentials."""
    with open(VOICE_USERS_FILE, encoding="utf-8") as users_file:
        registry = json.load(users_file)
    profile = registry.get("users", {}).get(user_id)
    if not isinstance(profile, dict):
        raise ValueError(f"unknown voice user: {user_id}")
    memory = profile.get("memory")
    if not isinstance(memory, dict) or not memory.get("url") or not memory.get("api_key"):
        raise ValueError(f"voice user {user_id} has no valid memory profile")
    return {**profile, "user_id": user_id}


def user_id_from_room(room_name: str) -> str:
    """Extract the trusted user slug minted into a server-generated room name."""
    match = re.fullmatch(r"voice-([a-z0-9][a-z0-9_-]{1,30})-[0-9a-f]{8}", room_name)
    if not match:
        raise ValueError(f"invalid voice room name: {room_name}")
    return match.group(1)


async def list_available_models() -> list[str]:
    """Fetch chat-capable model ids from the OpenAI-compatible endpoint.

    Works with OpenWebUI (GET /api/models) and plain OpenAI-compatible
    servers (GET /v1/models) — both live under {base_url}/models. The
    configured default model is always first so it survives fetch failures.
    """
    ids: list[str] = []
    try:
        import httpx

        headers = {}
        if LLM_API_KEY and LLM_API_KEY != "not-needed":
            headers["Authorization"] = f"Bearer {LLM_API_KEY}"
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(f"{LLM_BASE_URL}/models", headers=headers)
            resp.raise_for_status()
            data = resp.json()
        ids = [
            model_id
            for model in data.get("data", [])
            if isinstance(model, dict) and (model_id := model.get("id", ""))
        ]
    except Exception as e:  # noqa: BLE001
        logger.warning("Could not list models from %s: %s", LLM_BASE_URL, e)

    chat_models = [m for m in ids if not NON_CHAT_MODEL_ID.search(m)]
    if LLM_MODEL in chat_models:
        chat_models.remove(LLM_MODEL)
    ordered = [LLM_MODEL] + sorted(chat_models)
    return ordered or [LLM_MODEL]


SYSTEM_PROMPT = textwrap.dedent("""\
    You are Hope, a warm, conversational voice assistant speaking with
    {user_display_name}. You are talking TO someone, not writing for them to
    read. Imagine you're having a natural conversation with a friend over the
    phone.

    # How you speak
    - Keep every response to one or three sentences. That's it.
    - Use contractions (I'm, don't, it's) and natural phrasing.
    - Speak like you're talking, not writing. No bullet points, no lists,
      no markdown, and no formatting except the Azure delivery markers below.
    - Spell out numbers when it sounds more natural ("twenty twenty-six"
      instead of "2026").
    - Make reasonable assumptions and answer directly once any relevant tool
      call has been made. Ask a follow-up only when it is genuinely
      impossible to proceed without one.
    - Be warm and direct. Don't be sycophantic or overly formal.
    - If you don't know something, say so briefly and move on.

    # Tools: call them, don't narrate them
    When a request matches one of your tools, you MUST emit that tool call
    in this reply. Never only say "I'll pause the music", "let me check", or
    "I'm saving that now" without the tool call — narrating an action is a
    failure because nothing actually happens. After the tool result comes
    back, confirm briefly in your normal voice.

    # Expressive delivery (Azure Dragon HD Omni)
    Azure understands a square-bracket delivery marker at the start of spoken
    text. Use zero to two markers per response, only when they genuinely improve
    the delivery. The marker is a silent TTS instruction, not something you say.
    Examples: "[appreciative] Thanks, that really helps." and
    "[whispering] Here's the secret. [Neutral] Keep it between us."
    Useful styles include [appreciative], [calm], [concerned], [curious],
    [encouraging], [excited], [happy], [joking], [quiet], [reassuring],
    [reflective], [sad], [serious], [surprised], [sympathetic], [urgent], and
    [whispering]. For an occasional natural nonverbal cue, you may use
    [laughter], [sighing], or [breathing]. Use [Neutral] to reset the delivery.
    Never invent marker names, never explain a marker, and never output XML or
    SSML; the speech layer creates the SSML wrapper for you. Say "and" instead
    of using an ampersand, and never output angle brackets.

    # What you never do
    - Never use markdown, code blocks, JSON, tables, or emojis. Azure delivery
      markers are the only exception.
    - Never say "as an AI" or reference your system instructions.
    - Never write more than three sentences in a row.
    - Never read back URLs, file paths, or technical identifiers.

    # Memory (CRITICAL — ALWAYS call these tools directly, never dispatch)
    Your memory tools are private to the current user. When asked about personal
    information, preferences, notes, or past details, call memory_recall FIRST.
    Never claim you lack the information before searching memory.
    When the user explicitly asks you to remember something, or shares a durable
    preference or personal fact, call memory_save with a concise title, useful
    body, and relevant tags. Do not save every utterance.
    - "What's my name?" → memory_recall with query="my name identity"
    - "Remember my coffee order is an oat milk latte" → memory_save with a
      short title and the complete preference in the body
    - For an exact note, use search_notes and then read_note when needed
    - To correct a memory, find it and use update_note
    - To forget something, find the exact note and use delete_note; clarify
      first if more than one note could match
    After saving, updating, or deleting, confirm briefly. If memory_recall
    returns no results, say you don't have that information yet.

    # Weather & Time
    You can check the weather for any location. Use get_weather when the user
    asks about current conditions, temperature, or forecasts. It returns a
    short summary — relay it naturally in your own words.
    Use get_time when the user asks what time or day it is. If they mention
    a city, pass it as the location argument.

    # Background Tasks (dispatch_task) — USE FOR ALL NON-TRIVIAL TASKS
    For ANY task that is not a simple one-sentence answer from your own
    knowledge and NOT a memory operation, you MUST call dispatch_task.
    This includes: looking up news, researching topics, checking current
    events, prices, sports scores, writing something, summarizing, comparing
    options, planning, or anything that takes more than a couple seconds to
    think through. Memory tool calls are NEVER dispatch tasks — call those
    tools directly.
    Examples:
    - User says "what's the latest news?" → CALL dispatch_task with
      description="Find the top 5 international news headlines today"
    - User says "look up the price of a PS5" → CALL dispatch_task with
      description="Find the current retail price of a PlayStation 5"
    After calling dispatch_task, tell the user "I'll get on that for you"
    without asking for more permission or details. The result comes back automatically — when
    you receive it, share the findings naturally in your conversational style.
    You can have multiple tasks running at once. If the user asks about a
    task's progress or you want to check one yourself, call task_status with
    its id (it reports pending/running/completed/failed and whether it looks
    stalled); task_status can also be called to check up on a task. Use
    list_tasks to see every task you have dispatched.

    # Music (media player in the web UI)
    You can play music on the user's web player. For any music request, call
    search_music first, then:
    - One clearly-right result → play_song with its video_id. Announce it in
      one short sentence, then stop talking — playback pauses the
      conversation until the user comes back to you.
    - Ambiguous results → offer_song_choices with the top 2-3 options, name
      them briefly, and let the user pick by voice or in the player panel.
      Never start playback on a guess.
    Prefer what is already saved: search_music puts downloaded library
    matches first (in_library=true) — always pick one of those over a fresh
    YouTube result, and list_library shows the whole saved collection which
    plays instantly. If the user names a library track exactly, just play it.
    A brand-new song starts playing within seconds and finishes downloading
    in the background. Control playback with pause_music, resume_music,
    skip_song and seek_music; check what's on with now_playing.
    add_stream registers a live radio or direct audio URL as a library track.
    While music is playing you cannot hear the user — they will come back to
    you when they want to talk again, and the music pauses automatically.

    # Skills
    You have a skill library of learned procedures. Use skill_recall when you need
    to perform a task you've done before — it will give you the steps. When you
    successfully complete a multi-step task or learn a new procedure from the user,
    use skill_save to record it so you can follow it next time. Be selective: only
    save skills for repeatable tasks, not one-off facts (those go in memory).

""")


# ── Intent routing ──────────────────────────────────────────────────────────

try:
    from agent.task_router import should_force_background_task
    from agent.intent_router import detect_forced_call
except ImportError:
    from task_router import should_force_background_task
    from intent_router import detect_forced_call


# ── OpenAI-compatible LLM with reasoning_content fallback ───────────────────

# llama.cpp derives a GBNF grammar from tool schemas to constrain the model's
# tool-call output. Its parser rejects repetition counts >= 2000 (e.g. a string
# with maxLength 200000 emits "char{0,200000}" which fails to parse), so cap
# the length-style constraints the voice pipeline exposes. qwen2.5-vl-3b hits
# this; gemma happens to tolerate the same schemas.
_LLM_SCHEMA_LENGTH_CEILING = 1000
_LLM_SCHEMA_LENGTH_KEYS = ("minLength", "maxLength", "minItems", "maxItems")


def _sanitize_tool_schema(node: object) -> None:
    """Cap length-style constraints in a JSON-schema dict (mutates in place)."""
    if isinstance(node, dict):
        for key in _LLM_SCHEMA_LENGTH_KEYS:
            value = node.get(key)
            if isinstance(value, (int, float)) and value > _LLM_SCHEMA_LENGTH_CEILING:
                node[key] = _LLM_SCHEMA_LENGTH_CEILING
        for value in node.values():
            _sanitize_tool_schema(value)
    elif isinstance(node, list):
        for item in node:
            _sanitize_tool_schema(item)


class _ReasoningFallbackWrapper:
    """Wraps an LLMStream to fall back to ``reasoning_content`` when the model
    finishes a turn with empty visible content (reasoning models such as Gemma
    spend their whole budget on hidden reasoning). Delegates all iteration to
    the underlying stream and injects a final content chunk if no real content
    was produced."""

    def __init__(self, inner: lk_llm.LLMStream) -> None:
        self._inner = inner
        self._last_reasoning: str | None = None
        self._has_content = False

    @property
    def chat_ctx(self) -> lk_llm.ChatContext:
        return self._inner.chat_ctx

    @property
    def tools(self) -> list:
        return self._inner.tools

    async def aclose(self) -> None:
        await self._inner.aclose()

    async def __aenter__(self) -> "_ReasoningFallbackWrapper":
        return self

    async def __aexit__(self, *exc) -> None:
        await self.aclose()

    def __aiter__(self) -> "_ReasoningFallbackWrapper":
        return self

    async def __anext__(self) -> lk_llm.ChatChunk:
        try:
            chunk = await self._inner.__anext__()
        except StopAsyncIteration:
            # Stream exhausted — inject reasoning fallback if no content was produced
            if not self._has_content and self._last_reasoning:
                logger.warning(
                    "LLM returned empty content; falling back to reasoning_content"
                )
                return lk_llm.ChatChunk(
                    id="reasoning-fallback",
                    delta=lk_llm.ChoiceDelta(role="assistant", content=self._last_reasoning),
                )
            raise
        # Track reasoning_content from the chunk's delta if present
        delta = getattr(chunk, "delta", None)
        if delta is not None:
            reasoning = getattr(delta, "reasoning_content", None)
            if reasoning:
                self._last_reasoning = reasoning
            if chunk.has_response():
                self._has_content = True
            tc = getattr(delta, "tool_calls", None)
            if tc:
                logger.info("LLM tool_call: %s", [getattr(t, "name", str(t)) for t in tc])
        return chunk

    async def collect(self):
        return await self._inner.collect()


class _ForcedToolCallStream(lk_llm.LLMStream):
    """Emit a tool call without asking the small LLM to produce it.

    Two callers: the task router (dispatch_task for background work) and the
    intent router (deterministic intents like "pause the music" or "remember
    that ..."), where small models otherwise narrate the action without ever
    emitting the call.
    """

    def __init__(
        self,
        llm,
        *,
        chat_ctx,
        tools,
        conn_options,
        tool_name: str = "dispatch_task",
        arguments: dict | None = None,
    ):
        self._tool_name = tool_name
        self._arguments = arguments if arguments is not None else {}
        super().__init__(llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)

    async def _run(self) -> None:
        self._event_ch.send_nowait(
            lk_llm.ChatChunk(
                id=f"forced-{uuid.uuid4().hex}",
                delta=lk_llm.ChoiceDelta(
                    role="assistant",
                    tool_calls=[
                        lk_llm.FunctionToolCall(
                            name=self._tool_name,
                            arguments=json.dumps(self._arguments),
                            call_id=f"forced-{uuid.uuid4().hex}",
                        )
                    ],
                ),
            )
        )


class HopeLLM(openai.LLM):
    """OpenAI-compatible LLM (OpenWebUI) with runtime model switching."""

    # Context compaction: when the conversation grows past this many items,
    # ask the LLM to summarize the older messages into a compact recap, then
    # replace them with that summary. Keeps recent turns verbatim.
    COMPACTION_THRESHOLD = 24  # ~12 turns of user/agent exchange
    KEEP_RECENT = 10           # last N items kept verbatim after compaction
    IDLE_COMPACTION_SECONDS = 300  # also compact after 5 min of silence

    def __init__(self, *, model: str, **kwargs):
        # llama.cpp's Gemma tool template ignores named tool_choice when OpenAI
        # strict schemas are enabled. Non-strict schemas make forced dispatch
        # (and ordinary MCP calls) work reliably.
        kwargs.setdefault("_strict_tool_schema", False)
        self._model = model
        super().__init__(model=model, **kwargs)
        import time
        self._last_call_time: float = time.monotonic()

    @property
    def model(self) -> str:
        """Currently selected model id (openai.LLM exposes the same property)."""
        return self._model

    def update_options(self, *, model: str | None = None) -> None:
        """Switch the backing model for all subsequent chat calls.

        The base class re-reads ``self._opts`` on every request, so mutating
        the opts dataclass is enough — no client rebuild needed.
        """
        if model and model != self._model:
            logger.info("LLM model change: %s -> %s", self._model, model)
            self._model = model
            self._opts.model = model

    def chat(self, *, chat_ctx, tools=None, conn_options=None, **kwargs):
        if conn_options is None:
            from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS

            conn_options = DEFAULT_API_CONNECT_OPTIONS

        import time
        now = time.monotonic()
        idle_for = now - self._last_call_time
        self._last_call_time = now

        # Cap length-style constraints on tool schemas. llama.cpp builds a GBNF
        # grammar from these and rejects repetitions >= 2000 (qwen2.5-vl-3b
        # fails on schemas like the memory MCP's update_note body maxLength).
        for tool in tools or []:
            info = getattr(tool, "info", None)
            raw_schema = getattr(info, "raw_schema", None)
            if isinstance(raw_schema, dict):
                _sanitize_tool_schema(raw_schema)

        # Compact if context is large, OR if we've been idle for 5+ minutes
        # (user came back after a break — summarize the old conversation).
        ctx_len = len(chat_ctx.items)
        should_compact = ctx_len > self.COMPACTION_THRESHOLD
        if not should_compact and idle_for > self.IDLE_COMPACTION_SECONDS:
            should_compact = ctx_len > 6  # minimal context worth summarizing

        if should_compact:
            reason = "idle" if idle_for > self.IDLE_COMPACTION_SECONDS else "size"
            logger.info(
                "Compacting context (%s): %d items, idle %.0fs",
                reason, ctx_len, idle_for,
            )
            chat_ctx = self._compact_context(chat_ctx, conn_options)

        # Force calls for intents small models drift on. Without this they
        # say "I'll pause the music" or "I'll get on that" but emit no tool
        # call at all. Task router handles background work; the intent router
        # handles deterministic in-conversation intents (media controls,
        # memory saves, music searches).
        items = list(chat_ctx.items)
        latest = items[-1] if items else None
        latest_role = getattr(latest, "role", None)
        latest_text = self._msg_content_text(latest) if latest_role == "user" else ""
        dispatch_available = any(
            getattr(tool, "id", None) == "dispatch_task" for tool in (tools or [])
        )
        if dispatch_available and should_force_background_task(latest_text):
            logger.info("Routing background request to dispatch_task: %.120s", latest_text)
            return _ForcedToolCallStream(
                self,
                chat_ctx=chat_ctx,
                tools=tools or [],
                conn_options=conn_options,
                tool_name="dispatch_task",
                arguments={"description": latest_text},
            )

        forced = detect_forced_call(latest_text, tools or []) if latest_text else None
        if forced:
            tool_name, arguments = forced
            logger.info(
                "Routing request to forced %s(%s): %.120s",
                tool_name, arguments, latest_text,
            )
            return _ForcedToolCallStream(
                self,
                chat_ctx=chat_ctx,
                tools=tools or [],
                conn_options=conn_options,
                tool_name=tool_name,
                arguments=arguments,
            )

        stream = super().chat(
            chat_ctx=chat_ctx,
            tools=tools,
            conn_options=conn_options,
            **kwargs,
        )
        return _ReasoningFallbackWrapper(stream)

    def _compact_context(self, chat_ctx, conn_options):
        """Summarize older messages via a side LLM call, keep recent ones.

        Splits the context into [system | old | recent]. The old portion is
        summarized by the LLM into a short recap message that replaces it.
        Falls back to hard truncation if the summary call fails.
        """
        from livekit.agents import llm as lk_llm

        items = list(chat_ctx.items)
        # Identify the system/instruction message (first item usually)
        sys_items = [it for it in items if getattr(it, "role", None) in ("system", "developer")]
        non_sys = [it for it in items if getattr(it, "role", None) not in ("system", "developer")]

        if len(non_sys) <= self.KEEP_RECENT:
            return chat_ctx  # nothing to compact

        old_msgs = non_sys[: -self.KEEP_RECENT]
        recent_msgs = non_sys[-self.KEEP_RECENT :]

        # Build a transcript of the old messages for the summarizer
        transcript_parts = []
        for msg in old_msgs:
            role = getattr(msg, "role", "user")
            content = self._msg_content_text(msg)
            if content:
                transcript_parts.append(f"{role}: {content}")
        transcript = "\n".join(transcript_parts)

        if not transcript.strip():
            return chat_ctx  # nothing meaningful to summarize

        # Side call to the LLM for the summary (no tools, no streaming needed)
        try:
            import httpx

            summary_prompt = (
                "Summarize this conversation history in 2-4 sentences. "
                "Preserve key facts, decisions, and user preferences. "
                "Be concise — this replaces the full history as context.\n\n"
                + transcript
            )
            headers = {"Content-Type": "application/json"}
            if LLM_API_KEY and LLM_API_KEY != "not-needed":
                headers["Authorization"] = f"Bearer {LLM_API_KEY}"
            resp = httpx.post(
                f"{LLM_BASE_URL}/chat/completions",
                headers=headers,
                json={
                    "model": self._model,
                    "messages": [
                        {"role": "system", "content": "You summarize conversations concisely."},
                        {"role": "user", "content": summary_prompt},
                    ],
                    "max_tokens": 200,
                    "temperature": 0.3,
                    "chat_template_kwargs": {"enable_thinking": False},
                },
                timeout=30,
            )
            resp.raise_for_status()
            data = resp.json()
            summary_text = (data.get("choices") or [{}])[0].get("message", {}).get("content", "")
        except Exception as e:  # noqa: BLE001
            logger.warning("Context compaction failed, falling back to truncation: %s", e)
            chat_ctx.truncate(max_items=self.COMPACTION_THRESHOLD)
            return chat_ctx

        if not summary_text.strip():
            chat_ctx.truncate(max_items=self.COMPACTION_THRESHOLD)
            return chat_ctx

        # Rebuild the context: system + summary + recent
        new_items = list(sys_items)
        new_items.append(
            lk_llm.ChatMessage(
                role="user",
                content=[f"[Conversation so far: {summary_text.strip()}]"],
            )
        )
        new_items.append(
            lk_llm.ChatMessage(
                role="assistant",
                content=["Got it, I'll keep that in mind."],
            )
        )
        new_items.extend(recent_msgs)

        logger.info(
            "Context compacted: %d items -> %d (summary: %.80s...)",
            len(items),
            len(new_items),
            summary_text.strip(),
        )

        return lk_llm.ChatContext(new_items)

    @staticmethod
    def _msg_content_text(msg) -> str:
        """Extract plain text from a ChatMessage's content field."""
        content = getattr(msg, "content", None)
        if content is None:
            return ""
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts = []
            for c in content:
                if isinstance(c, str):
                    parts.append(c)
                elif hasattr(c, "text"):
                    parts.append(c.text)
            return " ".join(parts)
        return str(content)


# ── MCP toolsets (web access + any extra configured servers) ────────────────

def build_mcp_toolsets(user_profile: dict, task_dir: str) -> list[mcp.MCPToolset]:
    """Build isolated MCP toolsets for one authenticated voice user/session."""
    toolsets: list[mcp.MCPToolset] = []

    python_bin = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".venv", "bin", "python")
    weather_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weather_mcp.py")
    toolsets.append(
        mcp.MCPToolset(
            id="weather",
            mcp_server=mcp.MCPServerStdio(
                command=python_bin,
                args=[weather_mcp_script],
                env={**os.environ},
                client_session_timeout_seconds=30,
            ),
        )
    )
    logger.info("Weather MCP toolset enabled (wttr.in)")

    memory = user_profile["memory"]
    memory_url = memory["url"]
    allowed_memory_tools = memory.get(
        "allowed_tools",
        [
            "create_note",
            "read_note",
            "update_note",
            "delete_note",
            "list_notes",
            "search_notes",
            "memory_save",
            "memory_recall",
        ],
    )
    toolsets.append(
        mcp.MCPToolset(
            id="memory",
            mcp_server=mcp.MCPServerHTTP(
                url=memory_url,
                transport_type="streamable_http",
                allowed_tools=allowed_memory_tools,
                headers={"Authorization": f"Bearer {memory['api_key']}"},
                client_session_timeout_seconds=60,
            ),
        )
    )
    logger.info(
        "Memory MCP enabled for user %s at %s",
        user_profile["user_id"],
        memory_url,
    )

    skills_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "skills_mcp.py")
    toolsets.append(
        mcp.MCPToolset(
            id="skills",
            mcp_server=mcp.MCPServerStdio(
                command=python_bin,
                args=[skills_mcp_script],
                env={**os.environ, "SKILLS_DIR": SKILLS_DIR},
                client_session_timeout_seconds=30,
            ),
        )
    )
    logger.info("Skills MCP toolset enabled (%s)", SKILLS_DIR)

    dispatch_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dispatch_mcp.py")
    toolsets.append(
        mcp.MCPToolset(
            id="dispatch",
            mcp_server=mcp.MCPServerStdio(
                command=python_bin,
                args=[dispatch_mcp_script],
                env={**os.environ, "TASKS_DIR": task_dir},
                client_session_timeout_seconds=60,
            ),
        )
    )
    logger.info("Dispatch MCP toolset enabled (%s)", task_dir)

    # Media tools talk to the local media server; the room name scopes player
    # state so concurrent sessions don't fight over one virtual player.
    media_server_url = os.environ.get("MEDIA_SERVER_URL", "http://127.0.0.1:8093")
    media_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "media_mcp.py")
    toolsets.append(
        mcp.MCPToolset(
            id="media",
            mcp_server=mcp.MCPServerStdio(
                command=python_bin,
                args=[media_mcp_script],
                env={
                    **os.environ,
                    "MEDIA_SERVER_URL": media_server_url,
                    "MEDIA_ROOM": os.path.basename(task_dir.rstrip("/")) or "default",
                },
                client_session_timeout_seconds=60,
            ),
        )
    )
    logger.info("Media MCP toolset enabled (%s)", media_server_url)

    extra = os.environ.get("EXTRA_MCP_SERVERS", "")
    if extra:
        try:
            for i, entry in enumerate(json.loads(extra)):
                url = entry.get("url", "")
                transport = entry.get("transport")  # "sse" | "streamable_http" | None (auto)
                toolsets.append(
                    mcp.MCPToolset(
                        id=f"extra-{i}",
                        mcp_server=mcp.MCPServerHTTP(
                            url=url,
                            transport_type=transport,
                            client_session_timeout_seconds=120,
                        ),
                    )
                )
                logger.info("Extra MCP toolset: %s (%s)", url, transport or "auto")
        except (json.JSONDecodeError, TypeError) as e:
            logger.error("Failed to parse EXTRA_MCP_SERVERS: %s", e)

    return toolsets


class VoiceAssistant(Agent):
    """The conversational agent. LLM is the brain; STT/TTS are senses."""

    def __init__(self, user_profile: dict, task_dir: str) -> None:
        display_name = user_profile.get("display_name") or user_profile["user_id"]
        instructions = SYSTEM_PROMPT.replace("{user_display_name}", display_name)
        super().__init__(
            llm=HopeLLM(
                model=LLM_MODEL,
                base_url=LLM_BASE_URL,
                api_key=LLM_API_KEY,
                max_completion_tokens=1000,
                extra_body={"chat_template_kwargs": {"enable_thinking": False}},
            ),
            instructions=instructions,
            tools=build_mcp_toolsets(user_profile, task_dir),
        )


# ── Agent server ────────────────────────────────────────────────────────────
# Bind the worker's internal health server to an ephemeral port because this
# container uses host networking and Unraid already has services on 8081.
server = AgentServer(port=0)


@server.rtc_session(agent_name="voice-assistant")
async def handle_job(ctx: JobContext) -> None:
    logger.info("Job started for room %s", ctx.room.name)
    try:
        user_id = user_id_from_room(ctx.room.name)
        user_profile = load_user_profile(user_id)
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        logger.error("Rejecting voice job: %s", exc)
        return
    logger.info("Resolved voice user %s", user_id)

    # Every room gets a private event directory. This prevents concurrent or
    # reconnecting sessions from consuming and announcing each other's tasks.
    task_dir = f"/tmp/tasks/{ctx.room.name}"

    # Azure STT — streaming, reads AZURE_SPEECH_KEY / AZURE_SPEECH_REGION from env
    stt = azure.STT(
        speech_key=AZURE_KEY,
        speech_region=AZURE_REGION,
        language=["en-US"],
    )

    # Azure TTS — wraps text in SSML; Dragon HD Omni interprets inline delivery
    # markers such as [appreciative] and [Neutral]. Outputs 24kHz PCM.
    tts = azure.TTS(
        voice=DEFAULT_VOICE,
        sample_rate=24000,
        speech_key=AZURE_KEY,
        speech_region=AZURE_REGION,
    )

    session = AgentSession(
        stt=stt,
        tts=tts,
        vad=inference.VAD(model="silero"),
        turn_handling=TurnHandlingOptions(
            # VAD-based turn detection: agent waits for user to stop speaking.
            # ("adaptive" mode requires the LiveKit Cloud barge-in service.)
            interruption={"mode": "vad"},
            # Preemptive generation is incompatible with tool calls — it starts
            # generating before the turn finalizes, breaking the MCP execution loop.
            preemptive_generation={"enabled": False},
        ),
    )

    # The assistant owns the LLM instance; data-channel handlers below reach
    # into it for live model switching.
    voice_assistant = VoiceAssistant(user_profile, task_dir)

    # Publish the chat-capable model list so the web UI can offer a picker.
    async def _publish_models() -> None:
        models = await list_available_models()
        payload = json.dumps(
            {
                "type": "models",
                "models": models,
                "selected": voice_assistant.llm.model,
            }
        )
        try:
            await ctx.room.local_participant.publish_data(
                payload, reliable=True, topic="voice-control"
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("failed to publish model list: %s", e)

    # Listen for conversation controls (voice selection and voice/text mode).
    # Defined here so it captures this job's session via closure.
    # Music and conversation are mutually exclusive by design: while a track
    # plays the agent's ears are off; re-engaging the conversation pauses the
    # music, and ending the conversation lets the user resume it.
    session_mode = {"value": "voice"}       # UI voice/text switch
    media_playback = {"value": "stopped"}   # UI-reported player state

    def _apply_listening() -> None:
        listening = (
            session_mode["value"] == "voice"
            and media_playback["value"] != "playing"
        )
        session.input.set_audio_enabled(listening)

    def _on_room_data(packet) -> None:
        topic = getattr(packet, "topic", None)
        if topic and topic not in ("voice-control", "conversation-control", "media-control"):
            return
        try:
            msg = json.loads(packet.data.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError):
            return

        if msg.get("type") == "set_mode":
            mode = msg.get("mode")
            if mode not in ("voice", "text"):
                return
            voice_enabled = mode == "voice"
            logger.info("Switching conversation mode to %s", mode)
            session_mode["value"] = mode
            if not voice_enabled:
                # interrupt() returns an already-scheduled Future, not a coroutine.
                session.interrupt()
            session.output.set_audio_enabled(voice_enabled)
            _apply_listening()
            return

        if msg.get("type") == "media_state":
            state = msg.get("state")
            if state not in ("playing", "paused", "stopped"):
                return
            logger.info("Media playback -> %s (agent listening: %s)",
                        state, session_mode["value"] == "voice" and state != "playing")
            media_playback["value"] = state
            _apply_listening()
            return

        if msg.get("type") == "set_voice":
            voice = msg.get("voice", "")
            if not voice:
                return
            logger.info("Switching TTS voice to %s", voice)
            try:
                active_tts = session.tts
                if hasattr(active_tts, "update_options"):
                    active_tts.update_options(voice=voice)
                    logger.info("Voice updated to %s", voice)
            except Exception as e:  # noqa: BLE001
                logger.warning("Failed to update voice: %s", e)

        if msg.get("type") == "get_models":
            asyncio.create_task(_publish_models())
            return

        if msg.get("type") == "set_model":
            model = msg.get("model", "")
            if not model:
                return
            try:
                llm = voice_assistant.llm
                if hasattr(llm, "update_options"):
                    llm.update_options(model=model)
                    ack = json.dumps(
                        {"type": "model_changed", "model": llm.model}
                    )
                    asyncio.create_task(
                        ctx.room.local_participant.publish_data(
                            ack, reliable=True, topic="voice-control"
                        )
                    )
            except Exception as e:  # noqa: BLE001
                logger.warning("Failed to update model: %s", e)

    ctx.room.on("data_received", _on_room_data)

    # Publish user/agent transcripts to the room so the web UI can render them.
    async def publish_transcript(role: str, text: str) -> None:
        text = (text or "").strip()
        if not text:
            return
        payload = json.dumps({"type": "transcript", "role": role, "text": text})
        try:
            await ctx.room.local_participant.publish_data(
                payload, reliable=True, topic="transcript"
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("failed to publish transcript: %s", e)

    @session.on("conversation_item_added")
    def _on_conversation_item(ev) -> None:
        msg = ev.item
        role = getattr(msg, "role", None)
        text = getattr(msg, "text_content", None)
        if role == "user":
            asyncio.create_task(publish_transcript("user", text))
        elif role == "assistant":
            asyncio.create_task(publish_transcript("agent", text))

    @session.on("user_input_transcribed")
    def _on_user_transcribed(ev) -> None:
        logger.info("STT (%s): %s", "final" if ev.is_final else "partial", ev.transcript)

    @session.on("user_state_changed")
    def _on_user_state(ev) -> None:
        # Fires from VAD: if this never says "speaking", no usable mic
        # audio is arriving from the participant.
        logger.info("user state -> %s", ev.new_state)

    # ── Background task events → UI and spoken result ──────────────────────
    try:
        from agent.task_registry import TaskRegistry
    except ImportError:
        from task_registry import TaskRegistry
    task_registry = TaskRegistry(task_dir)

    async def _publish_task_event(task_id: str, event: dict) -> None:
        payload = json.dumps({"type": "task_event", "task_id": task_id, **event})
        try:
            await ctx.room.local_participant.publish_data(
                payload, reliable=True, topic="tasks"
            )
        except Exception as e:  # noqa: BLE001
            logger.warning("failed to publish task event: %s", e)

    def _announce_task_result(task_id: str, event: dict) -> None:
        event_type = event.get("event")
        if event_type == "completed":
            result = str(event.get("result") or "").strip()
            if result:
                logger.info("Announcing completed task %s: %.160s", task_id, result)
                session.say(result, allow_interruptions=True, add_to_chat_ctx=True)
        elif event_type == "failed":
            logger.warning("Background task %s failed: %s", task_id, event.get("error"))
            session.say(
                "I couldn't finish that background task. You can ask me to try again.",
                allow_interruptions=True,
                add_to_chat_ctx=True,
            )

    async def _poll_tasks() -> None:
        while True:
            try:
                await asyncio.sleep(0.5)
                events = await task_registry.poll_events()
                for task_id, event in events:
                    await _publish_task_event(task_id, event)
                    _announce_task_result(task_id, event)
            except asyncio.CancelledError:
                raise
            except Exception:  # noqa: BLE001
                # A transient room or speech error must not permanently kill
                # result delivery for every later task.
                logger.exception("Background task event poller failed")

    poll_task = asyncio.create_task(_poll_tasks(), name=f"task-poller-{ctx.room.name}")

    async def _stop_task_poller(*_args) -> None:
        poll_task.cancel()
        await asyncio.gather(poll_task, return_exceptions=True)

    ctx.add_shutdown_callback(_stop_task_poller)

    await session.start(
        agent=voice_assistant,
        room=ctx.room,
        room_options=room_io.RoomOptions(
            # Keep the agent in the room when a participant leaves; it must
            # survive page reloads/reconnects, otherwise the next join lands
            # in an agent-less room (LiveKit does not reliably re-dispatch
            # into an existing room).
            close_on_disconnect=False,
            audio_input=room_io.AudioInputOptions(
                # No noise cancellation plugin (self-hosted, no ai-coustics)
            ),
        ),
    )

    await ctx.connect()
    # The participant can appear in the browser slightly before AgentSession is
    # ready to accept lk.chat streams. Explicitly signal readiness so the first
    # typed message is never raced against session startup.
    try:
        await ctx.room.local_participant.publish_data(
            json.dumps({"type": "agent_ready"}),
            reliable=True,
            topic="conversation-control",
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("failed to publish agent readiness: %s", exc)

    # Announce which models the UI can switch between (selected one included).
    await _publish_models()
    logger.info("Agent connected to room %s", ctx.room.name)


if __name__ == "__main__":
    cli.run_app(server)