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