api/scheduler.py

"""
backend/api/scheduler.py — Scheduler server-side persistente.

Sostituisce il loop client-side browser con un ciclo asyncio sul backend.
I task sopravvivono alla chiusura del browser iPhone.

Architettura:
  - asyncio background task (tick ogni 60s)
  - JSON file per persistenza (sopravvive al processo, si resetta al restart HF Space)
  - Frontend re-sincronizza Dexie → backend al mount (POST /api/scheduler/sync)
  - SSE push in real-time (<100ms) invece di polling 30s
  - Timeout task: 120s server-side vs 30s client-side
  - Un solo task per tick (same invariant del client-side)

Route:
  GET    /api/scheduler/tasks              lista tutti i task
  POST   /api/scheduler/tasks             crea task dal frontend
  PATCH  /api/scheduler/tasks/{id}        pausa / riprendi / aggiorna
  DELETE /api/scheduler/tasks/{id}        cancella task
  POST   /api/scheduler/sync              bulk upsert da Dexie (idempotente)
  POST   /api/scheduler/trigger/{id}      esecuzione immediata (debug/manuale)
  POST   /api/scheduler/tick              pacemaker esterno (CF Worker, GHA)
  GET    /api/scheduler/status            stato del loop asyncio
  GET    /api/scheduler/webhook/sse       SSE stream real-time per frontend
"""

import asyncio
import datetime
import json
from .state import safe_json_dumps
import os
import time
import uuid
from pathlib import Path
from typing import Any, AsyncGenerator, Optional

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import logging
_logger = logging.getLogger("agente_ai")  # S-BUGFIX

logger = logging.getLogger("agente_ai.scheduler")

def _log_task_exc(task):  # GAP-2.6: log silently-dropped exceptions in fire-and-forget tasks
    if not task.cancelled():
        exc = task.exception()
        if exc:
            logger.warning("[scheduler] background task raised %s: %s", type(exc).__name__, exc)
router = APIRouter(prefix="/api/scheduler", tags=["scheduler"])

# ─── Telegram notifications (fire-and-forget) ─────────────────────────────────
try:
    from .telegram_notify import (
        notify_task_done as _tg_done, notify_task_error as _tg_error,
        notify_task_start as _tg_start, notify_task_heartbeat as _tg_heartbeat,
    )
