api/private_state.py

"""API private per il cutover browser → backend delle tabelle Supabase sensibili.

Questi endpoint sono destinati esclusivamente alle Pages Functions, che inoltrano
``X-Internal-Token`` al master B. Nessun client browser riceve una service-role key
o accede direttamente alle tabelle private.
"""
from __future__ import annotations

import asyncio
import json
import logging
import math
import re
import time
from datetime import datetime, timedelta, timezone
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field, field_validator

from .auth_guard import require_private_state_machine
from .state import get_supabase

_logger = logging.getLogger("agente_ai.api.private_state")
router = APIRouter(
    prefix="/api/private-state",
    tags=["private-state"],
    dependencies=[Depends(require_private_state_machine)],
)

_RAG_LANGUAGE = "rag_chunk"
_RAG_PREFIX = "__rag_chunk"
_MAX_RAG_CHUNKS = 100
_MAX_RAG_CONTENT_CHARS = 10_000
_MAX_EMBEDDING_DIMENSIONS = 4_096
_MAX_TASK_PAGE = 100


def _db() -> Any:
    """Restituisce il client service-role del backend o un errore non sensibile."""
    client = get_supabase()
    if client is None:
        raise HTTPException(status_code=503, detail="Archivio privato temporaneamente non disponibile")
    return client


async def _call(operation):
    """Esegue il client sincrono Supabase senza bloccare l'event loop FastAPI."""
    try:
        return await asyncio.to_thread(operation, _db())
    except HTTPException:
        raise
    except Exception as exc:  # Non esporre dettagli backend, query o dati al browser.
        _logger.warning("[private-state] database operation failed: %s", type(exc).__name__)
        raise HTTPException(status_code=502, detail="Operazione sullo stato privato non riuscita") from exc


def _json_object(value: object) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
            return parsed if isinstance(parsed, dict) else {}
        except (TypeError, ValueError):
            return {}
    return {}


def _as_epoch_ms(value: object) -> int:
    """Normalizza i valori `timestamptz` PostgREST in millisecondi browser-safe."""
    if isinstance(value, (int, float)):
        return int(value)
    if isinstance(value, datetime):
        moment = value
    elif isinstance(value, str):
        try:
            moment = datetime.fromisoformat(value.replace("Z", "+00:00"))
        except ValueError:
            return 0
    else:
        return 0
    if moment.tzinfo is None:
        moment = moment.replace(tzinfo=timezone.utc)
    return int(moment.timestamp() * 1_000)


def _finite_vector(values: list[float]) -> list[float]:
    if not values or len(values) > _MAX_EMBEDDING_DIMENSIONS:
        raise ValueError("dimensione embedding non valida")
    if any(not math.isfinite(value) for value in values):
        raise ValueError("embedding contiene valori non finiti")
    return values


class TelegramConfigIn(BaseModel):
    bot_token: str = Field(min_length=1, max_length=512)
    chat_id: str = Field(min_length=1, max_length=128)


class SkillPatternIn(BaseModel):
    id: str = Field(min_length=1, max_length=128)
    task_signature: str = Field(min_length=1, max_length=200)
    tool_sequence: list[str] = Field(min_length=1, max_length=8)
    success_count: int = Field(ge=0, le=1_000_000)
    total_count: int = Field(ge=1, le=1_000_000)
    last_used: int = Field(ge=0)
    confidence: float = Field(ge=0, le=1)

    @field_validator("tool_sequence")
    @classmethod
    def validate_tools(cls, tools: list[str]) -> list[str]:
        clean = [tool.strip()[:120] for tool in tools if isinstance(tool, str) and tool.strip()]
        if not clean:
            raise ValueError("tool_sequence non valida")
        return clean


class RagChunkIn(BaseModel):
    id: str = Field(min_length=1, max_length=128)
    path: str = Field(min_length=1, max_length=256)
    content: str = Field(min_length=51, max_length=_MAX_RAG_CONTENT_CHARS)
    embedding: list[float] | None = Field(default=None, max_length=_MAX_EMBEDDING_DIMENSIONS)

    @field_validator("embedding")
    @classmethod
    def validate_embedding(cls, value: list[float] | None) -> list[float] | None:
        return _finite_vector(value) if value is not None else None


class RagIndexIn(BaseModel):
    file_id: str = Field(min_length=1, max_length=40)
    chunks: list[RagChunkIn] = Field(min_length=1, max_length=_MAX_RAG_CHUNKS)

    @field_validator("file_id")
    @classmethod
    def validate_file_id(cls, value: str) -> str:
        if not re.fullmatch(r"[a-z0-9_]+", value):
            raise ValueError("file_id non valido")
        return value


