FabledScribe/src/fabledassistant/services/generation_task.py

"""Background asyncio task for LLM generation.

Streams from Ollama into a GenerationBuffer, periodically flushing to DB.
Runs independently of any HTTP connection.
"""

import asyncio
import json
import logging
import re
import time
from collections.abc import AsyncGenerator

import httpx

from sqlalchemy import update

from fabledassistant.config import Config
from fabledassistant.models import async_session
from fabledassistant.models.conversation import Message
from fabledassistant.services.generation_buffer import (
    GenerationBuffer,
    GenerationState,
)
from fabledassistant.services.llm import ChatChunk, build_context, generate_completion, pick_num_ctx, stream_chat, stream_chat_with_tools, summarize_history_for_context
from fabledassistant.services.chat import update_conversation_title
from fabledassistant.services.settings import get_setting
from fabledassistant.services.logging import log_generation
from fabledassistant.services.tools import get_tools_for_user, execute_tool
from fabledassistant.services.research import run_research_pipeline

logger = logging.getLogger(__name__)

# Mistral prefixes tool-call responses with "[TOOL_CALLS]" as visible text
_TOOL_CALL_MARKER = re.compile(r"^\s*\[TOOL_CALLS\]\s*", re.IGNORECASE)

DB_FLUSH_INTERVAL = 5.0  # seconds between partial DB flushes


# Human-readable labels for each tool, shown in the status indicator
_TOOL_LABELS: dict[str, str] = {
    "create_note": "Creating note/task",
    "update_note": "Updating note/task",
    "delete_note": "Deleting note/task",
    "read_note": "Reading note",
    "list_notes": "Listing notes",
    "list_tasks": "Searching tasks",
    "search_notes": "Searching notes (semantic)",
    "create_event": "Creating calendar event",
    "list_events": "Searching calendar",
    "search_events": "Searching calendar",
    "update_event": "Updating calendar event",
    "delete_event": "Removing calendar event",
    "list_calendars": "Listing calendars",
    "lookup": "Looking up information",
    "research_topic": "Researching topic",
}


async def _generate_title(messages: list[dict], user_id: int) -> str:
    """Ask the LLM for a concise conversation title.

    Only uses user messages to avoid feeding tool-call JSON, system prompt
    fragments, or other noise into the title generator. Caps input length
    to keep the task fast and focused.
    """
    user_texts = []
    for m in messages:
        if m["role"] == "user":
            content = (m.get("content") or "").strip()
            if content:
                user_texts.append(content[:300])
    if not user_texts:
        return ""
    # First + last user messages capture intent best
    if len(user_texts) > 2:
        user_texts = [user_texts[0], user_texts[-1]]

    prompt_messages = [
        {
            "role": "user",
            "content": (
                "Generate a concise 3-8 word title for a conversation that started with:\n\n"
                + "\n\n".join(user_texts)
                + "\n\nReply with ONLY the title. No quotes, no punctuation, no explanation."
            ),
        },
    ]
    bg_model = await get_setting(user_id, "background_model", Config.OLLAMA_BACKGROUND_MODEL)
    title = await generate_completion(prompt_messages, bg_model, max_tokens=30, num_ctx=1024)
    # Strip common LLM noise: quotes, thinking tags, role labels
    title = title.strip().strip('"\'').strip()
    for prefix in ("Title:", "title:", "Assistant:", "User:"):
        if title.startswith(prefix):
            title = title[len(prefix):].strip()
    # Drop anything after a newline (model sometimes adds explanation)
    if "\n" in title:
        title = title.split("\n")[0].strip()
    return title[:80] if title else ""


async def _update_message(
    message_id: int,
    content: str,
    status: str,
    tool_calls: list[dict] | None = None,
) -> None:
    values: dict = {"content": content, "status": status}
    if tool_calls is not None:
        values["tool_calls"] = tool_calls
    async with async_session() as session:
        await session.execute(
            update(Message)
            .where(Message.id == message_id)
            .values(**values)
        )
        await session.commit()