except Exception:
    async def _tg_done(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_error(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_start(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_heartbeat(*_a, **_kw): pass  # type: ignore[misc]

# ─── Persistenza JSON ─────────────────────────────────────────────────────────
# Usa /tmp su HF Space (ephemeral ma dura ore).
# Il frontend re-sincronizza Dexie → backend al mount: zero task persi.

_TASKS_FILE = Path(os.getenv("SCHEDULER_TASKS_FILE", "/tmp/agente_scheduler.json"))
_TASKS_BAK  = Path(str(os.getenv("SCHEDULER_TASKS_FILE", "/tmp/agente_scheduler.json")) + ".bak")
_tasks: dict[str, dict] = {}        # id → task (in-memory, fonte di verità)
_lock  = asyncio.Lock()             # serializza tutti i write (no race conditions)

# DEAD-LETTER-WATCHDOG: task rimasti "running" oltre questo timeout vengono
# resettati a "pending" dal tick — previene blocco permanente del loop.
_STUCK_TIMEOUT_S = 300              # 5 min — > 120s timeout _run_goal + margine

# ─── Supabase persistence (MX11-SCHED) ────────────────────────────────────────
# Backup permanente dei task scheduler: ad ogni mutazione salva su Supabase
# in modo fire-and-forget. Al boot, se /tmp è vuoto, carica da Supabase.
# Rende i task immortali: sopravvivono a qualsiasi Railway restart.

_SB_URL = os.getenv("SUPABASE_URL", "").rstrip("/")
_SB_KEY = os.getenv("SUPABASE_KEY", "") or os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")  # GAP1-FIX


async def _sb_upsert_tasks(tasks_snapshot: list) -> None:
    """Upsert su Supabase scheduler_tasks (fire-and-forget, non bloccante)."""
    if not _SB_URL or not _SB_KEY or not tasks_snapshot:
        return
    try:
        import httpx
        rows = [
            {
                "id":         t["id"],
                "data":       t,
                "updated_at": datetime.datetime.utcnow().isoformat() + "Z",
            }
            for t in tasks_snapshot
            if isinstance(t, dict) and "id" in t
        ]
        async with httpx.AsyncClient(timeout=8.0) as client:
            await client.post(
                f"{_SB_URL}/rest/v1/scheduler_tasks",
                headers={
                    "apikey":        _SB_KEY,
                    "Authorization": f"Bearer {_SB_KEY}",
                    "Content-Type":  "application/json",
                    "Prefer":        "resolution=merge-duplicates",
                },
                json=rows,
            )
    except Exception as _exc:
        logger.debug("Scheduler: sb_upsert silenced: %s", _exc)


async def _sb_delete_task_row(task_id: str) -> None:
    """Rimuove un task da Supabase (fire-and-forget)."""
    if not _SB_URL or not _SB_KEY:
        return
    try:
        import httpx
        async with httpx.AsyncClient(timeout=5.0) as client:
            await client.delete(
                f"{_SB_URL}/rest/v1/scheduler_tasks?id=eq.{task_id}",
                headers={
                    "apikey":        _SB_KEY,
                    "Authorization": f"Bearer {_SB_KEY}",
                },
            )
    except Exception as _exc:
        logger.debug("Scheduler: sb_delete silenced: %s", _exc)


async def _sb_load_tasks_async() -> dict:
    """Carica task da Supabase — fallback al boot se /tmp è vuoto."""
    if not _SB_URL or not _SB_KEY:
        return {}
    try:
        import httpx
        async with httpx.AsyncClient(timeout=10.0) as client:
            r = await client.get(
                f"{_SB_URL}/rest/v1/scheduler_tasks?select=id,data&order=updated_at.asc",
                headers={
                    "apikey":        _SB_KEY,
                    "Authorization": f"Bearer {_SB_KEY}",
                },
            )
            rows = r.json()
            if not isinstance(rows, list):
                return {}
            result = {}
            for row in rows:
                data = row.get("data")
                if isinstance(data, dict) and "id" in data:
                    result[data["id"]] = data
            logger.info("Scheduler: caricati %d task da Supabase (fallback boot)", len(result))
            return result
    except Exception as _exc:
        logger.warning("Scheduler: sb_load_tasks failed: %s", _exc)
        return {}


def _trigger_sb_sync() -> None:
    """
    Schedula Supabase upsert in background dopo ogni mutazione.
    Chiamato da _broadcast_sse() (già eseguita con _lock acquisito).
    Fire-and-forget: un fallback non blocca l'operazione principale.
    """
    try:
        snapshot = list(_tasks.values())
        asyncio.create_task(_sb_upsert_tasks(snapshot)).add_done_callback(_log_task_exc)
    except RuntimeError:
        pass   # no event loop attivo (chiamata da contesto sync pre-startup)


def _load_tasks() -> None:
    """Gap-7-FIX: carica da file principale, fallback a backup se corrotto."""
    global _tasks
    for _path in (_TASKS_FILE, _TASKS_BAK):
        try:
            if _path.exists():
                raw    = json.loads(_path.read_text(encoding="utf-8"))
                _tasks = {t["id"]: t for t in raw if isinstance(t, dict) and "id" in t}
                logger.info("Scheduler: caricati %d task da %s", len(_tasks), _path)
                return
        except Exception as exc:
            logger.warning("Scheduler: load da %s fallito (%s) — provo backup", _path, exc)
    _tasks = {}
    logger.warning("Scheduler: nessun task salvato trovato — partenza vuota (proverò Supabase)")


def _save_tasks_sync() -> None:
    """Gap-7-FIX: write atomico (tmp→rename) + backup — resistente a crash mid-write."""
    try:
        _TASKS_FILE.parent.mkdir(parents=True, exist_ok=True)
        data = safe_json_dumps(list(_tasks.values()), indent=2)
        # Write atomico: scrive su .tmp poi rinomina — evita file corrotto se process killed
        _tmp = _TASKS_FILE.with_suffix(".tmp")
        _tmp.write_text(data, encoding="utf-8")
        _tmp.replace(_TASKS_FILE)
        # Backup separato — fallback se il file principale si corrompe al prossimo avvio
        try:
            _TASKS_BAK.write_text(data, encoding="utf-8")
        except Exception:
            pass  # backup non critico — non bloccare il path principale
    except Exception as exc:
        logger.warning("Scheduler: save fallito: %s", exc)


# ─── SSE fan-out ──────────────────────────────────────────────────────────────
# Una Queue per ogni client SSE connesso.
# _broadcast_sse() serializza i task correnti e li invia a tutti.

_sse_clients: list[asyncio.Queue] = []


def _broadcast_sse() -> None:
    """
    Invia la lista task aggiornata a tutti i client SSE connessi.
    Fire-and-forget: chiamato dopo ogni mutazione (create/patch/delete/execute).
    Deve essere chiamato con _lock già acquisito (legge _tasks direttamente).
    MX11-SCHED: schedula anche Supabase sync (backup permanente).
    """
    if not _sse_clients:
        _trigger_sb_sync()   # sync Supabase anche senza client SSE
        return
    payload = safe_json_dumps(list(_tasks.values()))
    event   = f"event: tasks_updated\ndata: {payload}\n\n"
    for q in _sse_clients:
        try:
            q.put_nowait(event)
        except asyncio.QueueFull:
            pass   # client lento — skip questo evento, riceverà il prossimo
    _trigger_sb_sync()       # backup Supabase ad ogni mutazione


async def _sse_generator(queue: asyncio.Queue, request: Request) -> AsyncGenerator[str, None]:
    """
    Genera eventi SSE per un client connesso.
    Si chiude quando il client disconnette (request.is_disconnected()).
    Heartbeat ogni 25s per mantenere la connessione viva su iOS/proxy.
    """
    # Invia subito la lista task corrente (snapshot iniziale)
    async with _lock:
        snapshot = safe_json_dumps(list(_tasks.values()))
    yield f"event: tasks_updated\ndata: {snapshot}\n\n"

    while True:
        if await request.is_disconnected():
            break
        try:
            # Attendi evento o heartbeat dopo 25s
            event = await asyncio.wait_for(queue.get(), timeout=25.0)
            yield event
        except asyncio.TimeoutError:
            # Heartbeat — mantiene viva la connessione su iOS Safari / Cloudflare
            yield ": heartbeat\n\n"


# ─── Helpers trigger ──────────────────────────────────────────────────────────

def _is_due(task: dict, now_ms: int) -> bool:
    t = task.get("trigger", {})
    tt = t.get("type")
    if tt == "once":       return now_ms >= t.get("runAt",   0)
    if tt == "interval":   return now_ms >= t.get("nextRun", 0)
    if tt == "daily":      return now_ms >= t.get("nextRun", 0)
    if tt == "on_open":    return True          # boot-time task
    if tt == "issue_poll": return now_ms >= t.get("nextRun", 0)
    return False


def _advance_trigger(trigger: dict, now_ms: int) -> dict:
    t  = dict(trigger)
    tt = t.get("type")
    if tt in ("interval", "issue_poll"):
        t["nextRun"] = now_ms + t.get("intervalMs", 3_600_000)
    elif tt == "daily":
        hour   = t.get("hour",   9)
        minute = t.get("minute", 0)
        nxt    = datetime.datetime.now().replace(
            hour=hour, minute=minute, second=0, microsecond=0
        )
        nxt_ms = int(nxt.timestamp() * 1000)
        if nxt_ms <= now_ms:
            nxt    = nxt + datetime.timedelta(days=1)
            nxt_ms = int(nxt.timestamp() * 1000)
        t["nextRun"] = nxt_ms
    # once / on_open: nessun avanzamento
    return t


# ─── Esecutore task ───────────────────────────────────────────────────────────

async def _run_goal(goal: str, conversation_id: Optional[str] = None) -> str:
    """
    Esegue il goal tramite UnifiedAgentLoop (stesso path di api/agent.py).
    Timeout: 120s — 4× il budget client-side (30s).
    """
    try:
        from agents.unified_loop import UnifiedAgentLoop
        from api.state import (
            _get_ai_client, _get_mem_manager_async,
            _get_executor, _get_planner,
        )

        client   = _get_ai_client()
        memory   = await _get_mem_manager_async()
        executor = _get_executor()
        planner  = _get_planner()

        try:
            from agents.critic           import Critic
            from agents.response_verifier import ResponseVerifier
            critic   = Critic(llm_client=client)
            verifier = ResponseVerifier()
        except Exception:
            critic   = None
            verifier = None

        loop = UnifiedAgentLoop(
            llm_client=client, critic=critic, verifier=verifier,
            memory=memory, executor=executor, planner=planner,
        )

        result = await asyncio.wait_for(
            loop.run(goal=goal, context="", max_steps=8),
            timeout=120.0,
        )
        output = result.get("output", "") if isinstance(result, dict) else str(result)
        return str(output)[:1000]

    except asyncio.TimeoutError:
        return "❌ Timeout: task terminato dopo 120s"
    except Exception as exc:
        logger.error("Scheduler._run_goal error: %s", exc, exc_info=True)
        return f"❌ Errore: {str(exc)[:400]}"



async def _sb_write_scheduler_result(task_id: str, goal: str, status: str, result: str, now_ms: int) -> None:
    """Scrive risultato scheduler su Supabase (fire-and-forget). Sopravvive al riavvio HF Space."""
    try:
        from .persistence import sb_upsert_task, sb_append_event
        import json as _json
        await sb_upsert_task(task_id, goal[:500], status, 0, [], now_ms)
        _evt = safe_json_dumps({"type": "task_done", "result": result[:800], "source": "scheduler", "taskId": task_id})
        await sb_append_event(task_id, 0, f"data: {_evt}\n\n")
    except Exception as _exc:
        logger.warning("Scheduler: sb_write_result failed: %s", _exc)


async def _execute_task(task_id: str) -> None:
    """Esegue un task, aggiorna status e salva."""
    now_ms = int(time.time() * 1000)

    # Marca running + broadcast SSE
    async with _lock:
        task = _tasks.get(task_id)
        if not task:
            return
        task["status"]    = "running"
        task["lastRunAt"] = now_ms
        _task_notify  = task.get("notify", True)
        _task_label   = task.get("label", task.get("goal", ""))[:200]
        _task_goal    = task.get("goal", _task_label)[:200]
        _save_tasks_sync()
        _broadcast_sse()
    if _task_notify:
        asyncio.create_task(_tg_start(task_id, _task_goal)).add_done_callback(_log_task_exc)

    try:
        result = await _run_goal(task["goal"], task.get("conversationId"))

        async with _lock:
            task = _tasks.get(task_id)
            if not task:
                return
            ttype     = task["trigger"].get("type")
            one_shot  = ttype in ("once", "on_open")
            task["status"]     = "done" if one_shot else "pending"
            task["trigger"]    = _advance_trigger(task["trigger"], now_ms)
            task["lastRunAt"]  = now_ms
            task["lastResult"] = result
            task["errorCount"] = 0
            _save_tasks_sync()
            _broadcast_sse()
            _sb_goal_ok = task.get("goal", task.get("label", ""))[:500]
            _sb_stat_ok = "done" if one_shot else "pending"

        logger.info("Scheduler: ✓ task '%s' (%s)", task.get("label"), task_id)
        asyncio.create_task(_sb_write_scheduler_result(task_id, _sb_goal_ok, _sb_stat_ok, result, now_ms)).add_done_callback(_log_task_exc)
        if _task_notify:
            asyncio.create_task(_tg_done(task_id, _task_goal, result[:500])).add_done_callback(_log_task_exc)

    except Exception as exc:
        async with _lock:
            task = _tasks.get(task_id)
            if not task:
                return
            task["errorCount"] = task.get("errorCount", 0) + 1
            failed             = task["errorCount"] >= task.get("maxErrors", 3)
            task["status"]     = "failed" if failed else "pending"
            if not failed:
                task["trigger"] = _advance_trigger(
                    task["trigger"], now_ms + 5 * 60_000
                )
            task["lastRunAt"]  = now_ms
            task["lastResult"] = f"❌ {str(exc)[:300]}"
            _save_tasks_sync()
            _broadcast_sse()

        logger.error("Scheduler: ✗ task %s: %s", task_id, exc)
        # GAP-A1: log incident in registry (fire-and-forget, non-blocking)
        try:
            from .incident_registry import log_incident as _log_inc
            asyncio.create_task(_log_inc(
                task_id=task_id, goal=_task_goal, error=str(exc), source="scheduler"
            )).add_done_callback(_log_task_exc)
        except Exception as _exc:
            _logger.debug("[scheduler] silenced %s", type(_exc).__name__)  # noqa: BLE001
        _sb_goal_err = task.get("goal", task.get("label", ""))[:500] if task else ""
        _sb_stat_err = "failed" if failed else "pending"
        asyncio.create_task(_sb_write_scheduler_result(task_id, _sb_goal_err, _sb_stat_err, f"❌ {str(exc)[:300]}", now_ms)).add_done_callback(_log_task_exc)
        if _task_notify:
            asyncio.create_task(_tg_error(task_id, _task_goal, str(exc)[:300])).add_done_callback(_log_task_exc)


# ─── Background loop ──────────────────────────────────────────────────────────

_loop_task: Optional[asyncio.Task] = None
_current_running: Optional[str]    = None   # task_id in esecuzione


async def _tick() -> None:
    """Controlla task scaduti; lancia il primo trovato (uno per tick)."""
    global _current_running

    # DEAD-LETTER-WATCHDOG: se il task corrente è "running" da troppo tempo,
    # è probabilmente bloccato (crash silenzioso, HF Space OOM, ecc.).
    # Reset a "pending" + notifica Telegram + incrementa errorCount.
    if _current_running:
        async with _lock:
            t = _tasks.get(_current_running, {})
        if t.get("status") == "running":
            last_run = t.get("lastRunAt") or 0
            elapsed_s = (time.time() * 1000 - last_run) / 1000
            if elapsed_s > _STUCK_TIMEOUT_S:
                # Task bloccato — reset e notifica
                elapsed_min = elapsed_s / 60
                goal = t.get("goal", t.get("label", ""))[:200]
                logger.warning(
                    "Scheduler DEAD-LETTER: task '%s' bloccato da %.1fmin — reset a pending",
                    _current_running, elapsed_min,
                )
                asyncio.create_task(_tg_heartbeat(
                    _current_running, goal, elapsed_min,
                    current_step="timeout watchdog",
                    force=True,
                )).add_done_callback(_log_task_exc)
                async with _lock:
                    t["status"]     = "pending"
                    t["lastResult"] = f"⚠️ Reset automatico: bloccato da {elapsed_min:.0f}min."
                    t["errorCount"] = t.get("errorCount", 0) + 1
                    _save_tasks_sync()
                    _broadcast_sse()
                _current_running = None
            else:
                return                   # ancora in esecuzione, entro timeout
        else:
            _current_running = None

    now_ms = int(time.time() * 1000)
    async with _lock:
        pending = [t for t in _tasks.values() if t.get("status") == "pending"]

    due = [t for t in pending if _is_due(t, now_ms)]
    if not due:
        return

    task             = due[0]
    _current_running = task["id"]
    logger.info("Scheduler: lancio task '%s'", task.get("label"))
    asyncio.create_task(_execute_task(task["id"])).add_done_callback(_log_task_exc)


async def _scheduler_loop() -> None:
    """Loop principale: avvio ritardato di 5s poi tick ogni 60s."""
    await asyncio.sleep(5)          # non bloccare FastAPI startup
    logger.info("Scheduler: loop asyncio avviato ✓")
    while True:
        try:
            await _tick()
        except Exception as exc:
            logger.error("Scheduler: tick error: %s", exc)
        await asyncio.sleep(60)


def start_scheduler() -> None:
    """
    Avvia il loop scheduler. Chiamato in _on_startup() di main.py.
    Idempotente — sicuro su multipli import.

    MX11-SCHED: usa _boot() coroutine che:
      1. Carica task da Supabase se /tmp era vuoto (cross-restart recovery)
      2. Avvia il normal _scheduler_loop()
    """
    global _loop_task
    _load_tasks()

    # Crash recovery: task rimasti 'running' dopo un restart
    for t in _tasks.values():
        if t.get("status") == "running":
            t["status"]     = "pending"
            t["lastResult"] = "⚠️ Ripristinato dopo riavvio backend."
    _save_tasks_sync()

    if _loop_task is None or _loop_task.done():
        # MX11-SCHED: boot coroutine — carica Supabase se /tmp vuoto, poi avvia loop
        _was_empty = len(_tasks) == 0

        async def _boot() -> None:
            if _was_empty:
                sb_tasks = await _sb_load_tasks_async()
                if sb_tasks:
                    async with _lock:
                        _tasks.update(sb_tasks)
                        _save_tasks_sync()
                    logger.info(
                        "Scheduler: ripristinati %d task da Supabase (cross-restart)",
                        len(sb_tasks),
                    )
            await _scheduler_loop()

        _loop_task = asyncio.create_task(_boot())
        _loop_task.add_done_callback(_log_task_exc)  # GAP-2.6: log silently-dropped exceptions
        logger.info("Scheduler: asyncio task creato ✓")


# ─── Pydantic models ──────────────────────────────────────────────────────────

class TaskCreate(BaseModel):
    id:             Optional[str]  = None
    label:          str
    goal:           str
    trigger:        dict
    notify:         bool           = True
    maxErrors:      int            = 3
    conversationId: Optional[str]  = None


class TaskPatch(BaseModel):
    status:  Optional[str]  = None   # "pending" | "paused"
    trigger: Optional[dict] = None
    label:   Optional[str]  = None


# ─── REST Endpoints ───────────────────────────────────────────────────────────

@router.get("/tasks")
async def list_tasks() -> list[dict]:
    """Polling dal frontend (fallback se SSE non disponibile) — fonte di verità server-side."""
    async with _lock:
        return list(_tasks.values())


@router.post("/tasks", status_code=201)
async def create_task(body: TaskCreate) -> dict:
    """Crea task sul backend. Il frontend chiama questo DOPO il salvataggio Dexie."""
    tid = body.id or f"sched_{int(time.time()*1000):x}_{uuid.uuid4().hex[:4]}"
    task: dict[str, Any] = {
        "id":             tid,
        "label":          body.label,
        "goal":           body.goal,
        "trigger":        body.trigger,
        "status":         "pending",
        "createdAt":      int(time.time() * 1000),
        "lastRunAt":      None,
        "lastResult":     None,
        "errorCount":     0,
        "maxErrors":      body.maxErrors,
        "notify":         body.notify,
        "conversationId": body.conversationId,
    }
    async with _lock:
        _tasks[tid] = task
        _save_tasks_sync()
        _broadcast_sse()

    logger.info("Scheduler: task creato '%s' (%s)", body.label, tid)
    return task


@router.patch("/tasks/{task_id}")
async def patch_task(task_id: str, body: TaskPatch) -> dict:
    """Pausa, riprendi, o aggiorna label/trigger di un task."""
    async with _lock:
        task = _tasks.get(task_id)
        if not task:
            raise HTTPException(404, "Task non trovato")
        if body.status  is not None: task["status"]  = body.status
        if body.trigger is not None: task["trigger"] = body.trigger
        if body.label   is not None: task["label"]   = body.label
        _save_tasks_sync()
        _broadcast_sse()
    return task


@router.delete("/tasks/{task_id}", status_code=204)
async def delete_task(task_id: str) -> None:
    """Cancella task dal backend + Supabase (MX11-SCHED)."""
    async with _lock:
        if task_id not in _tasks:
            raise HTTPException(404, "Task non trovato")
        del _tasks[task_id]
        _save_tasks_sync()
        _broadcast_sse()
    # Rimuovi anche da Supabase (fire-and-forget)
    asyncio.create_task(_sb_delete_task_row(task_id)).add_done_callback(_log_task_exc)


@router.post("/sync")
async def sync_tasks(body: list[dict]) -> dict:
    """
    Bulk upsert da Dexie → backend.
    Idempotente: inserisce solo i task assenti. Non sovrascrive quelli esistenti.
    Chiamato dal frontend al mount — porta i task locali iPhone nel backend.
    """
    added = 0
    async with _lock:
        for t in body:
            tid = t.get("id")
            if (
                tid
                and tid not in _tasks
                and t.get("status") in ("pending", "paused")
                and isinstance(t.get("trigger"), dict)
            ):
                _tasks[tid] = {
                    **t,
                    "lastResult": (t.get("lastResult") or "")
                                  + (" [synced da client]" if t.get("lastResult") else "[synced da client]"),
                }
                added += 1
        if added:
            _save_tasks_sync()
            _broadcast_sse()

    logger.info("Scheduler: sync — %d task aggiunti (totale: %d)", added, len(_tasks))
    return {"synced": added, "total": len(_tasks)}


@router.post("/trigger/{task_id}")
async def trigger_task_now(task_id: str) -> dict:
    """Esecuzione immediata ignorando il trigger temporale (debug / run manuale)."""
    async with _lock:
        task = _tasks.get(task_id)
        if not task:
            raise HTTPException(404, "Task non trovato")
    asyncio.create_task(_execute_task(task_id)).add_done_callback(_log_task_exc)
    return {"triggered": task_id, "label": task.get("label")}


@router.post("/tick")
async def external_tick(request: Request) -> dict:
    """
    Pacemaker esterno — MX11-SCHED.

    Chiamato dal CF Worker ogni 2min e da GHA daemon-cron come backup.
    Garantisce che lo scheduler giri anche senza browser aperto e
    sopravviva ai crash del loop asyncio Railway.

    Comportamento:
      - Esegue _tick() immediatamente (no attesa del ciclo 60s)
      - Auto-riavvia il loop asyncio se è morto (self-healing)
      - Idempotente: sicuro da più sorgenti concorrenti
      - Non richiede auth se INTERNAL_TOKEN non configurato

    Response: { ok, source, loopRevived, loopRunning, tasks, pending, running, ts }
    """
    global _loop_task

    source    = request.query_params.get("source", "external")

    # Token check opzionale — solo se INTERNAL_TOKEN è configurato
    _int_tok  = os.getenv("INTERNAL_TOKEN", "")
    if _int_tok:
        req_tok = request.headers.get("X-Internal-Token", "")
        if req_tok != _int_tok:
            raise HTTPException(403, "Unauthorized — X-Internal-Token richiesto")

    # Self-heal: riavvia il loop se morto
    loop_was_dead = _loop_task is None or _loop_task.done()
    if loop_was_dead:
        logger.warning(
            "Scheduler: loop morto rilevato da external tick (source=%s) — riavvio",
            source,
        )
        _loop_task = asyncio.create_task(_scheduler_loop())
        _loop_task.add_done_callback(_log_task_exc)

    # Esegui _tick() immediatamente (no attesa 60s)
    try:
        await _tick()
    except Exception as _exc:
        logger.error("Scheduler: external tick error: %s", _exc)

    async with _lock:
        total   = len(_tasks)
        pending = sum(1 for t in _tasks.values() if t.get("status") == "pending")
        running = sum(1 for t in _tasks.values() if t.get("status") == "running")

    loop_running = _loop_task is not None and not _loop_task.done()
    logger.info(
        "Scheduler tick (source=%s): loop=%s revived=%s tasks=%d pending=%d",
        source, loop_running, loop_was_dead, total, pending,
    )
    return {
        "ok":          True,
        "source":      source,
        "loopRevived": loop_was_dead,
        "loopRunning": loop_running,
        "tasks":       total,
        "pending":     pending,
        "running":     running,
        "ts":          int(time.time() * 1000),
    }


@router.get("/status")
async def scheduler_status() -> dict:
    """Stato del loop asyncio — usato dal frontend per il badge ☁️/📱."""
    loop_ok = _loop_task is not None and not _loop_task.done()
    async with _lock:
        total   = len(_tasks)
        pending = sum(1 for t in _tasks.values() if t.get("status") == "pending")
        running = sum(1 for t in _tasks.values() if t.get("status") == "running")
    return {
        "loopRunning":  loop_ok,
        "currentTask":  _current_running,
        "totalTasks":   total,
        "pendingTasks": pending,
        "runningTasks": running,
        "sseClients":   len(_sse_clients),
    }


@router.get("/delta")
async def scheduler_delta(since_ms: int = 0) -> dict:
    """GAP-A7: Delta-only view — solo task aggiornati dopo since_ms (epoch ms).

    Permette polling incrementale efficiente dal frontend:
      invece di ricevere tutti i task ad ogni tick, il client riceve solo
      quelli effettivamente cambiati dall'ultima lettura.

    Utilizzo:
      GET /api/scheduler/delta?since_ms=0         → tutti i task (full sync iniziale)
      GET /api/scheduler/delta?since_ms=1718000000000 → solo task aggiornati dopo ts

    Risposta:
      { "tasks": [...], "count": N, "since_ms": M, "now_ms": T, "total": TOT }

    Campo di riferimento per il filtro: updated_at (ms epoch).
    Se updated_at non presente sul task, usa createdAt come fallback.
    """
    import time as _t
    now_ms = int(_t.time() * 1000)
    async with _lock:
        if since_ms == 0:
            delta = list(_tasks.values())
        else:
            delta = [
                t for t in _tasks.values()
                if max(
                    t.get("updated_at", 0),
                    t.get("updatedAt",  0),
                    t.get("createdAt",  0),
                    t.get("created_at", 0),
                ) > since_ms
            ]
    return {
        "tasks":    delta,
        "count":    len(delta),
        "since_ms": since_ms,
        "now_ms":   now_ms,
        "total":    len(_tasks),
    }


@router.get("/webhook/sse")
async def sse_stream(request: Request) -> StreamingResponse:
    """
    SSE endpoint — il frontend si connette qui per ricevere aggiornamenti real-time.

    Protocollo:
      - event: tasks_updated  →  data: JSON array di tutti i task correnti
      - : heartbeat           →  commento SSE (mantiene viva la connessione su iOS/proxy)

    Il client riceve uno snapshot immediato alla connessione, poi push ad ogni mutazione.
    Riconnessione automatica gestita dal browser (EventSource ha retry built-in).
    """
    queue: asyncio.Queue = asyncio.Queue(maxsize=16)
    _sse_clients.append(queue)
    logger.info("Scheduler SSE: client connesso (totale: %d)", len(_sse_clients))

    async def cleanup_generator() -> AsyncGenerator[str, None]:
        try:
            async for event in _sse_generator(queue, request):
                yield event
        finally:
            try:
                _sse_clients.remove(queue)
            except ValueError as _exc:
                _logger.debug("[scheduler] silenced %s", type(_exc).__name__)  # noqa: BLE001
            logger.info("Scheduler SSE: client disconnesso (totale: %d)", len(_sse_clients))

    return StreamingResponse(
        cleanup_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control":    "no-cache",
            "X-Accel-Buffering": "no",   # disabilita buffering Nginx/proxy
            "Connection":       "keep-alive",
        },
    )