Presa-Mano/src/presa_mano/api/agent_routes.py

"""AI assistant routes.

- `POST /api/agent/chat`  — send a user message, get back the assistant reply
  and the trace of every tool the model invoked. Stateless on the server side:
  the GUI keeps the rolling history and posts it back in `history`.

- `GET  /api/agent/tools` — list of tools exposed to the model (debug aid).

- `GET/POST/DELETE /api/agent/tasks` — persistent registry of named prompt
  templates so frequently-used tasks ("calibra braccio dx", "sposta a home
  position", …) can be re-launched in one click without retyping the prompt.
"""

from __future__ import annotations

import json
import re
import time
from dataclasses import asdict
from typing import Any, Optional

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field

# Submodule imports (not `from ..agent import …`) to break a circular import:
# api/__init__.py is loaded first by main.py and pulls in agent_routes; if we
# imported via the agent package, agent/__init__.py would try to load
# agent.runner → agent.tools → ..api._deps while ..api is still mid-init.
from ..agent.runner import AgentRunner, OpenRouterError
from ..agent.tools import TOOL_DEFINITIONS
from ..config import data_dir, get_config

router = APIRouter()


# ---------- chat ----------

class ChatMessage(BaseModel):
    role: str
    content: Optional[str] = None
    # Pass-through fields used by the OpenAI tool-calling protocol.
    tool_calls: Optional[list[dict[str, Any]]] = None
    tool_call_id: Optional[str] = None
    name: Optional[str] = None


class ChatReq(BaseModel):
    message: str = Field(..., min_length=1, max_length=4000)
    history: list[ChatMessage] = Field(default_factory=list)


@router.get("/tools")
def list_tools() -> dict[str, Any]:
    """Tools exposed to the LLM. Useful for debugging the system prompt."""
    return {"tools": TOOL_DEFINITIONS}


@router.get("/health")
def agent_health() -> dict[str, Any]:
    """Reports whether OpenRouter credentials are available and which model
    is configured. The frontend uses this to grey-out the chat input when
    the env var is missing."""
    runner = AgentRunner()
    return {
        "configured": runner.has_credentials(),
        "model": runner.model,
        "api_key_env": runner.api_key_env,
        "base_url": runner.base_url,
        "temperature": runner.temperature,
        "max_tool_iterations": runner.max_tool_iterations,
        "timeout_s": runner.timeout_s,
    }


@router.get("/system_prompt_preview")
def system_prompt_preview() -> dict[str, Any]:
    """Returns the exact system prompt the model will receive at the next
    chat. Useful for the AI Settings tab so the operator can preview the
    effect of editing chat.openrouter.user_prompt_addendum or any other
    state that influences the prompt (active robot, marker config…)."""
    from ..agent.prompt import build_system_prompt
    return {"system_prompt": build_system_prompt()}


# Curated list shown in the AI Settings model selector. Plus a free-text
# field on the GUI side for anything not in the list. Keep the list short
# and focused on models that we've verified support OpenRouter's tool-use.
_KNOWN_MODELS = [
    {"id": "anthropic/claude-sonnet-4.5",        "label": "Claude Sonnet 4.5 (consigliato)"},
    {"id": "anthropic/claude-opus-4.1",          "label": "Claude Opus 4.1 (più capace, più caro)"},
    {"id": "anthropic/claude-haiku-4.5",         "label": "Claude Haiku 4.5 (veloce, economico)"},
    {"id": "openai/gpt-4o",                       "label": "GPT-4o"},
    {"id": "openai/gpt-4o-mini",                  "label": "GPT-4o mini"},
    {"id": "google/gemini-2.5-pro",              "label": "Gemini 2.5 Pro"},
    {"id": "google/gemini-2.5-flash",            "label": "Gemini 2.5 Flash"},
    {"id": "meta-llama/llama-3.3-70b-instruct",  "label": "Llama 3.3 70B"},
    {"id": "mistralai/mistral-large-2411",        "label": "Mistral Large 2411"},
]


@router.get("/known_models")
def known_models() -> dict[str, Any]:
    """Curated short-list of OpenRouter models that work with our tool-calling.
    The GUI also offers a free-text input for anything not in this list."""
    return {"models": _KNOWN_MODELS}


@router.post("/chat")
def chat(req: ChatReq) -> dict[str, Any]:
    runner = AgentRunner()
    if not runner.has_credentials():
        raise HTTPException(status_code=412,
                            detail=f"OpenRouter API key non impostata (env var {runner.api_key_env}).")
    history = [m.model_dump(exclude_none=True) for m in req.history]
    try:
        result, updated_history = runner.run(req.message, history=history)
    except OpenRouterError as e:
        raise HTTPException(status_code=502, detail=str(e))
    return {
        "reply": result.reply,
        "iterations": result.iterations,
        "finish_reason": result.finish_reason,
        "model": result.model,
        "usage": result.usage,
        "tool_calls": [asdict(t) for t in result.tool_calls],
        "history": updated_history,
    }


# ---------- saved task templates ----------

_NAME_RE = re.compile(r"^[a-zA-Z0-9 _\-àèéìòù]{1,80}$")


class TaskItem(BaseModel):
    name: str = Field(..., min_length=1, max_length=80)
    prompt: str = Field(..., min_length=1, max_length=4000)
    description: Optional[str] = Field(None, max_length=400)


def _tasks_path():
    return data_dir() / "agent_tasks.json"


def _load_tasks() -> list[dict[str, Any]]:
    p = _tasks_path()
    if not p.exists():
        return []
    try:
        with open(p, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, list) else []
    except json.JSONDecodeError:
        return []


def _save_tasks(items: list[dict[str, Any]]) -> None:
    p = _tasks_path()
    p.parent.mkdir(parents=True, exist_ok=True)
    with open(p, "w", encoding="utf-8") as f:
        json.dump(items, f, ensure_ascii=False, indent=2)


@router.get("/tasks")
def list_tasks() -> dict[str, Any]:
    return {"tasks": _load_tasks()}


@router.post("/tasks")
def save_task(req: TaskItem) -> dict[str, Any]:
    if not _NAME_RE.match(req.name):
        raise HTTPException(status_code=422, detail="nome non valido (max 80 char, alfanumerici/spazi/_-)")
    items = _load_tasks()
    now = time.time()
    # Replace by name if already exists, otherwise append.
    found = False
    for it in items:
        if it.get("name") == req.name:
            it["prompt"] = req.prompt
            it["description"] = req.description
            it["updated_at"] = now
            found = True
            break
    if not found:
        items.append({
            "name": req.name,
            "prompt": req.prompt,
            "description": req.description,
            "created_at": now,
            "updated_at": now,
        })
    _save_tasks(items)
    return {"ok": True, "total": len(items), "name": req.name}


@router.delete("/tasks/{name}")
def delete_task(name: str) -> dict[str, Any]:
    items = _load_tasks()
    new_items = [it for it in items if it.get("name") != name]
    if len(new_items) == len(items):
        raise HTTPException(status_code=404, detail=f"task '{name}' non trovata")
    _save_tasks(new_items)
    return {"ok": True, "total": len(new_items)}