ss-tools/backend/src/api/routes/assistant/_llm_planner_intent.py

# #region AssistantLlmPlannerIntent [C:3] [TYPE Module]
# @BRIEF LLM-based intent planning and authorization for the assistant API — separated from tool catalog.
# @LAYER API
# @INVARIANT Production deployments always require confirmation.
# @RELATION DEPENDS_ON -> [AssistantLlmPlanner]
# @RELATION DEPENDS_ON -> [AssistantResolvers]

from __future__ import annotations

import json
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from src.core.logger import belief_scope, logger
from src.core.config_manager import ConfigManager
from src.schemas.auth import User
from src.services.llm_provider import LLMProviderService
from src.services.llm_prompt_templates import (
    normalize_llm_settings,
    resolve_bound_provider_id,
)
from src.plugins.llm_analysis.service import LLMClient
from src.plugins.llm_analysis.models import LLMProviderType
from ._schemas import INTENT_PERMISSION_CHECKS
from ._resolvers import (
    _is_production_env,
    _resolve_provider_id,
)
from ._llm_planner import (
    _check_any_permission,
    _coerce_intent_entities,
)


# #region _plan_intent_with_llm [C:2] [TYPE Function]
# @BRIEF Use active LLM provider to select best tool/operation from dynamic catalog.
# @PRE tools list contains allowed operations for current user.
# @POST Returns normalized intent dict when planning succeeds; otherwise None.
async def _plan_intent_with_llm(
    message: str,
    tools: List[Dict[str, Any]],
    db: Session,
    config_manager: ConfigManager,
) -> Optional[Dict[str, Any]]:
    if not tools:
        return None

    llm_settings = normalize_llm_settings(config_manager.get_config().settings.llm)
    planner_provider_token = llm_settings.get("assistant_planner_provider")
    planner_model_override = llm_settings.get("assistant_planner_model")
    llm_service = LLMProviderService(db)
    providers = llm_service.get_all_providers()
    provider_id = _resolve_provider_id(planner_provider_token, db)
    provider = next((p for p in providers if p.id == provider_id), None)
    if not provider:
        return None
    api_key = llm_service.get_decrypted_api_key(provider.id)
    if not api_key:
        return None

    planner = LLMClient(
        provider_type=LLMProviderType(provider.provider_type),
        api_key=api_key,
        base_url=provider.base_url,
        default_model=planner_model_override or provider.default_model,
    )

    system_instruction = (
        "You are a deterministic intent planner for backend tools.\n"
        "Choose exactly one operation from available_tools or return clarify.\n"
        "Output strict JSON object:\n"
        "{"
        '"domain": string, '
        '"operation": string, '
        '"entities": object, '
        '"confidence": number, '
        '"risk_level": "safe"|"guarded"|"dangerous", '
        '"requires_confirmation": boolean'
        "}\n"
        "Rules:\n"
        "- Use only operation names from available_tools.\n"
        '- If input is ambiguous, operation must be "clarify" with low confidence.\n'
        "- If dashboard is provided as name/slug (e.g., COVID), put it into entities.dashboard_ref.\n"
        "- Keep entities minimal and factual.\n"
    )
    payload = {
        "available_tools": tools,
        "user_message": message,
        "known_environments": [
            {"id": e.id, "name": e.name} for e in config_manager.get_environments()
        ],
    }
    try:
        response = await planner.get_json_completion(
            [
                {"role": "system", "content": system_instruction},
                {"role": "user", "content": json.dumps(payload, ensure_ascii=False)},
            ]
        )
    except Exception as exc:
        import traceback

        logger.warning(
            f"[assistant.planner][fallback] LLM planner unavailable: {exc}\n{traceback.format_exc()}"
        )
        return None
    if not isinstance(response, dict):
        return None

    operation = response.get("operation")
    valid_ops = {tool["operation"] for tool in tools}
    if operation == "clarify":
        return {
            "domain": "unknown",
            "operation": "clarify",
            "entities": {},
            "confidence": float(response.get("confidence", 0.3)),
            "risk_level": "safe",
            "requires_confirmation": False,
        }
    if operation not in valid_ops:
        return None

    by_operation = {tool["operation"]: tool for tool in tools}
    selected = by_operation[operation]
    intent = {
        "domain": response.get("domain") or selected["domain"],
        "operation": operation,
        "entities": response.get("entities", {}),
        "confidence": float(response.get("confidence", 0.75)),
        "risk_level": response.get("risk_level") or selected["risk_level"],
        "requires_confirmation": bool(
            response.get("requires_confirmation", selected["requires_confirmation"])
        ),
    }
    intent = _coerce_intent_entities(intent)

    defaults = selected.get("defaults") or {}
    for key, value in defaults.items():
        if value and not intent["entities"].get(key):
            intent["entities"][key] = value

    if operation in {"deploy_dashboard", "execute_migration"}:
        env_token = intent["entities"].get("environment") or intent["entities"].get(
            "target_env"
        )
        if _is_production_env(env_token, config_manager):
            intent["risk_level"] = "dangerous"
            intent["requires_confirmation"] = True
    return intent


# #endregion _plan_intent_with_llm


# #region _authorize_intent [C:2] [TYPE Function]
# @BRIEF Validate user permissions for parsed intent before confirmation/dispatch.
# @PRE intent.operation is present for known assistant command domains.
# @POST Returns if authorized; raises HTTPException(403) when denied.
def _authorize_intent(intent: Dict[str, Any], current_user: User):
    operation = intent.get("operation")
    if operation in INTENT_PERMISSION_CHECKS:
        _check_any_permission(current_user, INTENT_PERMISSION_CHECKS[operation])


# #endregion _authorize_intent


# #endregion AssistantLlmPlannerIntent