api/integrity_manager.py

"""
backend/api/integrity_manager.py — Snapshot & Integrity Manager (P41)

Cattura VFS + Task + Blackboard in snapshot puntuale. Verifica regressioni.
Self-healing automatico a livello file.

Storage: VFS file /.agent/snapshots/snap_<ts>.json (zero migration richiesta).
Bot handlers: handle_snap_cmd / handle_verify_cmd / handle_heal_cmd
              accettano reply_fn callable — zero import ciclici.

Endpoints:
  POST /api/integrity/snap           — crea snapshot
  GET  /api/integrity/snapshots      — lista snapshot disponibili
  GET  /api/integrity/verify         — diff vs ultimo snapshot
  GET  /api/integrity/verify/{id}    — diff vs snapshot specifico
  POST /api/integrity/heal           — ripristina file regrediti
  POST /api/integrity/heal/{id}      — ripristina da snapshot specifico
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
import time
from typing import Any, Callable, Coroutine, Optional

from fastapi import APIRouter, Body
from fastapi.responses import JSONResponse

_logger = logging.getLogger("api.integrity")
router  = APIRouter(prefix="/api/integrity", tags=["integrity"])

# In-memory fallback (usato quando Supabase non disponibile o per rapidita')
_SNAP_CACHE: dict[str, dict] = {}   # snap_id -> snapshot dict
_MAX_CACHE   = 20                    # ultimi N snapshot tenuti in memoria


# ─── Utilities ────────────────────────────────────────────────────────────────

def _sha16(text: str) -> str:
    """SHA-256 troncato a 16 hex — fingerprint leggero per contenuto file."""
    return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()[:16]


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


def _snap_id() -> str:
    return f"snap_{int(time.time())}"


def _age_str(created_at_ms: int) -> str:
    s = (_now_ms() - created_at_ms) // 1000
    if s < 60:   return f"{s}s"
    if s < 3600: return f"{s//60}m {s%60}s"
    return f"{s//3600}h {(s%3600)//60}m"


# ─── VFS (Supabase vfs_files table) ──────────────────────────────────────────

async def _list_vfs_files(conversation_id: Optional[str] = None) -> list[dict]:
    """Legge tutti i file VFS da Supabase. Fallback: lista vuota."""
    from .state import _sb
    if not _sb:
        return []
    try:
        q = _sb.table("vfs_files").select(
            "id,path,content,language,conversation_id,updated_at"
        )
        if conversation_id:
            q = q.eq("conversation_id", conversation_id)
        data = await asyncio.to_thread(lambda: q.order("path").execute())
        return data.data or []
    except Exception as exc:
        _logger.warning("[integrity] list_vfs_files: %s", exc)
        return []


async def _restore_vfs_file(rec: dict, force: bool = False) -> bool:
    """Ripristina un file VFS su Supabase dal record snapshot. True = ok.

    GAP-VFS-FIX: confronto ottimistico su updated_at prima del restore.
    Se il file live ha updated_at > snapshot updated_at, il restore viene
    saltato per non sovrascrivere una versione più recente.
    force=True sovrascrive comunque (emergenza/rollback manuale esplicito).
    """
    from .state import _sb
    if not _sb:
        return False
    try:
        snap_updated_at = rec.get("updated_at", 0)
        file_id         = rec.get("id", "")

        # GAP-VFS-FIX: verifica che il file live non sia più recente dello snapshot
        if not force and file_id and snap_updated_at > 0:
            try:
                current = await asyncio.to_thread(
                    lambda: _sb.table("vfs_files")
                    .select("updated_at")
                    .eq("id", file_id)
                    .maybe_single()
                    .execute()
                )
                if current and current.data:
                    live_updated_at = current.data.get("updated_at", 0)
                    if live_updated_at > snap_updated_at:
                        _logger.info(
                            "[integrity] GAP-VFS: skip restore %s "
                            "(live=%d > snap=%d). Usa force=True per rollback.",
                            rec.get("path"), live_updated_at, snap_updated_at,
                        )
                        return False
            except Exception as _chk_exc:
                # Errore nel check → procedi con restore (fail-open per sicurezza)
                _logger.debug("[integrity] restore pre-check silenced: %s", _chk_exc)

        body: dict = {
            "path":            rec["path"],
            "content":         rec.get("content", ""),
            "language":        rec.get("language", "text"),
            "conversation_id": rec.get("conversation_id", ""),
            "updated_at":      _now_ms(),
        }
        if file_id:
            body["id"] = file_id
        await asyncio.to_thread(lambda: _sb.table("vfs_files").upsert(body).execute())
        # Background: lint + manifest dopo restore
        try:
            content = rec.get("content", "")
            lang    = rec.get("language", "text")
            path    = rec["path"]
            conv    = rec.get("conversation_id", "")
            if file_id and content:
                from .linter import lint_and_store
                from .project_manifest import update_manifest
                asyncio.create_task(lint_and_store(file_id, content, lang, path, conv))
                asyncio.create_task(update_manifest(conv, path, content, lang))
        except Exception:
            pass
        return True
    except Exception as exc:
        _logger.warning("[integrity] restore_vfs_file %s: %s", rec.get("path"), exc)
        return False


# ─── Blackboard (Upstash Redis REST) ─────────────────────────────────────────

async def _list_bb_keys(session_pattern: str = "bb:*") -> list[str]:
    """SCAN cursored su Upstash — ritorna tutte le chiavi blackboard."""
    import httpx
    url   = os.getenv("UPSTASH_REDIS_REST_URL", "")
    token = os.getenv("UPSTASH_REDIS_REST_TOKEN", "")
    if not url or not token:
        return []
    keys: list[str] = []
    cursor = "0"
    try:
        async with httpx.AsyncClient(timeout=4.0) as c:
            for _ in range(10):   # max 10 scan pages
                r = await c.post(
                    url,
                    json=["SCAN", cursor, "MATCH", session_pattern, "COUNT", "200"],
                    headers={"Authorization": f"Bearer {token}"},
                )
                if not r.is_success:
                    break
                result = r.json().get("result", [])
                if isinstance(result, list) and len(result) >= 2:
                    cursor = str(result[0])
                    batch  = result[1] if isinstance(result[1], list) else []
                    keys.extend(batch)
                    if cursor == "0":   # scan completo
                        break
                else:
                    break
    except Exception as exc:
        _logger.warning("[integrity] list_bb_keys: %s", exc)
    return keys


# ─── Task state ───────────────────────────────────────────────────────────────

def _active_task_ids() -> list[str]:
    try:
        from .state import _agent_tasks
        return [tid for tid, t in _agent_tasks.items()
                if t.get("status") in ("running", "pending")]
    except Exception:
        return []


# ─── Snapshot storage (via VFS /.agent/snapshots/) ────────────────────────────

async def _save_snapshot(snap: dict) -> None:
    sid  = snap["snapshot_id"]
    path = f"/.agent/snapshots/{sid}.json"
    body = json.dumps(snap, ensure_ascii=False, default=str)

    # Cache in-memory sempre
    _SNAP_CACHE[sid] = snap
    if len(_SNAP_CACHE) > _MAX_CACHE:
        oldest = min(_SNAP_CACHE, key=lambda k: _SNAP_CACHE[k].get("created_at", 0))
        del _SNAP_CACHE[oldest]

    # Persist su Supabase via vfs_files
    from .state import _sb
    if not _sb:
        return
    try:
        await asyncio.to_thread(lambda: _sb.table("vfs_files").upsert({
            "path":            path,
            "content":         body,
            "language":        "json",
            "conversation_id": "__integrity__",
            "updated_at":      _now_ms(),
            "created_at":      snap["created_at"],
        }).execute())
    except Exception as exc:
        _logger.warning("[integrity] save_snapshot %s: %s", sid, exc)


async def _load_all_snapshots() -> list[dict]:
    """Lista snapshot da Supabase + cache, ordinati per data desc."""
    merged: dict[str, dict] = dict(_SNAP_CACHE)
    from .state import _sb
    if _sb:
        try:
            data = await asyncio.to_thread(lambda: _sb.table("vfs_files")
                .select("content,updated_at")
                .eq("conversation_id", "__integrity__")
                .like("path", "/.agent/snapshots/snap_%.json")
                .order("updated_at", desc=True)
                .limit(30)
                .execute()
            )
            for row in (data.data or []):
                try:
                    s = json.loads(row["content"])
                    if sid := s.get("snapshot_id"):
                        merged[sid] = s
                except Exception:
                    pass
        except Exception as exc:
            _logger.warning("[integrity] load_snapshots: %s", exc)
    return sorted(merged.values(), key=lambda s: s.get("created_at", 0), reverse=True)


async def _latest_snapshot() -> Optional[dict]:
    snaps = await _load_all_snapshots()
    return snaps[0] if snaps else None


async def _snapshot_by_id(snap_id: str) -> Optional[dict]:
    if snap_id in _SNAP_CACHE:
        return _SNAP_CACHE[snap_id]
    snaps = await _load_all_snapshots()
    return next((s for s in snaps if s.get("snapshot_id") == snap_id), None)


# ─── Core: take_snapshot ──────────────────────────────────────────────────────

async def take_snapshot(conversation_id: Optional[str] = None) -> dict:
    """Cattura snapshot completo: VFS + task attivi + blackboard keys."""
    sid = _snap_id()
    ts  = _now_ms()

    # VFS
    vfs_files = await _list_vfs_files(conversation_id)
    vfs: dict[str, dict] = {}
    for f in vfs_files:
        p = f.get("path", "")
        if not p or p.startswith("/.agent/snapshots/"):
            continue
        c = f.get("content") or ""
        vfs[p] = {
            "id":              f.get("id", ""),
            "content_sha":     _sha16(c),
            "content":         c,
            "language":        f.get("language", "text"),
            "conversation_id": f.get("conversation_id", ""),
            "updated_at":      f.get("updated_at", 0),
        }

    # Task
    active = _active_task_ids()

    # Blackboard
    bb_keys = await _list_bb_keys()

    snap = {
        "snapshot_id":     sid,
        "created_at":      ts,
        "conversation_id": conversation_id,
        "vfs":             vfs,
        "tasks": {
            "active_ids": active,
            "count":      len(active),
        },
        "blackboard": {
            "keys":      bb_keys,
            "key_count": len(bb_keys),
        },
        "summary": {
            "vfs_files":    len(vfs),
            "active_tasks": len(active),
            "bb_keys":      len(bb_keys),
        },
    }
    await _save_snapshot(snap)
    _logger.info("[integrity] snapshot %s: %d VFS, %d tasks, %d BB",
                 sid, len(vfs), len(active), len(bb_keys))
    return snap


# ─── Core: verify_snapshot ────────────────────────────────────────────────────

async def verify_snapshot(snap_id: Optional[str] = None) -> dict:
    """Confronta stato corrente vs snapshot. Ritorna diff report."""
    snap = (await _snapshot_by_id(snap_id) if snap_id
            else await _latest_snapshot())
    if not snap:
        return {"ok": False, "error": "Nessuno snapshot trovato. Esegui /snap prima di verificare."}

    snap_vfs  = snap.get("vfs", {})
    conv_id   = snap.get("conversation_id")

    current_files = await _list_vfs_files(conv_id)
    current: dict[str, str] = {}
    for f in current_files:
        p = f.get("path", "")
        if not p or p.startswith("/.agent/snapshots/"):
            continue
        current[p] = _sha16(f.get("content") or "")

    snap_paths    = set(snap_vfs)
    current_paths = set(current)

    missing  = sorted(snap_paths - current_paths)
    added    = sorted(current_paths - snap_paths)
    modified = sorted(p for p in snap_paths & current_paths
                      if current[p] != snap_vfs[p]["content_sha"])
    ok_count = len(snap_paths & current_paths) - len(modified)

    regression = bool(missing or modified)
    return {
        "ok":                True,
        "snapshot_id":       snap["snapshot_id"],
        "snapshot_age_s":    (_now_ms() - snap["created_at"]) // 1000,
        "vfs": {
            "missing":        missing,
            "modified":       modified,
            "added":          added,
            "ok":             ok_count,
            "total_snapshot": len(snap_paths),
            "total_current":  len(current_paths),
        },
        "tasks": {
            "snapshot_count": snap.get("tasks", {}).get("count", 0),
            "current_count":  len(_active_task_ids()),
        },
        "regression_detected": regression,
        "heal_candidates":     missing + modified,
    }


# ─── Core: heal_from_snapshot ─────────────────────────────────────────────────

async def heal_from_snapshot(
    snap_id: Optional[str] = None,
    paths:   Optional[list[str]] = None,
) -> dict:
    """Ripristina file regrediti/mancanti dallo snapshot."""
    snap = (await _snapshot_by_id(snap_id) if snap_id
            else await _latest_snapshot())
    if not snap:
        return {"ok": False, "error": "Nessuno snapshot trovato."}

    if paths is None:
        report = await verify_snapshot(snap.get("snapshot_id"))
        paths  = report.get("heal_candidates", [])

    if not paths:
        return {"ok": True, "healed": [], "failed": [],
                "message": "Nessuna regressione — sistema integro."}

    snap_vfs = snap.get("vfs", {})
    healed: list[str] = []
    failed: list[dict] = []

    for p in paths:
        rec = snap_vfs.get(p)
        if not rec:
            failed.append({"path": p, "reason": "path non nel snapshot"})
            continue
        ok = await _restore_vfs_file(rec)
        (healed if ok else failed.append({"path": p, "reason": "restore fallito"}) or []).append(p) if ok else None

    _logger.info("[integrity] heal: %d ok, %d failed", len(healed), len(failed))
    return {
        "ok":      True,
        "healed":  healed,
        "failed":  failed,
        "message": f"Ripristinati {len(healed)}/{len(paths)} file.",
    }


# ─── FastAPI endpoints ────────────────────────────────────────────────────────

@router.post("/snap")
async def api_snap(body: dict = Body(default={})) -> JSONResponse:
    """POST /api/integrity/snap — cattura snapshot dell'ambiente."""
    try:
        conv_id = body.get("conversation_id") if isinstance(body, dict) else None
        snap = await take_snapshot(conv_id)
        return JSONResponse({"ok": True, "snapshot_id": snap["snapshot_id"],
                             "summary": snap["summary"]})
    except Exception as exc:
        _logger.error("[integrity] snap: %s", exc)
        return JSONResponse({"ok": False, "error": str(exc)[:200]}, status_code=500)


