backend-python/src/services/generation.py

"""
Simple generation services (title, daily tip, related questions, agent selection).
"""

from __future__ import annotations

from datetime import datetime
import re
from typing import Any
from zoneinfo import ZoneInfo

from ..models.generation import (
    DailyTipResponse,
    TitleSpaceResponse,
)
from ..models.stream_chat import StreamChatRequest
from .llm_utils import run_agent_completion, safe_json_parse


def _get_output_value(output_obj: Any, *keys: str) -> Any:
    if not output_obj:
        return None
    if isinstance(output_obj, dict):
        for key in keys:
            if key in output_obj:
                return output_obj.get(key)
        return None
    for key in keys:
        if hasattr(output_obj, key):
            return getattr(output_obj, key)
    return None


def _build_time_context(user_timezone: str | None, user_locale: str | None) -> str:
    timezone = user_timezone or "UTC"
    locale = user_locale or "en-US"
    try:
        tzinfo = ZoneInfo(timezone)
        now = datetime.now(tzinfo)
    except Exception:
        timezone = "UTC"
        now = datetime.utcnow()
    formatted = now.strftime("%Y-%m-%d %H:%M:%S")
    return (
        "\n\nLocal time context:\n"
        f"##today local time: {formatted} ({timezone})\n"
        f"locale: {locale}\n"
        f"iso: {now.isoformat()}\n"
    )


def _append_time_context(
    messages: list[dict[str, Any]],
    user_timezone: str | None,
    user_locale: str | None,
    reference_text: str | None = None,
) -> list[dict[str, Any]]:
    time_context = _build_time_context(user_timezone, user_locale)
    if not messages:
        return [{"role": "system", "content": time_context.strip()}]
    updated = list(messages)
    system_index = next((i for i, m in enumerate(updated) if m.get("role") == "system"), -1)
    if system_index != -1:
        updated[system_index] = {
            **updated[system_index],
            "content": f"{updated[system_index].get('content', '')}{time_context}",
        }
    else:
        updated.insert(0, {"role": "system", "content": time_context.strip()})
    return updated


def _normalize_title_text_output(raw: str) -> str:
    text = str(raw or "").strip()
    if not text:
        return ""
    # Common wrappers from models
    text = text.replace("\r", "\n").strip()
    first_line = next((line.strip() for line in text.split("\n") if line.strip()), "")
    text = first_line or text
    for prefix in ("title:", "标题：", "标题:", "- ", "* ", "# "):
        if text.lower().startswith(prefix.lower()):
            text = text[len(prefix):].strip()
    for prefix in (
        "suggested title:",
        "suggested title：",
        "generated title:",
        "generated title：",
        "conversation title:",
        "conversation title：",
    ):
        if text.lower().startswith(prefix.lower()):
            text = text[len(prefix):].strip()
    text = text.replace("**", "").replace("__", "").strip()
    text = text.strip().strip("\"'“”‘’")
    text = re.sub(r"^(?:user|assistant|system)\s*:\s*", "", text, flags=re.IGNORECASE)
    text = re.sub(
        r"^(?:the user(?:'s)? (?:is asking|asks|wants|needs) (?:about|for)\s+|"
        r"this conversation is about\s+|the conversation is about\s+|"
        r"topic\s*:\s*|summary\s*:\s*)",
        "",
        text,
        flags=re.IGNORECASE,
    )
    text = re.split(r"[.?!。！？]\s*", text, maxsplit=1)[0].strip()
    text = re.sub(r"\s+", " ", text).strip(" ,;:[](){}")
    text = text.strip().strip("\"'“”‘’")
    return text[:120].strip()


def _truncate_title_phrase(text: str, max_words: int = 5, max_chars: int = 36) -> str:
    normalized = re.sub(r"\s+", " ", str(text or "")).strip()
    if not normalized:
        return ""
    if " " not in normalized:
        return normalized[:max_chars].strip(" ,;:-")
    words = normalized.split(" ")
    return " ".join(words[:max_words]).strip(" ,;:-")