class RagSearchIn(BaseModel):
    query_embedding: list[float] | None = Field(default=None, max_length=_MAX_EMBEDDING_DIMENSIONS)
    query: str = Field(default="", max_length=2_000)
    similarity_threshold: float = Field(default=0.22, ge=-1, le=1)
    match_count: int = Field(default=5, ge=1, le=10)

    @field_validator("query_embedding")
    @classmethod
    def validate_query_embedding(cls, value: list[float] | None) -> list[float] | None:
        return _finite_vector(value) if value is not None else None

    @field_validator("query")
    @classmethod
    def validate_query(cls, value: str) -> str:
        if not value.strip() and value == "":
            return ""
        return value.strip()


@router.get("/sessions")
async def list_sessions(
    max_age_ms: int = Query(default=300_000, ge=10_000, le=3_600_000),
    limit: int = Query(default=100, ge=1, le=200),
) -> dict[str, object]:
    """Restituisce esclusivamente i metadati delle sessioni agente ancora vive."""
    cutoff = (datetime.now(timezone.utc) - timedelta(milliseconds=max_age_ms)).isoformat()

    def operation(client: Any):
        return client.table("agent_tasks").select("task_id,context,updated_at") \
            .eq("status", "__session__").gte("updated_at", cutoff) \
            .order("updated_at", desc=True).limit(limit).execute()

    result = await _call(operation)
    sessions: list[dict[str, object]] = []
    for row in result.data or []:
        context = _json_object(row.get("context"))
        session_id = str(context.get("sessionId") or row.get("task_id") or "").strip()
        if not session_id:
            continue
        claimed = context.get("claimedFiles")
        sessions.append({
            "session_id": session_id,
            "session_name": str(context.get("sessionName") or session_id)[:160],
            "sprint": str(context["sprint"])[:120] if context.get("sprint") else None,
            "claimed_files": [str(item)[:300] for item in claimed[:100]] if isinstance(claimed, list) else [],
            "last_heartbeat": _as_epoch_ms(context.get("lastHeartbeat")) or _as_epoch_ms(row.get("updated_at")),
            "current_task": str(context["currentTask"])[:500] if context.get("currentTask") else None,
        })
    return {"sessions": sessions}


@router.get("/tasks")
async def list_tasks(
    limit: int = Query(default=20, ge=1, le=_MAX_TASK_PAGE),
    offset: int = Query(default=0, ge=0, le=10_000),
    status: str | None = Query(default=None, max_length=64),
) -> dict[str, object]:
    """Lista task non di configurazione per il TMA, con conteggi per stato."""
    normalized_status = status.strip().upper() if status else ""

    def operation(client: Any):
        query = client.table("agent_tasks").select("task_id,goal,status,updated_at") \
            .neq("status", "__session__").neq("status", "__config__")
        if normalized_status:
            query = query.eq("status", normalized_status)
        page = query.order("updated_at", desc=True).range(offset, offset + limit - 1).execute()
        all_statuses = client.table("agent_tasks").select("status") \
            .neq("status", "__session__").neq("status", "__config__").limit(2_000).execute()
        return page, all_statuses

    page, all_statuses = await _call(operation)
    counts: dict[str, int] = {}
    for row in all_statuses.data or []:
        key = str(row.get("status") or "UNKNOWN").upper()
        counts[key] = counts.get(key, 0) + 1
    tasks = [
        {
            "task_id": str(row.get("task_id") or ""),
            "goal": str(row.get("goal") or "")[:1_000],
            "status": str(row.get("status") or "UNKNOWN"),
            "updated_at": _as_epoch_ms(row.get("updated_at")),
        }
        for row in page.data or []
    ]
    return {"tasks": tasks, "counts": counts, "offset": offset, "limit": limit}


@router.post("/telegram-config")
async def save_telegram_config(payload: TelegramConfigIn) -> dict[str, bool]:
    """Salva la configurazione Telegram nel record privato del daemon."""
    now = datetime.now(timezone.utc).isoformat()
    row = {
        "task_id": "__telegram_config__",
        "goal": "__telegram_config__",
        "status": "__config__",
        "max_steps": 0,
        "context": json.dumps({"botToken": payload.bot_token, "chatId": payload.chat_id}),
        "updated_at": now,
    }

    def operation(client: Any):
        return client.table("agent_tasks").upsert(row, on_conflict="task_id").execute()

    await _call(operation)
    return {"ok": True}


@router.get("/skill-patterns")
async def list_skill_patterns(limit: int = Query(default=100, ge=1, le=100)) -> dict[str, object]:
    """Carica pattern cloud per il merge con lo storage locale Dexie."""

    def operation(client: Any):
        return client.table("skill_patterns").select(
            "id,task_signature,tool_sequence,success_count,total_count,last_used,confidence"
        ).order("confidence", desc=True).limit(limit).execute()

    result = await _call(operation)
    return {"patterns": result.data or []}


