FabledGrimoire/src/fabledassistant/services/intent.py

"""Intent routing — classify user message before streaming.

Makes a fast non-streaming LLM call to detect tool intent and extract
parameters.  When a tool call is detected the caller can execute it
directly, bypassing the model's native (and sometimes unreliable)
structured tool-calling API.
"""

import json
import logging
import re
from dataclasses import dataclass, field
from datetime import date as date_type

from fabledassistant.services.llm import generate_completion

logger = logging.getLogger(__name__)


@dataclass
class IntentResult:
    tool_name: str | None = None  # None = no tool, just chat
    arguments: dict = field(default_factory=dict)
    confidence: str = "high"  # "high", "medium", or "low"
    ack: str | None = None  # One-sentence acknowledgment to stream immediately

    @property
    def should_execute(self) -> bool:
        """True if a tool was identified with sufficient confidence."""
        return self.tool_name is not None and self.confidence != "low"


def _build_tool_summary(tools: list[dict]) -> str:
    """Build a compact tool description string from Ollama tool defs."""
    lines: list[str] = []
    for tool in tools:
        fn = tool.get("function", {})
        name = fn.get("name", "")
        desc = fn.get("description", "")
        params = fn.get("parameters", {}).get("properties", {})
        required = set(fn.get("parameters", {}).get("required", []))

        param_parts: list[str] = []
        for pname, pinfo in params.items():
            req = " (required)" if pname in required else ""
            pdesc = pinfo.get("description", "")
            param_parts.append(f"  - {pname}: {pdesc}{req}")

        lines.append(f"- {name}: {desc}")
        lines.extend(param_parts)
    return "\n".join(lines)


