memory/sync.py

"""
sync.py — Memory Sync Protocol

Sincronizza in modo idempotente la memoria locale del browser con il backend.
Il protocollo accetta batch di elementi firmati da source/id/versione e salva
il contenuto nei layer episodic/semantic senza richiedere dipendenze aggiuntive.

GAP-VAULT-AUTH (fix): aggiunge _require_sync_auth (Bearer VAULT_ADMIN_TOKEN) su
push / pull / export / import — impedisce dump completo della memoria semantica
via curl non autenticata.
"""
from __future__ import annotations

import hashlib
import os
import time
from typing import Any, Iterable, Literal, Optional

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


MemoryLayer = Literal["working", "episodic", "semantic", "reflection"]
MemoryType = Literal["fact", "preference", "summary", "skill", "episode", "reflection"]


class SyncMemoryItem(BaseModel):
    id: str = Field(..., min_length=1)
    content: str = Field(..., min_length=1)
    type: MemoryType = "fact"
    layer: MemoryLayer = "semantic"
    topic: str | None = None
    weight: float = 1.0
    created_at: int | None = None
    updated_at: int | None = None
    source: str = "browser"
    metadata: dict[str, Any] = Field(default_factory=dict)


class MemorySyncPushRequest(BaseModel):
    device_id: str = "browser"
    since: int | None = None
    items: list[SyncMemoryItem] = Field(default_factory=list)


class MemorySyncPullRequest(BaseModel):
    device_id: str = "browser"
    query: str | None = None
    since: int | None = None
    limit: int = Field(default=50, ge=1, le=200)
    layer: MemoryLayer | None = None


class MemorySyncStatus(BaseModel):
    ok: bool
    protocol: str = "memory-sync-v1"
    server_time: int
    stats: dict[str, Any]


class MemoryImportRequest(BaseModel):
    """Payload di import definito a livello modulo per lo schema OpenAPI."""
    records: list[dict[str, Any]] = Field(default_factory=list)
    overwrite: bool = False


# ── GAP-VAULT-AUTH: autenticazione Bearer ─────────────────────────────────────
_SYNC_ADMIN_TOKEN = os.getenv('VAULT_ADMIN_TOKEN', '')  # stessa variabile del vault


async def _require_sync_auth(authorization: Optional[str] = Header(None)) -> None:
    """Richiede Authorization: Bearer VAULT_ADMIN_TOKEN se configurato.

    Protegge push/pull/export/import da dump non autenticati via curl.
    /status rimane pubblico (nessun dato esposto, solo statistiche aggregate).
    """
    # GAP-VAULT-AUTH-STRICT: fail-closed se VAULT_ADMIN_TOKEN non è impostata (tranne in local dev)
    if not _SYNC_ADMIN_TOKEN:
        if os.getenv('ENV', 'production') == 'development':
            return
        raise HTTPException(
            status_code=500,
            detail='Memory sync configuration error: admin token missing',
        )
    if authorization != f'Bearer {_SYNC_ADMIN_TOKEN}':
        raise HTTPException(
            status_code=401,
            detail='Memory sync: non autorizzato — Bearer token non valido o mancante',
        )


def _now_ms() -> int:
    return int(time.time() * 1000)


def _fingerprint(item: SyncMemoryItem) -> str:
    raw = f"{item.source}:{item.id}:{item.updated_at or item.created_at or 0}:{item.content}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()[:16]


def _iter_local_items(memory: Any, layer: MemoryLayer | None, limit: int) -> Iterable[dict[str, Any]]:
    """Best-effort export from available memory layers."""
    if layer in (None, "working") and hasattr(memory, "working"):
        # S571-GAP8: WorkingMemory usa self._buf (deque), NON entries.
        recent = memory.working.get_recent(n=limit)
        for idx, entry in enumerate(recent):
            content = entry.get("content", str(entry))
            role    = entry.get("role", "unknown")
            yield {
                "id": f"working-{idx}",
                "content": content,
                "type": "episode",
                "layer": "working",
                "topic": role,
                "weight": 1,
                "updated_at": _now_ms(),
                "source": "backend",
                "metadata": {"role": role},
            }