@router.get("/snapshots")
async def api_list_snapshots() -> JSONResponse:
    """GET /api/integrity/snapshots — lista snapshot disponibili."""
    try:
        snaps = await _load_all_snapshots()
        return JSONResponse({"ok": True, "snapshots": [
            {"snapshot_id": s["snapshot_id"], "created_at": s["created_at"],
             "summary": s.get("summary", {})}
            for s in snaps
        ]})
    except Exception as exc:
        return JSONResponse({"ok": False, "error": str(exc)[:200]}, status_code=500)


@router.get("/verify")
async def api_verify() -> JSONResponse:
    """GET /api/integrity/verify — diff vs ultimo snapshot."""
    return JSONResponse(await verify_snapshot())


@router.get("/verify/{snap_id}")
async def api_verify_id(snap_id: str) -> JSONResponse:
    """GET /api/integrity/verify/{id} — diff vs snapshot specifico."""
    return JSONResponse(await verify_snapshot(snap_id))


@router.post("/heal")
async def api_heal(body: dict = Body(default={})) -> JSONResponse:
    """POST /api/integrity/heal — ripristina file regrediti."""
    try:
        snap_id = body.get("snap_id") if isinstance(body, dict) else None
        paths   = body.get("paths") if isinstance(body, dict) else None
        return JSONResponse(await heal_from_snapshot(snap_id, paths))
    except Exception as exc:
        _logger.error("[integrity] heal: %s", exc)
        return JSONResponse({"ok": False, "error": str(exc)[:200]}, status_code=500)