def _finalize_title(raw: Any, fallback: str = "New Conversation") -> str:
    title = _normalize_title_text_output(str(raw or ""))
    if not title:
        return fallback

    disallowed_starts = (
        "here is",
        "here's",
        "i can",
        "i would",
        "sure",
        "certainly",
        "let me",
        "user:",
        "assistant:",
        "system:",
    )
    lowered = title.lower()
    if lowered.startswith(disallowed_starts):
        return fallback

    title = _truncate_title_phrase(title)
    return title or fallback


def _title_prompt(task_suffix: str = "") -> str:
    suffix = f"\n{task_suffix.strip()}" if task_suffix and task_suffix.strip() else ""
    return (
        "Generate a short conversation title from the provided text.\n"
        "The text may be a user message or a short transcript.\n\n"
        "Rules:\n"
        "- Return only the title.\n"
        "- Maximum 5 words.\n"
        "- Do not answer the user.\n"
        "- Do not explain.\n"
        "- Do not use quotes or markdown."
        f"{suffix}\n\n"
        "Return only the title text."
    )


def _extract_single_emoji_text(raw: str) -> str:
    import re

    text = str(raw or "").strip()
    if not text:
        return ""

    # JSON / key-value fallback parsing still allowed, but not required.
    parsed = safe_json_parse(text)
    if isinstance(parsed, dict):
        emojis = parsed.get("emojis")
        if isinstance(emojis, list) and emojis:
            return str(emojis[0]).strip()
        if isinstance(emojis, str) and emojis.strip():
            return emojis.strip()

    m = re.search(r"emojis=\[[\"']?(.*?)[\"']?\]", text)
    if m:
        return m.group(1).strip()

    # Extract first likely emoji grapheme-ish token
    # Covers most symbols/pictographs and optional variation selector / ZWJ tails.
    m = re.search(
        r"([\u2600-\u27BF\U0001F300-\U0001FAFF](?:\uFE0F)?(?:\u200D[\u2600-\u27BF\U0001F300-\U0001FAFF](?:\uFE0F)?)*)",
        text,
    )
    if m:
        return m.group(1).strip()

    # Last resort: first non-empty token
    token = next((part for part in re.split(r"\s+", text) if part), "")
    return token[:8].strip()


