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: derivato da Policy Engine per risk level (safe=30s, medium=90s, risky=180s, dangerous=300s)
  - 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)
  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 zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from .state import safe_json_dumps
from .task_tool_policy import build_task_tool_policy
import os
import time
import uuid
from pathlib import Path
from typing import Any, AsyncGenerator, Optional

from fastapi import APIRouter, Depends, HTTPException, Request
from .auth_guard import require_role, AuthRole
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"], dependencies=[Depends(require_role(AuthRole.MACHINE))])  # GAP-1-fix: tutti endpoint scheduler ora fail-closed

# ─── Telegram notifications (fire-and-forget) ─────────────────────────────────
# Definisci sempre le funzioni dummy PRIMA di qualsiasi import tentativo.
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]

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 as _tg_import_err:
    logger.debug("[scheduler] Telegram notifications unavailable: %s", _tg_import_err)

# ─── 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.
# ─── Policy Engine integration (ARCH-I4.6) ────────────────────────────────────
# Fail-open: se policy non disponibile → fallback ai valori originali hardcoded.
try:
    from .policy import (
        RISK_TIMEOUT_S as _POLICY_TIMEOUT_S,
        RISK_MAX_RETRY as _POLICY_MAX_RETRY,
        _quota_check   as _policy_quota_check,
        _quota_consume as _policy_quota_consume,
    )
    _POLICY_AVAILABLE = True
except Exception as _policy_import_err:  # pragma: no cover
    logger.warning("[scheduler] Policy Engine non disponibile — fallback hardcoded: %s", _policy_import_err)
    _POLICY_TIMEOUT_S  = {"safe": 30, "medium": 120, "risky": 180, "dangerous": 300}
    _POLICY_MAX_RETRY  = {"safe": 3,  "medium": 2,   "risky": 1,   "dangerous": 0}
    def _policy_quota_check(sid, tool, risk):  # type: ignore[misc]
        return True, 99
    def _policy_quota_consume(sid, tool):  # type: ignore[misc]
        pass
    _POLICY_AVAILABLE  = False

_VALID_RISK   = frozenset(_POLICY_TIMEOUT_S)
_DEFAULT_RISK = "medium"

# DEAD-LETTER-WATCHDOG: margine sopra il timeout massimo del livello dangerous.
_STUCK_TIMEOUT_S = max(_POLICY_TIMEOUT_S.values()) + 120  # 300 + 120 = 420s


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")


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).
    """
    if not _sse_clients:
        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


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 _daily_timezone(trigger: dict) -> ZoneInfo | None:
    """Ritorna il fuso IANA salvato dal browser, se disponibile e valido.

    I task daily creati prima dell'introduzione del campo ``timeZone`` restano
    compatibili: l'assenza o un valore non valido mantiene il calcolo nel fuso
    locale del server invece di bloccare la pianificazione.
    """
    time_zone = trigger.get("timeZone")
    if not isinstance(time_zone, str) or not time_zone:
        return None
    try:
        return ZoneInfo(time_zone)
    except ZoneInfoNotFoundError:
        logger.warning("Scheduler: timezone daily non valida (%r), fallback server-local", time_zone)
        return None


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)
        time_zone = _daily_timezone(t)
        # Usa il timestamp dell'esecuzione, non l'orologio nel momento in cui
        # il task termina: preserva la semantica esistente anche per task lunghi.
        now = datetime.datetime.fromtimestamp(
            now_ms / 1000,
            tz=time_zone,
        ) if time_zone else datetime.datetime.fromtimestamp(now_ms / 1000)
        nxt = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
        if nxt <= now:
            nxt = nxt + datetime.timedelta(days=1)
        t["nextRun"] = int(nxt.timestamp() * 1000)
    # once / on_open: nessun avanzamento
    return t


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

async def _run_goal(goal: str, conversation_id: Optional[str] = None, risk: str = "medium") -> str:
    """
    Esegue il goal tramite UnifiedAgentLoop (stesso path di api/agent.py).
    Timeout: derivato da Policy Engine per risk level (safe=30s, medium=90s, risky=180s, dangerous=300s).
    """
    _task_policy = build_task_tool_policy(goal)
    if _task_policy.literal_response:
        return _task_policy.literal_response

    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,
        )

        _timeout_s = float(_POLICY_TIMEOUT_S.get(risk, 120))
        result = await asyncio.wait_for(
            loop.run(goal=goal, context="", max_steps=8,
                     allow_tools=not _task_policy.forbid_tools),
            timeout=_timeout_s,
        )
        if isinstance(result, dict):
            # Preserve structured loop outcomes; never turn controlled failures into empty strings.
            output = next((result.get(key) for key in ("output", "answer", "explanation", "error")
                           if result.get(key)), "")
        else:
            output = str(result)
        return str(output)[:1000]

    except asyncio.TimeoutError:
        return f"❌ Timeout: task terminato dopo {int(_POLICY_TIMEOUT_S.get(risk, 120))}s"
    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]
        _task_risk    = task.get("risk", _DEFAULT_RISK)
        if _task_risk not in _VALID_RISK:
            _task_risk = _DEFAULT_RISK
        _save_tasks_sync()
        _broadcast_sse()
    # Policy Engine (ARCH-I4.6): quota check fail-open — non blocca, solo log warning.
    _quota_ok, _quota_rem = _policy_quota_check("scheduler", "scheduled_task", _task_risk)
    if not _quota_ok:
        logger.warning(
            "[scheduler] quota esaurita (risk=%s) per task %s — eseguo comunque (fail-open)",
            _task_risk, task_id,
        )
    else:
        _policy_quota_consume("scheduler", "scheduled_task")
    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"), risk=_task_risk)

        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", _POLICY_MAX_RETRY.get(_task_risk, 2))
            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.
    """
    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():
        _loop_task = asyncio.create_task(_scheduler_loop())
        _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
    risk:           str            = "medium"  # ARCH-I4.6: safe | medium | risky | dangerous


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,
        "risk":           body.risk if body.risk in _VALID_RISK else _DEFAULT_RISK,
    }
    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."""
    async with _lock:
        if task_id not in _tasks:
            raise HTTPException(404, "Task non trovato")
        del _tasks[task_id]
        _save_tasks_sync()
        _broadcast_sse()


@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.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",
        },
    )