_SYSTEM_PROMPT_TEMPLATE = """\
You are an intent classifier. Given a user message (and recent conversation \
history for context), decide whether it requires calling one of the available \
tools or is just general chat.

Today's date is {today}.

Available tools:
{tool_summary}

Respond with ONLY a JSON object, no other text:
- If a tool should be called: {{"tool": "tool_name", "arguments": {{...}}, "confidence": "high"|"medium"|"low", "ack": "One short sentence describing what you're about to do."}}
- If it's general chat: {{"tool": null, "confidence": "high", "ack": null}}

Confidence levels:
- "high": the intent is clear and all required arguments are unambiguous
- "medium": probably requires the tool but some argument is uncertain or inferred
- "low": uncertain whether this needs a tool at all, or the message is too ambiguous to act on

Rules:
- Use recent conversation history to resolve references like "it", "that event", "the meeting", "move it", etc.
- For dates like "tomorrow", "next Friday", "in 3 days", resolve them to YYYY-MM-DD format.
- For datetime parameters, use ISO 8601 format (e.g. 2026-09-30T14:00:00).
- Only include arguments the user actually specified or that can be clearly inferred.
- Infer reasonable defaults: birthdays and holidays are all-day + yearly recurring; "weekly meeting" is weekly recurring.
- Use descriptive titles: "My Birthday" not just "Birthday", "Team Standup" not just "Meeting".
- "add to", "update", "edit", "expand", "flesh out", "modify", "append to", "continue writing" a note → use update_note with query=<note title from context> and mode="append" for additions or mode="replace" for full rewrites.
- "mark as done", "complete", "finish", "mark in progress", "start" a task → use update_note with status field.
- "set priority", "change priority", "make it high priority" → use update_note with priority field.
- "set due date", "move due date", "due on Friday" for a task → use update_note with due_date field.
- "what are my overdue tasks", "show overdue", "tasks due today", "high priority tasks", "in progress tasks", "what's due this week" → use list_tasks with appropriate status/priority/due_before/due_after filters. For overdue, set due_before to today's date.
- If a note was created earlier in the conversation and the user provides more content for it, use update_note (not create_note).
- Use create_note ONLY when the user explicitly wants a brand new note that doesn't already exist.
- "update", "change", "rename", "set due date on" a CalDAV todo → use update_todo.
- "update", "change", "move", "reschedule" an event → use update_event.
- "delete", "cancel", "remove" an event → use delete_event.
- "which calendars", "list calendars", "my calendars" → use list_calendars.
- "create a calendar todo", "add a CalDAV todo" → use create_todo. Default to create_task for general tasks unless the user explicitly mentions CalDAV or calendar todo.
- "show calendar todos", "list calendar todos" → use list_todos.
- "find calendar todo", "search calendar todos", "find todo named X" → use search_todos with query=<keyword>.
- "completed/finished calendar todo" → use complete_todo.
- "delete/remove calendar todo" → use delete_todo.
- "remind me X minutes/hours before" an event → convert to reminder_minutes parameter on create_event.
- When the user asks about events in a time period (e.g. "events in September", "what's on next week", "my schedule for March"), use list_events with date_from/date_to covering that period. Do NOT use search_events for time-based queries — search_events is only for keyword matching against event titles.
- "delete", "remove", "trash", "get rid of" a note (not a task) → use delete_note with query=<note name/keyword>. NEVER use this for tasks.
- "delete", "remove", "trash", "get rid of" a task → use delete_task with query=<task name/keyword>. NEVER use this for notes.
- "read", "open", "show me", "what does X say", "display", "pull up" a specific note → use get_note with query=<note name>.
- "list my notes", "show notes", "recent notes", "browse notes", "notes tagged X" → use list_notes (with optional q or tags).
- "tag X with Y", "add tag Y to X", "untag Y from X", "remove tag Y from X" → use update_note with tags=[Y] and tag_mode="add" or "remove".
- search_web: user explicitly wants a quick factual answer from the web — current events, version
  numbers, real-time facts ("what is the latest version of X", "who won the game last night").
  ONLY use when the information is clearly not in their notes and they need something fresh from
  the internet. Do NOT use for creative questions, brainstorming, game design, writing help, or
  when the user is building on content they already have. Do NOT use when the user references
  their own notes or prior research — use search_notes instead.
- research_topic: user wants a comprehensive, multi-section research note created from web sources.
  Use whenever the user wants to deeply understand, learn about, or get a full written reference
  on any subject — regardless of how they phrase it. The topic can be anything: technical subjects,
  shopping decisions, comparisons, how things work, historical topics, etc.
  The "topic" argument should capture the full subject matter of the request.
  Prefer this over search_web when the user's request implies wanting thorough coverage rather than
  a quick answer.
- For creative/ideation requests ("think of", "come up with", "ideas for", "help me design",
  "imagine", "brainstorm") use null (chat) — the main model answers these directly. Only route
  to a tool if the user also explicitly asks to search, look something up, or create/save something.
- If the user references notes or research the assistant previously created, prefer search_notes
  (to retrieve the relevant note) or null (chat) so the main model can use its tools to find it.
  Never use search_web when existing notes likely contain the answer.
- "ack": one short, natural sentence confirming the action (tool path only). Vary phrasing — do not always start with "Let me". Omit (null) for chat-only responses.
- Do NOT wrap the JSON in markdown code fences."""


# Fast-path: "Research: <topic>" sent by the Research button.
_RESEARCH_PREFIX = re.compile(r"^[Rr]esearch:\s+(.+)", re.DOTALL)

# When the user refers to work the assistant previously did, they want the main
# model to answer using existing context — not a web search.  Skip intent and
# fall through to the streaming path so the model can call search_notes itself.
_PRIOR_WORK_REFS = re.compile(
    r"\b("
    r"research (you|that you|we) did"
    r"|research (you|that you|we) (made|created|compiled|wrote)"
    r"|note (you|that you|we) (made|created|wrote)"
    r"|notes (you|that you|we) (made|created|wrote)"
    r"|using (your|the) research"
    r"|based on (your|the) research"
    r"|from (your|the) (research|note|notes)"
    r")\b",
    re.IGNORECASE,
)


