hope-voice-api/agent/intent_router.py

"""Deterministic intent router — forces tool calls for high-frequency intents.

Small local models drift under long context: they narrate an action ("Sure,
I'll pause the music") without emitting the tool call, which is exactly the
failure mode _ForcedToolCallStream was built to bypass for dispatch_task.
This module extends the same idea to the other unambiguous, high-frequency
voice intents: music playback controls, now-playing/library questions,
"remember that ..." memory saves, and explicit YouTube Music searches.

Conservatism rules:
- Never fires when the matching tool is not available in the current
  session's toolset (MCP toolsets differ per user config).
- Never fires on explanation/questions ("how do I pause the music?") for
  action intents — those deserve a spoken answer, not a side effect.
- "play <something>" only forces a search for what looks like a song name;
  generic objects ("play some music", "play a game") fall through to the
  LLM.

Returns (tool_name, arguments) or None. The agent emits the call directly;
the tool result then flows back through the normal loop, so Hope still
speaks a natural confirmation afterwards.
"""
from __future__ import annotations

import re


# Question/explanation guard for *action* intents (not for question intents
# like now_playing, which are questions by design).
_QUESTION_GUARD = re.compile(
    r"\b(how (?:do|can) i|how to|what happens if|what does|can you explain|"
    r"tell me about|why (?:do|does|is)|is there a way to)\b",
    re.I,
)

_MEDIA_CONTROLS: list[tuple[re.Pattern[str], str]] = [
    (re.compile(r"\bpause\b|\bput .* on hold\b", re.I), "pause_music"),
    (re.compile(
        r"\bstop\b(?:\s+(?:the|this|that))?\s*(?:music|song|playback|player|tune)s?\b",
        re.I,
    ), "stop_music"),
    (re.compile(r"\b(?:resume|unpause)\b", re.I), "resume_music"),
    (re.compile(
        r"\b(?:skip|next)\b.*\b(?:song|track|one)\b|\bskip\s+(?:this|it|that)\b",
        re.I,
    ), "skip_song"),
    (re.compile(r"\bplay\b.*\bagain\b|\bfrom the (?:top|beginning)\b", re.I), "resume_music"),
]

_NOW_PLAYING = re.compile(
    r"\bwhat(?:'s| is|s)\s+(?:currently\s+)?playing\b"
    r"|\bwhat song is (?:this|playing)\b"
    r"|\bwhich song is (?:this|playing)\b"
    r"|\bname this song\b"
    r"|\bwhat(?:'s| is) this (?:track|tune)\b",
    re.I,
)

_LIBRARY = re.compile(
    r"\bwhat(?:'s| is)?\s*(?:in|of)\s+my\s+(?:music\s+)?library\b"
    r"|\b(?:list|show)(?:\s+me)?\s+my\s+(?:music|library|songs|tracks)\b",
    re.I,
)

_MEMORY_SAVE = re.compile(
    r"^\s*(?:please\s+)?remember\s+(?:that\s+|this\s*[::]\s*)?(?P<body>.+?)\s*[.!]?\s*$",
    re.I,
)

# Personal-fact recall. Excluded when the utterance is really a
# weather/time/date question — those tools already work lexically on small
# models and must win.
_MEMORY_RECALL = re.compile(
    r"\bwhat(?:'s| is|s)\s+my\b"
    r"|\bwhat(?:'s| is)\s+(?:the\s+)?(?:name|birthday|anniversary)\b.*\bmy\b"
    r"|\bdo you remember\b"
    r"|\bwhat did i (?:say|tell you|mention)\b"
    r"|\bwho am i\b"
    r"|\bwhere (?:do|did) i (?:live|work|leave)\b",
    re.I,
)
_RECALL_EXCLUDE = re.compile(
    r"\b(?:weather|temperature|forecast|time|date|day is it|clock)\b", re.I
)

_SEARCH_NOTES = re.compile(
    r"\bdo i have (?:any |some )?notes\b"
    r"|\b(?:check|search|look (?:in|through|at))\s+my\s+notes\b"
    r"|\bsearch\s+notes\b",
    re.I,
)

_SKILL_RECALL = re.compile(
    r"\bdo i have (?:a|any) (?:skill|procedure)\b"
    r"|\b(?:check|search)\s+(?:your|my)\s+skills\b"
    r"|\bany skills (?:for|about)\b",
    re.I,
)