def create_memory_sync_router(memory: Any) -> APIRouter:
    router = APIRouter(prefix="/api/memory/sync", tags=["memory-sync"])

    @router.get("/status", response_model=MemorySyncStatus)
    async def sync_status() -> MemorySyncStatus:
        """Pubblico: espone solo statistiche aggregate, nessun contenuto."""
        return MemorySyncStatus(ok=True, server_time=_now_ms(), stats=memory.stats())

    @router.post("/push")
    async def sync_push(
        req: MemorySyncPushRequest,
        _auth: None = Depends(_require_sync_auth),
    ) -> dict[str, Any]:
        accepted: list[str] = []
        skipped: list[dict[str, str]] = []
        now = _now_ms()

        for item in req.items:
            if not item.content.strip():
                skipped.append({"id": item.id, "reason": "empty-content"})
                continue

            item.metadata.setdefault("sync_id", item.id)
            item.metadata.setdefault("device_id", req.device_id)
            item.metadata.setdefault("fingerprint", _fingerprint(item))
            item.metadata.setdefault("synced_at", now)
            item.metadata.setdefault("source", item.source)
            item.metadata.setdefault("topic", item.topic or "generale")

            try:
                if item.layer == "working" and hasattr(memory, "working"):
                    role = str(item.metadata.get("role") or item.type)
                    memory.working.add(role, item.content)
                elif item.layer == "episodic":
                    await memory.save_episode(item.type, item.topic or item.id, item.content, True, tags=["sync", req.device_id])
                else:
                    if getattr(memory.semantic, "available", False):
                        memory.semantic.add(item.content, {"type": item.type, **item.metadata})
                    else:
                        await memory.save_episode(item.type, item.topic or item.id, item.content, True, tags=["sync", req.device_id])
                accepted.append(item.id)
            except Exception as exc:
                skipped.append({"id": item.id, "reason": str(exc)[:300]})

        return {
            "ok": True,
            "protocol": "memory-sync-v1",
            "accepted": accepted,
            "skipped": skipped,
            "server_time": now,
            "stats": memory.stats(),
        }

    @router.post("/pull")
    async def sync_pull(
        req: MemorySyncPullRequest,
        _auth: None = Depends(_require_sync_auth),
    ) -> dict[str, Any]:
        query = (req.query or "").strip()
        items: list[dict[str, Any]] = []

        if query:
            results = await memory.search(query, req.limit, layer=req.layer)
            for idx, hit in enumerate(results):
                content = str(hit.get("content", ""))
                items.append({
                    "id": str(hit.get("id") or f"backend-search-{idx}"),
                    "content": content,
                    "type": str(hit.get("type") or "fact"),
                    "layer": str(hit.get("layer") or req.layer or "semantic"),
                    "topic": str(hit.get("topic") or "generale"),
                    "weight": float(hit.get("similarity") or 1),
                    "updated_at": _now_ms(),
                    "source": "backend",
                    "metadata": {k: v for k, v in hit.items() if k not in {"content"}},
                })
        else:
            items.extend(list(_iter_local_items(memory, req.layer, req.limit)))

        return {
            "ok": True,
            "protocol": "memory-sync-v1",
            "items": items[: req.limit],
            "server_time": _now_ms(),
            "stats": memory.stats(),
        }

    @router.get("/export")
    async def memory_export(
        limit: int = 2000,
        _auth: None = Depends(_require_sync_auth),
    ) -> dict[str, Any]:
        """Esporta l'intera semantic memory. Richiede Bearer VAULT_ADMIN_TOKEN."""
        if not getattr(memory.semantic, "available", False):
            return {"ok": False, "error": "semantic memory non disponibile", "records": [], "count": 0}
        import asyncio as _asyncio
        records = await _asyncio.to_thread(memory.semantic.export_all, limit)
        return {
            "ok": True,
            "protocol": "semantic-export-v1",
            "count": len(records),
            "records": records,
            "server_time": _now_ms(),
        }

    @router.post("/import")
    async def memory_import(
        req: MemoryImportRequest,
        _auth: None = Depends(_require_sync_auth),
    ) -> dict[str, Any]:
        """Importa records nella semantic memory. Richiede Bearer VAULT_ADMIN_TOKEN."""
        if not getattr(memory.semantic, "available", False):
            return {"ok": False, "error": "semantic memory non disponibile", "imported": 0, "skipped": 0, "total": 0}
        if not req.records:
            return {"ok": True, "imported": 0, "skipped": 0, "total": 0, "server_time": _now_ms()}
        import asyncio as _asyncio
        result = await _asyncio.to_thread(memory.semantic.import_all, req.records, req.overwrite)
        return {
            "ok": True,
            "protocol": "semantic-export-v1",
            **result,
            "server_time": _now_ms(),
        }

    return router