async def classify_intent(
    user_message: str,
    tools: list[dict],
    model: str,
    history: list[dict] | None = None,
) -> IntentResult:
    """Classify user intent via a fast non-streaming LLM call.

    history is a list of recent {role, content} messages (user/assistant only,
    no system messages) for resolving anaphoric references.

    Returns an IntentResult. On any failure, returns IntentResult(tool_name=None)
    so the caller falls through to the normal streaming path.
    """
    if not tools:
        return IntentResult()

    # Fast-path: "Research: <topic>" is the canonical format sent by the Research
    # button in the UI. It always means research_topic — skip the LLM call entirely.
    valid_names = {t.get("function", {}).get("name") for t in tools}
    if "research_topic" in valid_names:
        m = _RESEARCH_PREFIX.match(user_message.strip())
        if m:
            topic = m.group(1).strip()
            logger.info("Intent fast-path: 'Research:' prefix → research_topic, topic='%s'", topic[:80])
            return IntentResult(
                tool_name="research_topic",
                arguments={"topic": topic},
                confidence="high",
                ack=f"I'll research that and compile a comprehensive note.",
            )

    # Fast-path: user references prior assistant work ("research you did", "note you
    # made", etc.) — this is a request to use existing content, not search the web.
    # Return no-tool so the main model answers conversationally with search_notes
    # available if it needs to retrieve the note.
    if _PRIOR_WORK_REFS.search(user_message):
        logger.info(
            "Intent fast-path: prior-work reference detected → skipping tool dispatch"
        )
        return IntentResult()

    tool_summary = _build_tool_summary(tools)
    today = date_type.today().isoformat()

    messages: list[dict] = [
        {
            "role": "system",
            "content": _SYSTEM_PROMPT_TEMPLATE.format(
                today=today, tool_summary=tool_summary
            ),
        },
    ]

    # Inject recent history turns so the model can resolve references
    if history:
        for turn in history:
            role = turn.get("role")
            content = turn.get("content", "")
            if role in ("user", "assistant") and content:
                messages.append({"role": role, "content": content})

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

    try:
        raw = await generate_completion(messages, model, max_tokens=350, num_ctx=4096)
    except Exception:
        logger.warning("Intent classification LLM call failed", exc_info=True)
        return IntentResult()

    return _parse_intent(raw, tools)


def _parse_intent(raw: str, tools: list[dict]) -> IntentResult:
    """Parse the LLM's JSON response into an IntentResult."""
    text = raw.strip()

    # Strip markdown code fences if present
    text = re.sub(r"^```(?:json)?\s*", "", text)
    text = re.sub(r"\s*```$", "", text)
    text = text.strip()

    # Try direct JSON parse
    parsed = _try_json(text)

    # Fallback: extract first JSON object from response
    if parsed is None:
        match = re.search(r"\{.*\}", text, re.DOTALL)
        if match:
            parsed = _try_json(match.group())

    if parsed is None or not isinstance(parsed, dict):
        logger.warning("Could not parse intent from LLM response: %s", text[:200])
        return IntentResult()

    tool_name = parsed.get("tool")
    confidence = parsed.get("confidence", "high")
    if confidence not in ("high", "medium", "low"):
        confidence = "high"

    if tool_name is None:
        return IntentResult(confidence=confidence)

    # Validate tool name against available tools
    valid_names = {
        t.get("function", {}).get("name") for t in tools
    }
    if tool_name not in valid_names:
        logger.warning("Intent returned unknown tool '%s'", tool_name)
        return IntentResult()

    arguments = parsed.get("arguments", {})
    if not isinstance(arguments, dict):
        arguments = {}

    ack = parsed.get("ack") or None
    if ack is not None:
        ack = ack.strip() or None

    logger.info(
        "Intent classified: tool=%s, confidence=%s, args=%s",
        tool_name, confidence, json.dumps(arguments)[:200],
    )
    return IntentResult(tool_name=tool_name, arguments=arguments, confidence=confidence, ack=ack)


def _try_json(text: str) -> dict | list | None:
    """Try to parse JSON, return None on failure."""
    try:
        return json.loads(text)
    except (json.JSONDecodeError, TypeError):
        return None