_SEARCH_EXPLICIT = re.compile(
    r"\b(?:find|search)\s+(?:for\s+)?(?P<q>.+?)\s+on\s+(?:youtube|yt)\s*music\b"
    r"|\bsearch\s+(?:youtube\s+music\s+)?for\s+(?P<q2>.+?)\s*(?:on\s+youtube\s*music)?\s*$",
    re.I,
)

_PLAY_IMPERATIVE = re.compile(r"^\s*(?:could you\s+|please\s+)?play\s+(?P<q>.+?)\s*[.!]?\s*$", re.I)

# "play <q>" where q is not a song request — fall through to the LLM.
_GENERIC_PLAY = re.compile(
    r"^(?:some |any )?(?:music|something|anything|tunes?|songs?|a game|games?|the radio|it|that|one)$",
    re.I,
)


def _tool_names(tool) -> list[str]:
    names = []
    for attr in ("name", "id"):
        value = getattr(tool, attr, None)
        if isinstance(value, str) and value:
            names.append(value)
    return names


def tool_available(tools, name: str) -> bool:
    """True if a tool with this name is in the session's toolset.

    Handles MCP name mangling (server prefix) by comparing the last
    __-separated segment.
    """
    for tool in tools or []:
        for candidate in _tool_names(tool):
            if candidate == name or candidate.split("__")[-1] == name:
                return True
    return False


def _title_from(body: str) -> str:
    words = body.split()
    title = " ".join(words[:6]).strip(" .,!?;:")
    return (title[:60] or body[:60]).strip()


def detect_forced_call(text: str, tools) -> tuple[str, dict] | None:
    """Match a user utterance to a forced tool call.

    Returns (tool_name, arguments) when an unambiguous high-frequency intent
    matches an available tool, else None. Media controls win over memory and
    search so "pause" is never eaten by a wider pattern.
    """
    if not text or not text.strip():
        return None
    utterance = text.strip()
    lowered = utterance.lower()

    # ── question intents (no action guard — they ARE questions) ────────────
    if tool_available(tools, "now_playing") and _NOW_PLAYING.search(utterance):
        return ("now_playing", {})
    if tool_available(tools, "list_library") and _LIBRARY.search(utterance):
        return ("list_library", {})

    guarded = bool(_QUESTION_GUARD.search(utterance))

    # ── media playback controls ─────────────────────────────────────────────
    if not guarded:
        for pattern, tool_name in _MEDIA_CONTROLS:
            if pattern.search(utterance) and tool_available(tools, tool_name):
                return (tool_name, {})

    # ── memory save ─────────────────────────────────────────────────────────
    if not guarded and tool_available(tools, "memory_save"):
        m = _MEMORY_SAVE.match(utterance)
        if m:
            body = m.group("body").strip()
            if len(body) >= 3:
                return ("memory_save", {
                    "title": _title_from(body),
                    "body": body,
                    "tags": ["voice"],
                })

    # ── YouTube Music search ────────────────────────────────────────────────
    if not guarded and tool_available(tools, "search_music"):
        m = _SEARCH_EXPLICIT.search(utterance)
        query = (m.group("q") or m.group("q2") or "").strip() if m else None
        if not query:
            m = _PLAY_IMPERATIVE.match(utterance)
            if m:
                candidate = m.group("q").strip()
                if not _GENERIC_PLAY.match(candidate):
                    query = candidate
        if query:
            return ("search_music", {"query": query})

    # ── recall (memory / notes / skills) ────────────────────────────────────
    # The whole utterance is passed as the query — the memory and skills
    # backends do their own semantic/keyword matching.
    if not guarded:
        if (tool_available(tools, "memory_recall")
                and _MEMORY_RECALL.search(utterance)
                and not _RECALL_EXCLUDE.search(utterance)):
            return ("memory_recall", {"query": utterance})
        if tool_available(tools, "search_notes") and _SEARCH_NOTES.search(utterance):
            return ("search_notes", {"query": utterance})
        if tool_available(tools, "skill_recall") and _SKILL_RECALL.search(utterance):
            return ("skill_recall", {"query": utterance})

    return None