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, Depends, Body
from .auth_guard import require_role, AuthRole
from fastapi.responses import JSONResponse

_logger = logging.getLogger("api.integrity")
router  = APIRouter(prefix="/api/integrity", tags=["integrity"], dependencies=[Depends(require_role(AuthRole.MACHINE))])  # GAP-1-fix: router-level auth

# 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
                # BUGFIX: create_task senza add_done_callback perde eccezioni silenziosamente.
                def _log_bg(t):
                    if not t.cancelled() and t.exception():
                        _logger.warning("[integrity] bg task raised: %s", t.exception())
                asyncio.create_task(lint_and_store(file_id, content, lang, path, conv)).add_done_callback(_log_bg)
                asyncio.create_task(update_manifest(conv, path, content, lang)).add_done_callback(_log_bg)
        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>")