async def generate_daily_tip(
    *,
    provider: str,
    language: str | None,
    category: str | None,
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> str:
    language_block = f"\n\n## Language\nReply in {language}." if language else ""
    category_block = f"\n\n## Category\n{category}" if category else ""
    messages = [
        {
            "role": "system",
            "content": (
                "## Task\n"
                "Generate a short, practical tip for today. Keep it to 1-2 sentences and avoid emojis."
                f"{category_block}{language_block}\n\n"
                "## Output\n"
                "Return only the tip text."
            ),
        },
        {"role": "user", "content": "Daily tip."},
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, "daily tip")

    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=response_format,
        output_schema=DailyTipResponse,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()
    output_obj = result.get("output")

    tip = None
    if output_obj:
        tip = _get_output_value(output_obj, "tip")

    if not tip:
        parsed = safe_json_parse(content)
        if isinstance(parsed, dict):
            tip = parsed.get("tip")
    if not tip and thought:
        parsed = safe_json_parse(thought)
        if isinstance(parsed, dict):
            tip = parsed.get("tip")
    if not tip:
        tip = content # Fallback for plain text

    # Cleanup if model included "tip='...'"
    if tip and isinstance(tip, str) and tip.startswith("tip="):
        import re
        match = re.search(r"tip=['\"](.*?)['\"]", tip)
        if match:
            tip = match.group(1)

    return (tip or content or "").strip()


async def generate_title(
    *,
    provider: str,
    first_message: str,
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> dict[str, Any]:
    messages = [
        {
            "role": "system",
            "content": _title_prompt(),
        },
        {"role": "user", "content": first_message},
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, first_message)
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=None,
        output_schema=None,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()

    title = ""
    parsed = safe_json_parse(content) or {}
    if isinstance(parsed, dict):
        title = str(parsed.get("title") or "").strip()
    if not title and thought:
        parsed_thought = safe_json_parse(thought) or {}
        if isinstance(parsed_thought, dict):
            title = str(parsed_thought.get("title") or "").strip()

    # Robust cleanup for models like GLM
    if (not title or str(title).strip().startswith("title=")) and content:
        import re
        m_title = re.search(r"title=['\"](.*?)['\"]", content, flags=re.DOTALL)
        if m_title:
            title = m_title.group(1)

    # Final normalization
    title = _finalize_title(title or content)
    return {"title": title}


async def generate_emoji(
    *,
    provider: str,
    first_message: str,
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> dict[str, Any]:
    messages = [
        {
            "role": "system",
            "content": (
                "## Task\n"
                "Select exactly 1 emoji that best matches the user's message topic.\n"
                "Do NOT answer the message.\n\n"
                "## Output\n"
                "Return only 1 emoji character. No JSON. No words."
            ),
        },
        {"role": "user", "content": first_message},
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, first_message)
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=None,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()

    emoji = _extract_single_emoji_text(content)
    if not emoji:
        emoji = _extract_single_emoji_text(thought)
    return {"emojis": [emoji] if emoji else []}


async def generate_title_and_space(
    *,
    provider: str,
    first_message: str,
    spaces: list[dict[str, Any]],
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> dict[str, Any]:
    space_labels = ", ".join([str(s.get("label", "")).strip() for s in spaces or []]).strip()
    messages = [
        {
            "role": "system",
            "content": (
                f"{_title_prompt()}\n\n"
                "## Additional Tasks\n"
                f'1. Select the most appropriate space from the following list: [{space_labels}]. '
                "If none fit well, return null.\n"
                "2. Select 1 emoji that best matches the conversation.\n\n"
                "## Output\n"
                'Return the result as a JSON object with keys "title", "spaceLabel", and "emojis".'
            ),
        },
        {"role": "user", "content": first_message},
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, first_message)
    response_format = {"type": "json_object"} if provider != "gemini" else None
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=response_format,
        output_schema=TitleSpaceResponse,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()

    # Try structured output first
    output_obj = result.get("output")

    title = None
    space_label = None
    emojis = []

    # If it's a dict or model, extract directly
    if output_obj:
        title = _get_output_value(output_obj, "title")
        space_label = _get_output_value(output_obj, "space_label", "spaceLabel")
        emojis = _get_output_value(output_obj, "emojis") or []

    # If above failed, fallback to manual parsing of content
    if not title:
        parsed = safe_json_parse(content) or {}
        if isinstance(parsed, dict):
            title = parsed.get("title")
            space_label = space_label or parsed.get("spaceLabel") or parsed.get("space_label")
            emojis = emojis or parsed.get("emojis")
    if not title and thought:
        parsed = safe_json_parse(thought) or {}
        if isinstance(parsed, dict):
            title = parsed.get("title")
            space_label = space_label or parsed.get("spaceLabel") or parsed.get("space_label")
            emojis = emojis or parsed.get("emojis")

    # Robust cleanup for models like GLM (even if parsing failed partially)
    if (not title or (isinstance(title, str) and title.strip().startswith("title="))) and content:
        import re
        m_title = re.search(r"title=['\"](.*?)['\"]", content, flags=re.DOTALL)
        if m_title:
            title = m_title.group(1)

        if not emojis:
            m_emojis = re.search(r"emojis=\[[\"']?(.*?)[\"']?\]", content)
            if m_emojis:
                emojis = [m_emojis.group(1)]

        if not space_label:
            m_space = re.search(r"space_?label=['\"](.*?)['\"]", content, re.IGNORECASE) or re.search(r"space=['\"](.*?)['\"]", content, re.IGNORECASE)
            if m_space:
                space_label = m_space.group(1)

    # Normalize values
    title = _finalize_title(title or content)

    # IMPROVED matching: Case-insensitive and trimmed
    search_label = str(space_label or "").strip().lower()
    selected_space = next(
        (s for s in spaces if s.get("label", "").strip().lower() == search_label),
        None
    )

    if not isinstance(emojis, list):
        emojis = []
    emojis = [str(item).strip().strip("'").strip('"') for item in emojis if str(item).strip()][:1]

    return {"title": title, "space": selected_space, "emojis": emojis}


def _sanitize_option_text(text: Any) -> str:
    return (
        str(text or "")
        .replace("\r", " ")
        .replace("\n", " ")
        .replace("{", "")
        .replace("}", "")
        .split()
    )


async def generate_title_space_and_agent(
    *,
    provider: str,
    first_message: str,
    spaces_with_agents: list[dict[str, Any]],
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> dict[str, Any]:
    space_lines = []
    for space in spaces_with_agents or []:
        agent_entries = []
        for agent in space.get("agents") or []:
            if isinstance(agent, str):
                agent_entries.append({"name": agent})
            else:
                agent_entries.append(
                    {"name": agent.get("name", ""), "description": agent.get("description", "")}
                )
        agent_tokens = []
        for agent in agent_entries:
            name = " ".join(_sanitize_option_text(agent.get("name")))
            description = " ".join(_sanitize_option_text(agent.get("description")))
            if name and description:
                agent_tokens.append(f"{name} - {description}")
            elif name:
                agent_tokens.append(name)
        space_label = " ".join(_sanitize_option_text(space.get("label")))
        space_description = " ".join(_sanitize_option_text(space.get("description")))
        space_token = f"{space_label} - {space_description}" if space_description else space_label
        space_lines.append(f"{space_token}:{{{','.join(agent_tokens)}}}")

    messages = [
        {
            "role": "system",
            "content": (
                f"{_title_prompt()}\n\n"
                "## Additional Tasks\n"
                "1. Select the most appropriate space from the list below and return its spaceLabel.\n"
                "2. If the chosen space has agents, select the best matching agent by agentName. Otherwise return null.\n"
                "3. Select 1 emoji that best matches the conversation.\n\n"
                "## Output\n"
                'Return the result as JSON with keys "title", "spaceLabel", "agentName", and "emojis". '
                '"emojis" must be an array with 1 emoji character.'
            ),
        },
        {
            "role": "user",
            "content": f"{first_message}\n\nSpaces and agents:\n" + "\n".join(space_lines),
        },
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, first_message)
    response_format = {"type": "json_object"} if provider != "gemini" else None
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=response_format,
        # Avoid provider-side grammar cache collisions for lightweight JSON routes.
        # We parse JSON content locally instead of forcing native structured output.
        output_schema=None,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()

    # Try structured output first
    output_obj = result.get("output")
    title = None
    space_label = None
    agent_name = None
    emojis = []

    # If it's a dict or model, extract directly
    if output_obj:
        title = _get_output_value(output_obj, "title")
        space_label = _get_output_value(output_obj, "space_label", "spaceLabel")
        agent_name = _get_output_value(output_obj, "agent_name", "agentName")
        emojis = _get_output_value(output_obj, "emojis") or []

    # Fallback to manual parsing of content
    if not title:
        parsed = safe_json_parse(content) or {}
        if isinstance(parsed, dict):
            title = parsed.get("title")
            space_label = space_label or parsed.get("spaceLabel") or parsed.get("space_label")
            agent_name = agent_name or parsed.get("agentName") or parsed.get("agent_name")
            emojis = emojis or parsed.get("emojis")
    if not title and thought:
        parsed = safe_json_parse(thought) or {}
        if isinstance(parsed, dict):
            title = parsed.get("title")
            space_label = space_label or parsed.get("spaceLabel") or parsed.get("space_label")
            agent_name = agent_name or parsed.get("agentName") or parsed.get("agent_name")
            emojis = emojis or parsed.get("emojis")

    # Robust cleanup for models like GLM
    if (not title or (isinstance(title, str) and title.strip().startswith("title="))) and content:
        import re
        m_title = re.search(r"title=['\"](.*?)['\"]", content, flags=re.DOTALL)
        if m_title:
            title = m_title.group(1)

        if not emojis:
            m_emojis = re.search(r"emojis=\[[\"']?(.*?)[\"']?\]", content)
            if m_emojis:
                emojis = [m_emojis.group(1)]

        if not space_label:
            m_space = re.search(r"space_?label=['\"](.*?)['\"]", content, re.IGNORECASE) or re.search(r"space=['\"](.*?)['\"]", content, re.IGNORECASE)
            if m_space:
                space_label = m_space.group(1)

        if not agent_name:
            m_agent = re.search(r"agent_?name=['\"](.*?)['\"]", content, re.IGNORECASE) or re.search(r"agent=['\"](.*?)['\"]", content, re.IGNORECASE)
            if m_agent:
                agent_name = m_agent.group(1)

    # Normalize values
    title = _finalize_title(title or content)

    # Strip descriptions if model included them (e.g. "Coding - Help" -> "Coding")
    if space_label and " - " in space_label:
        space_label = space_label.split(" - ")[0].strip()
    if agent_name and " - " in agent_name:
        agent_name = agent_name.split(" - ")[0].strip()

    if not isinstance(emojis, list):
        emojis = []
    emojis = [str(item).strip().strip("'").strip('"') for item in emojis if str(item).strip()][:1]

    return {
        "title": title,
        "spaceLabel": space_label,
        "agentName": agent_name,
        "emojis": emojis,
    }


async def generate_space_and_agent(
    *,
    provider: str,
    first_message: str,
    spaces_with_agents: list[dict[str, Any]],
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> dict[str, Any]:
    space_lines: list[str] = []
    for space in spaces_with_agents or []:
        agent_entries = []
        for agent in space.get("agents") or []:
            if isinstance(agent, str):
                agent_entries.append({"name": agent})
            else:
                agent_entries.append(
                    {"name": agent.get("name", ""), "description": agent.get("description", "")}
                )
        agent_tokens = []
        for agent in agent_entries:
            name = " ".join(_sanitize_option_text(agent.get("name")))
            description = " ".join(_sanitize_option_text(agent.get("description")))
            if name and description:
                agent_tokens.append(f"{name} - {description}")
            elif name:
                agent_tokens.append(name)
        space_label = " ".join(_sanitize_option_text(space.get("label")))
        space_description = " ".join(_sanitize_option_text(space.get("description")))
        space_token = f"{space_label} - {space_description}" if space_description else space_label
        space_lines.append(f"{space_token}:{{{','.join(agent_tokens)}}}")

    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant.\n"
                "## Task\n"
                "1. Select the most appropriate space from the list below and return its spaceLabel.\n"
                "2. If the chosen space has agents, select the best matching agent by agentName. Otherwise return null.\n\n"
                "## Output\n"
                'Return the result as JSON with keys "spaceLabel" and "agentName".'
            ),
        },
        {
            "role": "user",
            "content": f"{first_message}\n\nSpaces and agents:\n" + "\n".join(space_lines),
        },
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, first_message)
    response_format = {"type": "json_object"} if provider != "gemini" else None
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=response_format,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
        # Avoid provider-side grammar cache collisions for lightweight JSON routes.
        # We parse JSON content locally instead of forcing native structured output.
        output_schema=None,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()
    thought = result.get("thought", "").strip()

    output_obj = result.get("output")
    space_label = None
    agent_name = None
    if output_obj:
        space_label = _get_output_value(output_obj, "space_label", "spaceLabel")
        agent_name = _get_output_value(output_obj, "agent_name", "agentName")

    if not space_label:
        parsed = safe_json_parse(content) or {}
        if isinstance(parsed, dict):
            space_label = parsed.get("spaceLabel") or parsed.get("space_label")
            agent_name = agent_name or parsed.get("agentName") or parsed.get("agent_name")
    if not space_label and thought:
        parsed = safe_json_parse(thought) or {}
        if isinstance(parsed, dict):
            space_label = parsed.get("spaceLabel") or parsed.get("space_label")
            agent_name = agent_name or parsed.get("agentName") or parsed.get("agent_name")

    if space_label and " - " in str(space_label):
        space_label = str(space_label).split(" - ")[0].strip()
    if agent_name and " - " in str(agent_name):
        agent_name = str(agent_name).split(" - ")[0].strip()

    return {
        "spaceLabel": space_label,
        "agentName": agent_name,
    }


async def generate_agent_for_auto(
    *,
    provider: str,
    user_message: str,
    current_space: dict[str, Any] | None,
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> str | None:
    agent_entries = []
    for agent in (current_space or {}).get("agents") or []:
        if isinstance(agent, str):
            agent_entries.append({"name": agent})
        else:
            agent_entries.append(
                {"name": agent.get("name", ""), "description": agent.get("description", "")}
            )
    agent_tokens = []
    for agent in agent_entries:
        name = " ".join(_sanitize_option_text(agent.get("name")))
        description = " ".join(_sanitize_option_text(agent.get("description")))
        if name and description:
            agent_tokens.append(f"{name} - {description}")
        elif name:
            agent_tokens.append(name)

    space_label = (current_space or {}).get("label") or "Default"
    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant.\n"
                f"## Task\nSelect the best matching agent for the user's message from the \"{space_label}\" space. "
                "Consider the agent's name and description to determine which one is most appropriate. "
                "If no agent is a good match, return null.\n\n"
                "## Output\nReturn the result as JSON with key \"agentName\" (agent name only, or null)."
            ),
        },
        {
            "role": "user",
            "content": f"{user_message}\n\nAvailable agents in {space_label}:\n" + "\n".join(agent_tokens),
        },
    ]
    messages = _append_time_context(messages, user_timezone, user_locale, user_message)
    response_format = {"type": "json_object"} if provider != "gemini" else None
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=response_format,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
        # Avoid provider-side grammar cache collisions for this lightweight route.
        # We parse JSON content locally instead of forcing native structured output.
        output_schema=None,
    )
    result = await run_agent_completion(request)
    output_obj = result.get("output")
    if output_obj:
        selected_agent = _get_output_value(output_obj, "agentName", "agent_name")
        if selected_agent:
            return selected_agent

    parsed = safe_json_parse(result.get("content", "")) or {}
    if isinstance(parsed, dict):
        return parsed.get("agentName") or parsed.get("agent_name") or None

    thought_parsed = safe_json_parse(result.get("thought", "")) or {}
    if isinstance(thought_parsed, dict):
        return thought_parsed.get("agentName") or thought_parsed.get("agent_name") or None
    return None


