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