@router.post("/heal/{snap_id}")
async def api_heal_id(snap_id: str) -> JSONResponse:
    """POST /api/integrity/heal/{id} — ripristina da snapshot specifico."""
    return JSONResponse(await heal_from_snapshot(snap_id))


# ─── Telegram bot command handlers ────────────────────────────────────────────
# Chiamati da telegram_webhook.py; accettano reply_fn per evitare import ciclici.

ReplyFn = Callable[[Any, str], Coroutine[Any, Any, None]]


async def handle_snap_cmd(chat_id: Any, reply_fn: ReplyFn,
                           conversation_id: Optional[str] = None) -> None:
    try:
        await reply_fn(chat_id, "⏳ <b>Acquisisco snapshot...</b>")
        snap = await take_snapshot(conversation_id)
        s = snap["summary"]
        await reply_fn(chat_id,
            f"📸 <b>Snapshot acquisito</b>\n"
            f"<code>{snap['snapshot_id']}</code>\n\n"
            f"• VFS: <b>{s['vfs_files']}</b> file\n"
            f"• Task attivi: <b>{s['active_tasks']}</b>\n"
            f"• Blackboard keys: <b>{s['bb_keys']}</b>\n\n"
            f"Usa /verify per controllare regressioni in qualsiasi momento."
        )
    except Exception as exc:
        await reply_fn(chat_id, f"❌ Snapshot fallito: <code>{str(exc)[:150]}</code>")