async def _stream_with_retry(
    messages: list[dict],
    model: str,
    tools: list[dict],
    think: bool,
    num_ctx: int | None = None,
) -> AsyncGenerator[ChatChunk, None]:
    """stream_chat_with_tools with automatic retry on Ollama 500 errors.

    500s occur when Ollama is still loading a model or handling a concurrent
    request (e.g. tag suggestions racing with round 1). Retries up to 2 times
    with a short delay — by which point the model is warm and other calls done.
    """
    last_exc: BaseException | None = None
    for attempt in range(3):
        if attempt > 0:
            delay = 3.0 * attempt
            logger.warning(
                "Ollama stream 500 (attempt %d/3), retrying in %.0fs", attempt, delay
            )
            await asyncio.sleep(delay)
        try:
            async for chunk in stream_chat_with_tools(messages, model, tools=tools, think=think, num_ctx=num_ctx):
                yield chunk
            return
        except httpx.HTTPStatusError as exc:
            last_exc = exc
            if exc.response.status_code != 500:
                break  # non-500 is not retryable
        except BaseException as exc:
            last_exc = exc
            break
    if last_exc is not None:
        raise last_exc


async def run_generation(
    buf: GenerationBuffer,
    history: list[dict],
    model: str,
    user_id: int,
    conv_id: int,
    conv_title: str,
    user_content: str,
    context_note_id: int | None = None,
    include_note_ids: list[int] | None = None,
    excluded_note_ids: list[int] | None = None,
    think: bool = False,
    rag_project_id: int | None = None,
    workspace_project_id: int | None = None,
    user_timezone: str | None = None,
    voice_mode: bool = False,
) -> None:
    """Stream LLM response into buffer with periodic DB flushes."""
    MAX_TOOL_ROUNDS = 6
    msg_id = buf.assistant_message_id

    buf.append_event("status", {"status": "Building context..."})

    # Phase 1: Resolve the tools list for this user, scoped to conversation type.
    #
    # Journal conversations get NO tools (2026-05-22 architecture pivot):
    # the chat model talks, a separate background curator does tool calls
    # asynchronously. See services/curator.py. Removing tools here is the
    # mechanical change that makes the architecture real — the chat model
    # can no longer fire record_moment / create_task / etc. and therefore
    # can no longer lie about firing them.
    from fabledassistant.models import async_session as _async_session
    from fabledassistant.models.conversation import Conversation as _Conversation
    async with _async_session() as _sess:
        _conv = await _sess.get(_Conversation, conv_id)
    _conversation_type = (
        _conv.conversation_type if _conv and _conv.conversation_type else "chat"
    )
    if _conversation_type == "journal":
        tools = []
        logger.info(
            "Conv %d is journal: passing tools=[] to chat model "
            "(curator handles tool calls async)",
            conv_id,
        )
    else:
        tools = await get_tools_for_user(user_id, conversation_type=_conversation_type)

    logger.info(
        "Starting generation for conv %d: model=%s, tools=%d",
        conv_id, model, len(tools),
    )

    # Phase 2: Summarize long conversation history if needed.
    history_to_use = history
    history_summary: str | None = None
    if len(history) > 30:  # matches _HISTORY_SUMMARY_THRESHOLD in llm.py
        buf.append_event("status", {"status": "Summarizing conversation history..."})
        history_to_use, history_summary = await summarize_history_for_context(history, model)

    # Phase 3: Build context.
    # Note: Ollama lazy-loads models on the first /api/chat request, so polling
    # /api/ps for model readiness only causes delay. We proceed immediately and
    # let Ollama handle loading on demand.

    # Fetch voice_speech_style from user settings when voice_mode is active.
    voice_speech_style = "conversational"
    if voice_mode:
        from fabledassistant.services.settings import get_setting
        voice_speech_style = await get_setting(user_id, "voice_speech_style", "conversational")

    messages, context_meta = await build_context(
        user_id, history_to_use, context_note_id, user_content,
        history_summary=history_summary,
        include_note_ids=include_note_ids,
        excluded_note_ids=excluded_note_ids,
        rag_project_id=rag_project_id,
        workspace_project_id=workspace_project_id,
        user_timezone=user_timezone,
        conv_id=conv_id,
        voice_mode=voice_mode,
        voice_speech_style=voice_speech_style,
    )

    # Pick the smallest context tier that fits the current messages AND the
    # tool schemas (which can be 6-10K tokens on their own with ~40 tools).
    # Using the minimum needed tier reduces KV cache size and speeds up prefill.
    num_ctx = pick_num_ctx(messages, tools=tools)
    logger.debug("Adaptive num_ctx=%d for conv %d", num_ctx, conv_id)

    # Emit context event
    buf.append_event("context", {"context": context_meta})

    # Think mode is hardcoded off (2026-05-23). Historical context:
    # originally forced on for qwen3's combined think+tools template;
    # then made user-configurable when we decoupled the architecture;
    # now removed entirely because in the chat+curator world there's no
    # reason for the chat model to think. Chat has tools=[] — it just
    # talks, and think on a no-tools conversational pass is pure
    # latency cost (the May 2026 bench measured 1-2 min/turn for
    # unclear quality benefit). The curator (services/curator.py) also
    # already hardcodes think=False for its own reasons.
    think_requested = think
    think = False

    t_start = time.monotonic()
    timing: dict = {
        "think_requested": think_requested,
        "think": think,
        "num_ctx": num_ctx,
        "tools": [],
        "rounds": 0,
        "prompt_tokens": None,
        "output_tokens": None,
        "first_token_ms": None,
        "thinking_ms": None,
        "ttft_ms": None,
        "generation_ms": None,
        "total_ms": None,
    }

    last_flush = time.monotonic()
    all_tool_calls: list[dict] = []
    new_rag_scope: object = False  # sentinel; set to int|None when scope changes
    new_rag_scope_label: str | None = None

    try:
        cancelled = False
        research_completed = False

        for _round in range(MAX_TOOL_ROUNDS):
            timing["rounds"] = _round + 1
            round_tool_calls: list[dict] = []
            logger.info("Generation round %d started for conv %d (model=%s)", _round, conv_id, model)

            if cancelled:
                break

            buf.append_event("status", {"status": "Generating response..." if _round == 0 else "Composing response..."})
            t_stream = time.monotonic()

            approx_msg_chars = sum(len(str(m.get("content", ""))) for m in messages)
            round_content_start = len(buf.content_so_far)
            round_output_tokens_start = timing.get("output_tokens") or 0
            round_prompt_tokens_start = timing.get("prompt_tokens") or 0
            logger.info(
                "CTX_DIAG round_start conv=%d round=%d num_ctx=%d msgs=%d approx_chars=%d think=%s",
                conv_id, _round, num_ctx, len(messages), approx_msg_chars, think,
            )
            async for chunk in _stream_with_retry(messages, model, tools, think, num_ctx=num_ctx):
                if buf.cancel_event.is_set():
                    cancelled = True
                    break

                if chunk.type == "thinking":
                    if timing["first_token_ms"] is None:
                        timing["first_token_ms"] = int((time.monotonic() - t_start) * 1000)
                    buf.append_event("thinking_chunk", {"chunk": chunk.content})

                elif chunk.type == "content":
                    if timing["ttft_ms"] is None:
                        now_ms = int((time.monotonic() - t_start) * 1000)
                        timing["ttft_ms"] = now_ms
                        if timing["first_token_ms"] is None:
                            # No thinking phase occurred — first token IS content.
                            timing["first_token_ms"] = now_ms
                        else:
                            # Thinking phase duration = gap between first thinking
                            # token and first content token.
                            timing["thinking_ms"] = now_ms - timing["first_token_ms"]
                    buf.content_so_far += chunk.content
                    clean = _TOOL_CALL_MARKER.sub("", chunk.content)
                    if clean:
                        buf.append_event("chunk", {"chunk": clean})

                    now = time.monotonic()
                    if now - last_flush >= DB_FLUSH_INTERVAL:
                        try:
                            await _update_message(msg_id, buf.content_so_far, "generating")
                        except Exception:
                            logger.warning("Failed periodic flush for message %d", msg_id, exc_info=True)
                        last_flush = now

                elif chunk.type == "done":
                    if chunk.prompt_tokens is not None:
                        timing["prompt_tokens"] = (timing["prompt_tokens"] or 0) + chunk.prompt_tokens
                    if chunk.output_tokens is not None:
                        timing["output_tokens"] = (timing["output_tokens"] or 0) + chunk.output_tokens

                elif chunk.type == "tool_calls" and chunk.tool_calls:
                    logger.info("Round %d: model returned %d tool call(s)", _round, len(chunk.tool_calls))
                    for tc in chunk.tool_calls:
                        fn = tc.get("function", {})
                        tool_name = fn.get("name", "")
                        arguments = fn.get("arguments", {})
                        logger.info("Executing tool: %s(%s)", tool_name, json.dumps(arguments)[:200])
                        buf.append_event("status", {"status": f"{_TOOL_LABELS.get(tool_name, 'Working')}..."})

                        t_tool = time.monotonic()
                        if tool_name == "research_topic":
                            topic = arguments.get("topic", "")
                            try:
                                note = await run_research_pipeline(topic, user_id, model, buf, project_id=workspace_project_id)
                                result = {
                                    "success": True,
                                    "type": "research_note",
                                    "data": {"id": note.id, "title": note.title},
                                }
                                done_text = (
                                    f"\n\n---\n\nResearch complete! I've compiled a note: "
                                    f"**[{note.title}](/notes/{note.id})**."
                                )
                                buf.append_event("chunk", {"chunk": done_text})
                                buf.content_so_far += done_text
                            except Exception as e:
                                logger.exception("Research pipeline failed for topic: %s", topic)
                                err_msg = str(e) or f"{type(e).__name__}: unexpected error"
                                result = {"success": False, "error": err_msg}
                                err_text = f"\nResearch failed: {err_msg}"
                                buf.append_event("chunk", {"chunk": err_text})
                                buf.content_so_far += err_text
                            research_completed = True
                        else:
                            result = await execute_tool(
                                user_id, tool_name, arguments,
                                conv_id=conv_id,
                                workspace_project_id=workspace_project_id,
                            )

                        # Capture RAG scope change for SSE done event
                        if result.get("type") == "rag_scope_set" and result.get("success"):
                            new_rag_scope = arguments.get("project_id")
                            new_rag_scope_label = result.get("scope_label")

                        timing["tools"].append({"name": tool_name, "ms": int((time.monotonic() - t_tool) * 1000)})
                        logger.info("Tool %s result: success=%s", tool_name, result.get("success"))

                        tool_record = {
                            "function": tool_name,
                            "arguments": arguments,
                            "result": result,
                            "status": "success" if result.get("success") else "error",
                        }
                        round_tool_calls.append(tool_record)
                        all_tool_calls.append(tool_record)
                        buf.append_event("tool_call", {"tool_call": tool_record})

            round_content_added = len(buf.content_so_far) - round_content_start
            round_output_tokens_added = (timing.get("output_tokens") or 0) - round_output_tokens_start
            round_prompt_tokens = (timing.get("prompt_tokens") or 0) - round_prompt_tokens_start
            headroom = num_ctx - round_prompt_tokens if round_prompt_tokens else None
            is_silent = (
                not round_tool_calls
                and round_content_added == 0
                and round_output_tokens_added > 0
            )
            logger.info(
                "CTX_DIAG round_end conv=%d round=%d think=%s prompt_tokens=%d output_tokens=%d headroom=%s content_added=%d tool_calls=%d silent=%s",
                conv_id, _round, think, round_prompt_tokens, round_output_tokens_added,
                headroom, round_content_added, len(round_tool_calls), is_silent,
            )

            timing["generation_ms"] = int((time.monotonic() - t_stream) * 1000)

            if cancelled:
                logger.info("Generation cancelled for conv %d", conv_id)
                break

            if research_completed:
                logger.info("Research complete for conv %d, ending generation", conv_id)
                break

            if not round_tool_calls:
                logger.info("Round %d: no tool calls, final content length=%d", _round, len(buf.content_so_far))
                break

            logger.info("Round %d: %d tool call(s) executed, starting next round", _round, len(round_tool_calls))

            buf.content_so_far = _TOOL_CALL_MARKER.sub("", buf.content_so_far)

            messages.append({
                "role": "assistant",
                "content": buf.content_so_far,
                "tool_calls": [
                    {"function": {"name": tc["function"], "arguments": tc["arguments"]}}
                    for tc in round_tool_calls
                ],
            })
            for tc in round_tool_calls:
                messages.append({"role": "tool", "content": json.dumps(tc["result"])})

            buf.content_so_far = ""

        # Strip model artifacts from final content
        buf.content_so_far = _TOOL_CALL_MARKER.sub("", buf.content_so_far)

        # Silent-generation safety net: the model burned output tokens but
        # nothing landed in content or tool_calls (seen with qwen3:14b when
        # its tool-call emission doesn't parse). Show a visible fallback so
        # the user isn't staring at an empty bubble.
        if (
            not cancelled
            and not buf.content_so_far.strip()
            and not all_tool_calls
            and (timing.get("output_tokens") or 0) > 0
        ):
            logger.warning(
                "Silent generation for conv %d: output_tokens=%s but empty content "
                "and no tool calls (model=%s)",
                conv_id, timing.get("output_tokens"), model,
            )
            fallback = (
                "I wasn't able to produce a usable response — the model generated "
                "tokens that couldn't be parsed as content or a tool call. "
                "Please try rephrasing, or try again."
            )
            buf.content_so_far = fallback
            buf.append_event("chunk", {"chunk": fallback})

        # Final save
        logger.info("Generation complete for conv %d: content_length=%d, tool_calls=%d",
                     conv_id, len(buf.content_so_far), len(all_tool_calls))
        await _update_message(
            msg_id,
            buf.content_so_far,
            "complete",
            tool_calls=all_tool_calls if all_tool_calls else None,
        )

        timing["total_ms"] = int((time.monotonic() - t_start) * 1000)
        logger.info(
            "Generation timing for conv %d: total=%dms think=%s(req=%s) first_token=%s "
            "thinking=%s ttft=%s generation=%s tools=%s",
            conv_id, timing["total_ms"],
            timing["think"], timing["think_requested"],
            timing["first_token_ms"], timing["thinking_ms"], timing["ttft_ms"],
            timing["generation_ms"],
            [(t["name"], t["ms"]) for t in timing["tools"]],
        )
        try:
            await log_generation(user_id, conv_id, model, timing)
        except Exception:
            logger.warning("Failed to persist generation timing for conv %d", conv_id, exc_info=True)

        # Per-turn tool-call telemetry. Empirical surface for evaluating
        # model swaps without needing user reports — answers "did model X
        # actually fire record_moment when it should have?" The helper is
        # internally try/except so this never affects the user-facing flow.
        from fabledassistant.services.generation_log import log_tool_outcomes
        await log_tool_outcomes(
            user_id=user_id,
            conv_id=conv_id,
            assistant_message_id=msg_id,
            model=model,
            think_enabled=think,
            tools_available=[
                (t.get("function") or {}).get("name") for t in tools
            ],
            tool_calls=all_tool_calls,
        )

        buf.state = GenerationState.COMPLETED
        buf.finished_at = time.monotonic()
        done_payload: dict = {"done": True, "message_id": msg_id, "timing": timing}
        if new_rag_scope is not False:
            done_payload["new_rag_scope"] = new_rag_scope
            done_payload["new_rag_scope_label"] = new_rag_scope_label
        buf.append_event("done", done_payload)

        # Fire push notification when complete (non-critical, fire-and-forget)
        try:
            from fabledassistant.services.push import send_push_notification, vapid_enabled
            if vapid_enabled():
                text = buf.content_so_far.strip()
                if text:
                    preview = text[:120].rstrip()
                    if len(text) > 120:
                        preview += "…"
                else:
                    # Tool-only response — summarise what was done
                    tool_names = [tc.get("function") for tc in all_tool_calls if tc.get("function")]
                    if tool_names:
                        preview = f"Completed: {', '.join(tool_names[:3])}"
                    else:
                        preview = "Action completed"
                asyncio.create_task(send_push_notification(
                    user_id,
                    title="Response ready",
                    body=preview,
                    url=f"/chat/{conv_id}",
                ))
        except Exception:
            logger.warning("Failed to schedule push notification", exc_info=True)

        # Title generation is non-critical — fire-and-forget so done fires immediately
        non_system = [m for m in messages if m["role"] != "system"]
        msg_count = len(non_system)
        should_gen_title = not conv_title or (msg_count > 0 and msg_count % 10 == 0)

        if should_gen_title:
            # Feed the title model the *raw* conversation turns only — never
            # the post-build_context ``messages`` list. ``build_context``
            # prepends RAG snippets and URL content INTO the user message
            # string itself, so filtering by role="user" downstream still
            # surfaces that noise as the "user's message". That pollution
            # caused wildly-wrong titles (bug #109) — the small background
            # model was staring at article excerpts instead of what the user
            # actually typed. Pass the original history + the raw user_content
            # + the assistant reply.
            title_messages: list[dict] = [
                {"role": m["role"], "content": m.get("content") or ""}
                for m in history
                if m.get("role") in ("user", "assistant")
            ]
            title_messages.append({"role": "user", "content": user_content})
            title_messages.append({"role": "assistant", "content": buf.content_so_far})

            async def _bg_title() -> None:
                try:
                    title = await _generate_title(title_messages, user_id)
                    if title:
                        await update_conversation_title(user_id, conv_id, title)
                except Exception:
                    logger.warning("Failed to generate title for conversation %d", conv_id, exc_info=True)
                    if not conv_title:
                        fallback = user_content[:80]
                        if len(user_content) > 80:
                            fallback += "..."
                        await update_conversation_title(user_id, conv_id, fallback)

            asyncio.create_task(_bg_title())

    except Exception as e:
        logger.exception("Error in generation task for conversation %d", conv_id)
        # Save partial content with error status
        try:
            await _update_message(msg_id, buf.content_so_far, "error")
        except Exception:
            logger.warning("Failed to save error state for message %d", msg_id, exc_info=True)

        buf.state = GenerationState.ERRORED
        buf.finished_at = time.monotonic()
        buf.append_event("error", {"error": str(e)})