def _normalize_related_questions(parsed: Any) -> list[str]:
    def _sanitize_related_question_text(value: Any) -> str:
        if not isinstance(value, str):
            return ""
        cleaned = " ".join(value.split()).strip()
        if not cleaned:
            return ""
        lowered = cleaned.lower()
        if (
            "<|tool_" in lowered
            or "tool_calls_section" in lowered
            or "tool_call_begin" in lowered
            or "tool_call_end" in lowered
            or "endgroup" in lowered
        ):
            return ""
        if cleaned in {"{", "}", "[", "]"}:
            return ""
        if len(cleaned) > 240:
            return ""
        return cleaned

    def _normalize_list(items: list[Any]) -> list[str]:
        out: list[str] = []
        seen: set[str] = set()
        for item in items or []:
            normalized = _sanitize_related_question_text(item)
            if not normalized or normalized in seen:
                continue
            seen.add(normalized)
            out.append(normalized)
            if len(out) >= 3:
                break
        return out

    if isinstance(parsed, list):
        return _normalize_list(parsed)
    if isinstance(parsed, dict):
        if isinstance(parsed.get("questions"), list):
            return _normalize_list(parsed["questions"])
        if isinstance(parsed.get("related_questions"), list):
            return _normalize_list(parsed["related_questions"])
    return []