async def handle_verify_cmd(chat_id: Any, reply_fn: ReplyFn,
                             snap_id: Optional[str] = None) -> None:
    try:
        await reply_fn(chat_id, "🔍 <b>Verifico integrità...</b>")
        r = await verify_snapshot(snap_id)
        if not r.get("ok"):
            await reply_fn(chat_id, f"⚠️ {r.get('error', 'Errore sconosciuto')}")
            return
        vfs = r.get("vfs", {})
        age = _age_str(r.get("snapshot_age_s", 0) * 1000 + _now_ms() - r.get("snapshot_age_s", 0) * 1000)
        # compute age correctly
        age_s = r.get("snapshot_age_s", 0)
        if age_s < 60:   age = f"{age_s}s"
        elif age_s < 3600: age = f"{age_s//60}m {age_s%60}s"
        else: age = f"{age_s//3600}h {(age_s%3600)//60}m"

        if not r["regression_detected"]:
            await reply_fn(chat_id,
                f"✅ <b>Sistema integro</b>\n"
                f"Snapshot: <code>{r['snapshot_id']}</code> ({age} fa)\n\n"
                f"File verificati: {vfs.get('ok',0)}/{vfs.get('total_snapshot',0)} — "
                f"nessuna regressione."
            )
            return
        lines = [f"⚠️ <b>Regressioni rilevate!</b>",
                 f"Snapshot: <code>{r['snapshot_id']}</code> ({age} fa)\n"]
        if vfs.get("missing"):
            lines.append(f"🗑 <b>Eliminati ({len(vfs['missing'])}):</b>")
            for p in vfs["missing"][:5]: lines.append(f"  • <code>{p}</code>")
            if len(vfs["missing"]) > 5: lines.append(f"  … +{len(vfs['missing'])-5}")
        if vfs.get("modified"):
            lines.append(f"\n✏️ <b>Modificati ({len(vfs['modified'])}):</b>")
            for p in vfs["modified"][:5]: lines.append(f"  • <code>{p}</code>")
            if len(vfs["modified"]) > 5: lines.append(f"  … +{len(vfs['modified'])-5}")
        if vfs.get("added"):
            lines.append(f"\n➕ <b>Nuovi ({len(vfs['added'])}):</b>")
            for p in vfs["added"][:3]: lines.append(f"  • <code>{p}</code>")
        lines.append(f"\nUsa /heal per ripristinare i file regrediti.")
        await reply_fn(chat_id, "\n".join(lines))
    except Exception as exc:
        await reply_fn(chat_id, f"❌ Verify fallito: <code>{str(exc)[:150]}</code>")