@router.put("/skill-patterns/{pattern_id}")
async def upsert_skill_pattern(pattern_id: str, payload: SkillPatternIn) -> dict[str, bool]:
    """Sincronizza un pattern già validato dal layer locale del browser."""
    if pattern_id != payload.id:
        raise HTTPException(status_code=400, detail="Identificatore pattern non coerente")

    row = payload.model_dump()

    def operation(client: Any):
        return client.table("skill_patterns").upsert(row, on_conflict="id").execute()

    await _call(operation)
    return {"ok": True}


@router.post("/rag/index")
async def index_rag(payload: RagIndexIn) -> dict[str, int]:
    """Sostituisce i chunk RAG di un file senza esporre `vfs_files` al browser."""
    prefix = f"{_RAG_PREFIX}/{payload.file_id}/"
    rows: list[dict[str, object]] = []
    now = int(time.time() * 1000)
    for chunk in payload.chunks:
        if not chunk.path.startswith(prefix) or not chunk.id.startswith(f"rag-{payload.file_id}-"):
            raise HTTPException(status_code=400, detail="Chunk RAG non coerente con il file")
        row: dict[str, object] = {
            "id": chunk.id,
            "user_id": "default",
            "path": chunk.path,
            "content": chunk.content,
            "language": _RAG_LANGUAGE,
            "created_at": now,
            "updated_at": now,
        }
        if chunk.embedding:
            row["embedding"] = "[" + ",".join(str(value) for value in chunk.embedding) + "]"
            row["embedding_vec"] = chunk.embedding
        rows.append(row)

    def operation(client: Any):
        client.table("vfs_files").delete().eq("language", _RAG_LANGUAGE).like("path", prefix + "%").execute()
        return client.table("vfs_files").upsert(rows).execute()

    await _call(operation)
    return {"indexed": len(rows)}


def _parse_vector(value: object) -> list[float] | None:
    if isinstance(value, list):
        try:
            return [float(item) for item in value]
        except (TypeError, ValueError):
            return None
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
            return [float(item) for item in parsed] if isinstance(parsed, list) else None
        except (TypeError, ValueError):
            return None
    return None


def _cosine_similarity(left: list[float], right: list[float]) -> float:
    if len(left) != len(right) or not left:
        return 0.0
    numerator = sum(a * b for a, b in zip(left, right))
    left_norm = math.sqrt(sum(a * a for a in left))
    right_norm = math.sqrt(sum(b * b for b in right))
    return numerator / (left_norm * right_norm) if left_norm and right_norm else 0.0


@router.post("/rag/search")
async def search_rag(payload: RagSearchIn) -> dict[str, object]:
    """Ricerca pgvector con fallback server-side alla similarità coseno in memoria."""

    def operation(client: Any):
        if payload.query_embedding:
            try:
                rpc = client.rpc("match_rag_chunks", {
                    "query_embedding": payload.query_embedding,
                    "similarity_threshold": payload.similarity_threshold,
                    "match_count": payload.match_count,
                }).execute()
                if isinstance(rpc.data, list):
                    return {"mode": "pgvector", "rows": rpc.data}
            except Exception as exc:
                _logger.info("[private-state] rag RPC unavailable, using fallback: %s", type(exc).__name__)

        result = client.table("vfs_files").select("content,embedding,path") \
            .eq("language", _RAG_LANGUAGE).limit(300).execute()
        query_words = {word for word in re.split(r"\W+", payload.query.lower()) if len(word) > 3}
        scored: list[dict[str, object]] = []
        for row in result.data or []:
            content = str(row.get("content") or "")
            embedding = _parse_vector(row.get("embedding"))
            if payload.query_embedding and embedding:
                score = _cosine_similarity(payload.query_embedding, embedding)
            else:
                lower = content.lower()
                hits = sum(1 for word in query_words if word in lower)
                score = hits / len(query_words) if query_words else 0.0
            if score >= payload.similarity_threshold:
                scored.append({
                    "content": content,
                    "similarity": score,
                    "path": str(row.get("path") or ""),
                })
        scored.sort(key=lambda item: float(item["similarity"]), reverse=True)
        return {"mode": "cosine_fallback", "rows": scored[:payload.match_count]}

    result = await _call(operation)
    rows = [
        {
            "content": str(row.get("content") or ""),
            "similarity": float(row.get("similarity") or 0),
            "path": str(row.get("path") or ""),
        }
        for row in result["rows"]
        if isinstance(row, dict)
    ]
    return {"results": rows, "mode": result["mode"]}