async def run_assist_generation(
    buf: GenerationBuffer,
    messages: list[dict],
    model: str,
) -> None:
    """Stream LLM response for assist into buffer. No DB persistence.

    Retries up to 3 times on Ollama 500 errors (model still loading).
    On each retry the accumulated content is reset so the done event
    always reflects only the successful generation.
    """
    from fabledassistant.services.llm import pick_num_ctx
    input_chars = sum(len(m.get("content", "")) for m in messages)
    num_ctx = pick_num_ctx(messages)
    logger.info("Assist generation started: model=%s, input_chars=%d, num_ctx=%d", model, input_chars, num_ctx)

    last_exc: BaseException | None = None
    for attempt in range(3):
        if attempt > 0:
            delay = 3.0 * attempt
            logger.warning(
                "Ollama assist stream 500 (attempt %d/3), retrying in %.0fs", attempt, delay
            )
            await asyncio.sleep(delay)
        try:
            buf.content_so_far = ""
            async for chunk in stream_chat(messages, model, options={"num_predict": num_ctx}, num_ctx=num_ctx):
                buf.content_so_far += chunk
                buf.append_event("chunk", {"chunk": chunk})

            output_chars = len(buf.content_so_far)
            logger.info(
                "Assist generation complete: output_chars=%d, events=%d",
                output_chars, len(buf.events),
            )
            buf.state = GenerationState.COMPLETED
            buf.finished_at = time.monotonic()
            buf.append_event("done", {"done": True, "full_text": buf.content_so_far})
            logger.info("Assist done event appended (event index %d)", len(buf.events) - 1)
            return

        except httpx.HTTPStatusError as exc:
            last_exc = exc
            if exc.response.status_code != 500:
                break
        except Exception as exc:
            last_exc = exc
            break

    logger.exception("Error in assist generation task")
    buf.state = GenerationState.ERRORED
    buf.finished_at = time.monotonic()
    buf.append_event("error", {"error": str(last_exc)})