async def handle_heal_cmd(chat_id: Any, reply_fn: ReplyFn,
                           snap_id: Optional[str] = None) -> None:
    try:
        await reply_fn(chat_id, "🔧 <b>Self-healing in corso...</b>")
        r = await heal_from_snapshot(snap_id)
        if not r.get("ok"):
            await reply_fn(chat_id, f"❌ Heal fallito: {r.get('error')}")
            return
        healed = r.get("healed", [])
        failed = r.get("failed", [])
        if not healed and not failed:
            await reply_fn(chat_id,
                "✅ <b>Nessuna regressione</b> — il sistema è già integro.")
            return
        lines = ["🔧 <b>Self-healing completato</b>\n"]
        if healed:
            lines.append(f"✅ <b>Ripristinati ({len(healed)}):</b>")
            for p in healed[:8]: lines.append(f"  • <code>{p}</code>")
            if len(healed) > 8: lines.append(f"  … +{len(healed)-8}")
        if failed:
            lines.append(f"\n❌ <b>Falliti ({len(failed)}):</b>")
            for f in failed[:3]:
                lines.append(f"  • <code>{f['path']}</code>: {f['reason']}")
        await reply_fn(chat_id, "\n".join(lines))
    except Exception as exc:
        await reply_fn(chat_id, f"❌ Heal fallito: <code>{str(exc)[:150]}</code>")