async def generate_related_questions(
    *,
    provider: str,
    messages: list[dict[str, Any]],
    api_key: str,
    base_url: str | None = None,
    model: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    tool_ids: list[str] | None = None,
    user_tools: list[dict[str, Any]] | None = None,
    tool_choice: Any = None,
    response_format: dict[str, Any] | None = None,
    thinking: dict[str, Any] | bool | None = None,
    temperature: float | None = None,
    top_k: int | None = None,
    top_p: float | None = None,
    frequency_penalty: float | None = None,
    presence_penalty: float | None = None,
    context_message_limit: int | None = None,
    search_provider: str | None = None,
    tavily_api_key: str | None = None,
    user_timezone: str | None = None,
    user_locale: str | None = None,
) -> list[str]:
    prompt_messages = [
        *(messages or []),
        {
            "role": "user",
            "content": (
                "Based on our conversation, suggest 3 short, relevant follow-up questions I (the user) might ask you next. "
                "The questions MUST be from the user's perspective (e.g., 'How does X work?', 'Tell me more about Y'). "
                "Do NOT generate questions from the assistant's perspective (e.g., 'Do you want to know about...?'). "
                "Return the result as JSON with a 'questions' key containing the array of strings."
            ),
        },
    ]
    prompt_messages = _append_time_context(prompt_messages, user_timezone, user_locale, "related questions")
    response_format = {"type": "json_object"} if provider != "gemini" else None
    request = StreamChatRequest(
        provider=provider,
        apiKey=api_key,
        baseUrl=base_url,
        model=model,
        messages=prompt_messages,
        tools=tools or [],
        toolChoice=tool_choice,
        toolIds=tool_ids or [],
        userTools=user_tools or [],
        responseFormat=response_format,
        # Avoid provider-side grammar cache collisions for this lightweight route.
        # We parse JSON content locally instead of forcing native structured output.
        output_schema=None,
        thinking=thinking,
        temperature=temperature,
        top_k=top_k,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        contextMessageLimit=context_message_limit,
        searchProvider=search_provider,
        tavilyApiKey=tavily_api_key,
        skipDefaultTools=True,
        stream=True,
    )
    result = await run_agent_completion(request)
    content = result.get("content", "").strip()
    thought = result.get("thought", "").strip()
    output_obj = result.get("output")

    questions = []
    if output_obj:
        questions = _get_output_value(output_obj, "questions") or []

    if not questions:
        parsed = safe_json_parse(content)
        # If model returned a list directly, use it
        if isinstance(parsed, list):
            questions = parsed
        else:
            questions = _normalize_related_questions(parsed)
    if not questions and thought:
        thought_parsed = safe_json_parse(thought)
        if isinstance(thought_parsed, list):
            questions = thought_parsed
        else:
            questions = _normalize_related_questions(thought_parsed)

    # Robust cleanup for weird formats
    if not questions and content:
        import re
        # Try to find questions=["..." or questions=['...']
        match = re.search(r"questions=\[([\s\S]*?)\]", content)
        if match:
            raw_list = match.group(1)
            questions = [q.strip().strip("'").strip('"') for q in re.findall(r"['\"](.*?)['\"]", raw_list)]

        # New Fallback: Try to find numbered lists like 1. "Question" or 1. **Question**
        if not questions:
            # Matches strings starting with number, dot, maybe whitespace, maybe quotes/stars, then text, then closing quotes/stars
            numbered_matches = re.findall(r"^\d+\.\s+[*\"']+(.*?)[*\"']+", content, re.MULTILINE)
            if numbered_matches:
                questions = [q.strip() for q in numbered_matches]
            else:
                # Last ditch: simplified numbered list without quotes
                numbered_matches_simple = re.findall(r"^\d+\.\s+(.*?)$", content, re.MULTILINE)
                if numbered_matches_simple:
                    questions = [q.strip().strip('*').strip('"').strip("'") for q in numbered_matches_simple]

    return _normalize_related_questions(questions)