""" backend/api/state.py — Shared state for all API routers (S354). Contains: Supabase client, in-memory stores, singleton getters, shared Pydantic models, TTL constants, prune helpers. Extracted from main.py — zero behaviour change. """ import os, time, asyncio as _asyncio_mod, json as _json, re as _re from typing import Optional, Any from fastapi import HTTPException from pydantic import BaseModel, field_validator, model_validator import logging _logger = logging.getLogger("api.state") # ── Surrogate-safe JSON serialiser (shared utility) ───────────────────────── # Lone UTF-16 surrogates (U+D800-U+DFFF) crash json.dumps even with ensure_ascii=False. # Use safe_json_dumps() as a drop-in replacement wherever task/LLM data is serialised. _RE_SURR = _re.compile(r'[\ud800-\udfff]') def _strip_surr(v: object) -> object: if isinstance(v, str): return _RE_SURR.sub('', v) if isinstance(v, dict): return {k: _strip_surr(val) for k, val in v.items()} if isinstance(v, (list, tuple)): return [_strip_surr(i) for i in v] return v def safe_json_dumps(obj: object, *, ensure_ascii: bool = False, **kw) -> str: """Drop-in for json.dumps that strips lone UTF-16 surrogates before serialisation.""" return _json.dumps(_strip_surr(obj), ensure_ascii=ensure_ascii, **kw) # ── Supabase client ─────────────────────────────────────────────────────────── _sb: Any = None _sb2: Any = None try: _SUPA_URL = os.getenv('SUPABASE_URL', '') _SUPA_KEY = os.getenv('SUPABASE_KEY') or os.getenv('SUPABASE_ANON_KEY', '') if _SUPA_URL and _SUPA_KEY: from supabase import create_client _sb = create_client(_SUPA_URL, _SUPA_KEY) _SUPA_URL2 = os.getenv("SUPABASE_URL_2", "") _SUPA_KEY2 = os.getenv("SUPABASE_KEY_2", "") if _SUPA_URL2 and _SUPA_KEY2: _sb2 = create_client(_SUPA_URL2, _SUPA_KEY2) _logger.info("BOOT: Supabase #2 connected OK") _logger.info('BOOT: Supabase connected OK') else: _logger.warning('BOOT: Supabase not configured (SUPABASE_URL / SUPABASE_KEY missing)') except Exception as e: _logger.error('BOOT: Supabase init failed: %s', e) def sb() -> Any: if not _sb and not _sb2: raise HTTPException(503, detail={ 'error': 'supabase_not_configured', 'message': 'Imposta SUPABASE_URL e SUPABASE_KEY nelle variabili Railway/HuggingFace per abilitare la persistenza.', }) return _sb or _sb2 def sb_dual() -> list: """Ritorna entrambi i client se disponibili, per sharding o ridondanza.""" return [s for s in [_sb, _sb2] if s] # ── Sensitive keys mask ─────────────────────────────────────────────────────── SENSITIVE = { 'OPENROUTER_API_KEY', 'OPENAI_API_KEY', 'GEMINI_API_KEY', 'GROQ_API_KEY', 'HF_TOKEN', 'HUGGINGFACE_API_KEY', 'GH_TOKEN', 'GITHUB_TOKEN', 'QDRANT_API_KEY', 'DATABASE_URL', 'SESSION_SECRET', 'SECRET_KEY', 'RAILWAY_TOKEN', 'SUPABASE_KEY', 'SUPABASE_ANON_KEY', # security-fix: tutti i segreti esposti da /api/status 'TELEGRAM_BOT_TOKEN', 'TELEGRAM_CHAT_ID', 'CF_API_TOKEN', 'CLOUDFLARE_API_TOKEN', 'CF_ACCOUNT_ID', 'CEREBRAS_API_KEY', 'SAMBANOVA_API_KEY', 'VAULT_KEY', 'INTERNAL_TOKEN', 'DEPLOY_SECRET', 'WEBHOOK_TOKEN', 'TERMINAL_SECRET', 'EXEC_TOKEN', 'VITE_INTERNAL_TOKEN', 'VITE_TERMINAL_SECRET', 'VITE_OPENROUTER_API_KEY', 'VITE_HF_TOKEN', 'VITE_GROQ_API_KEY', 'GH_PAGES_TOKEN', 'VERCEL_TOKEN', } # ── In-memory fallback for agent memory (when Supabase not available) ───────── _mem_fallback: dict[str, dict] = {} # ── In-memory agent task registry (FASE 2.1) ────────────────────────────────── _agent_tasks: dict[str, dict] = {} # ── Active run-stream tasks (abort support) ────────────────────────────────── # Per-task: asyncio_task + asyncio_queue. Abort endpoint mette __abort__ nella queue # e chiama task.cancel(). Cleanup automatico nel finally della generate() closure. _run_stream_tasks: dict[str, dict] = {} # ── Running loop registry (S358: SSE reconnect safety) ──────────────────────── # Per-task: asyncio_task, event_buffer (list[str]), subscriber_queues, done flag. # On reconnect: replay buffer[resume_from:] + subscribe to fanout — NO re-run. _loop_registry: dict[str, dict] = {} _LOOP_REGISTRY_TTL_S: float = 10 * 60 # 10 min after completion # ── GAP-STATE: Supabase snapshot (boot restore + 60s checkpoint) ────────────── # Fix per: Railway/HF restart = perdita totale _agent_tasks in-memory. # Soluzione: Supabase è già importato — usiamo tabella backend_state (key/value). _last_snap_hash: str = "" async def persist_state_snapshot() -> None: """Bg task: snapshotta _agent_tasks su Supabase backend_state. ADAPTIVE-CHECKPOINT: 15s se ci sono task 'running', 60s se idle. Riduce la finestra di perdita dati da 60s a 15s durante esecuzione attiva. Silent failure se Supabase assente o tabella non esiste (free tier graceful). """ global _last_snap_hash import hashlib as _hs, json as _json, time as _t, asyncio as _aio while True: _has_running = any(v.get('status') == 'running' for v in _agent_tasks.values()) await _aio.sleep(15 if _has_running else 60) if _sb is None: continue try: # Solo campi scalari/JSON — esclude asyncio.Task, Event, Queue (non serializzabili) _snap: dict[str, dict] = { k: {ck: cv for ck, cv in v.items() if isinstance(cv, (str, int, float, bool, type(None), list))} for k, v in list(_agent_tasks.items())[-50:] } _payload = _json.dumps(_snap, ensure_ascii=False, default=str) _h = _hs.md5(_payload.encode()).hexdigest() if _h == _last_snap_hash: continue # nessuna variazione — non tocca Supabase _last_snap_hash = _h _sb.table("backend_state").upsert( {"key": "agent_tasks_snap", "value": _payload, "ts": _t.time()}, on_conflict="key", ).execute() except Exception as _snap_err: _logger.warning('STATE-SNAP warn: %s', _snap_err) async def restore_agent_tasks_from_snap() -> int: """Boot: ripopola _agent_tasks dall'ultimo Supabase snapshot. Ritorna n task ripristinati. GAP-STATE: previene perdita totale su restart.""" import json as _json if _sb is None: return 0 try: res = ( _sb.table("backend_state") .select("value") .eq("key", "agent_tasks_snap") .maybe_single() .execute() ) if not res or not res.data: return 0 snap: dict = _json.loads(res.data["value"]) n = 0 for task_id, data in snap.items(): if task_id not in _agent_tasks and isinstance(data, dict): data["_snap_restored"] = True # flag visibile nel debug _agent_tasks[task_id] = data n += 1 if n: _logger.info('BOOT: GAP-STATE restored %d agent_tasks from Supabase', n) return n except Exception as _re: _logger.warning('BOOT: GAP-STATE restore failed (non-critical): %s', _re) return 0 async def write_ahead_task_created(task_id: str, goal: str) -> None: """WRITE-AHEAD: persiste un task immediatamente alla creazione, senza aspettare il checkpoint. Riduce a zero la finestra di perdita per la fase di creazione task. Legge lo snapshot esistente, aggiunge la nuova entry, riscrive atomicamente. Silent failure su Supabase non disponibile (graceful degradation). """ if _sb is None: return import time as _t, json as _json try: _new_entry = {task_id: { 'goal': goal[:500], 'status': 'pending', 'created_at': int(_t.time() * 1000), '_write_ahead': True, }} try: _existing = ( _sb.table('backend_state') .select('value') .eq('key', 'agent_tasks_snap') .maybe_single() .execute() ) if _existing and _existing.data: import json as _j2 _current: dict = _j2.loads(_existing.data['value']) # Mantieni max 50 entry — stessa policy del checkpoint periodico if len(_current) >= 50: oldest_keys = sorted(_current, key=lambda k: _current[k].get('created_at', 0)) for _k in oldest_keys[:len(_current) - 49]: _current.pop(_k, None) _current.update(_new_entry) _new_entry = _current except Exception as _exc: _logger.debug("[state] silenced %s", type(_exc).__name__) # noqa: BLE001 _payload = _json.dumps(_new_entry, ensure_ascii=False, default=str) _sb.table('backend_state').upsert( {'key': 'agent_tasks_snap', 'value': _payload, 'ts': _t.time()}, on_conflict='key', ).execute() except Exception as _wa_err: _logger.warning('WRITE-AHEAD warn: %s', _wa_err) # ── In-memory task checkpoint store (Sessione 18: Reconnect Authority) ───────── _task_checkpoints: dict[str, dict] = {} _CHECKPOINT_TTL_MS = 2 * 60 * 60 * 1000 # 2 ore _CHECKPOINT_MAX = 100 # ── AI provider health cache — 60s TTL ─────────────────────────────────────── _ai_health_cache: dict = {"data": None, "at": 0.0} # S385: latency telemetry — circular buffer (max 200 samples per metric) # Sprint 5: aggiunti classify_ms, plan_ms, coder_ms, verifier_ms, browser_ms per fase breakdown _TIMING_STORE: dict[str, list[float]] = { "llm_first_token": [], "llm_total": [], "tool_call": [], "direct_tool": [], # Sprint 5: timing per fase del loop agente "classify_ms": [], # fase classificazione goal "plan_ms": [], # fase planner "coder_ms": [], # fase coder LLM (70B) "verifier_ms": [], # fase GoalVerifier "browser_ms": [], # fase browser verify (Playwright) # Gap-3: time-to-* metrics per sessione agente "ttfa_ms": [], # Time To First Action (ms dalla prima call LLM al primo tool) "ttfr_ms": [], # Time To First Response (ms al primo text_chunk) "ttr_ms": [], # Time To Resolution (ms durata totale run) "mean_fix_ms": [], # Durata media di un repair riuscito } _TIMING_MAX_SAMPLES = 200 def record_timing(label: str, ms: float) -> None: """Append a timing sample to _TIMING_STORE (thread-safe via GIL for list.append).""" buf = _TIMING_STORE.get(label) if buf is None: return buf.append(round(ms, 1)) if len(buf) > _TIMING_MAX_SAMPLES: del buf[:len(buf) - _TIMING_MAX_SAMPLES] # S395: Repair telemetry counters — syntax/runtime errors + repair outcomes + GREEN confirmation # S410: aggiunto goal_verify_* per tracciare l'efficacia del GoalVerifier # Sprint 5: aggiunti goal_success_count, goal_fail_count, repair_success_count, tool_failure_count _REPAIR_STATS: dict[str, int] = { "syntax_errors": 0, # SyntaxError rilevati "syntax_repaired": 0, # repair syntax OK "syntax_failed": 0, # repair syntax fallito "runtime_errors": 0, # runtime errors rilevati "runtime_repaired": 0, # repair runtime OK (LLM ha prodotto fix) "runtime_failed": 0, # repair runtime fallito (LLM timeout / error) "browser_dom_check_pass": 0, # S701: verify_goal_browser senza req → DOM ok "browser_dom_check_fail": 0, # S701: verify_goal_browser senza req → white screen/JS err "browser_quality_pass": 0, # quality_guardian HTML/browser test PASS "browser_quality_fail": 0, # quality_guardian HTML/browser test FAIL "green_confirmed": 0, # re-esecuzione post-repair: GREEN (rc==0) "green_failed": 0, # re-esecuzione post-repair: ancora errori # S410: GoalVerifier telemetry "goal_verify_initial_pass": 0, # goal soddisfatto già al primo check (no repair) "goal_verify_repair_triggered": 0, # coverage < threshold → repair avviato "goal_verify_repaired": 0, # repair applicato (re-verify OK o ≥ -5%) "goal_verify_no_improvement": 0, # repair peggiorativo → risposta originale mantenuta # Sprint 5: contatori qualità aggregata per TelemetryDashboard "goal_success_count": 0, # goal completati con successo (output reale) "goal_fail_count": 0, # goal falliti (LLM error o output vuoto) "repair_success_count": 0, # totale repair riusciti (syntax+runtime+goal) "tool_failure_count": 0, # totale tool call fallite "req_engine_used": 0, # RequirementEngine attivato (Sprint 2) "req_engine_reqs_total": 0, # requisiti decomposed totali "goal_verifier_v2_used": 0, # GoalVerifier 2.0 attivato su goal con requisiti # S701: browser verify outcome counters (dynamic key in unified_loop) "browser_verify_pass": 0, # verify_goal_browser → PASS "browser_verify_fail": 0, # verify_goal_browser → FAIL "browser_verify_unknown": 0, # verify_goal_browser → UNKNOWN (timeout/no-url) "browser_verify_timeout": 0, # verify_goal_browser asyncio.TimeoutError # S703: repair iteration counters "repair_iter2_used": 0, # quality_guardian iter 2 (rewrite) attivato "repair_iter3_used": 0, # quality_guardian iter 3 (simplify) attivato # S704: browser screenshot/DOM quality counters "browser_screenshot_blank": 0, # screenshot < 2500B = pagina bianca/vuota "browser_dom_sparse": 0, # DOM < 5 elementi = pagina quasi vuota } def increment_stat(key: str) -> None: """Increment a _REPAIR_STATS counter (thread-safe via GIL for int += op).""" if key in _REPAIR_STATS: _REPAIR_STATS[key] += 1 _AI_HEALTH_TTL = 60.0 # ── Provider heartbeat state ───────────────────────────────────────────────── _heartbeat_state: dict = { "last_run_at": None, "next_run_at": None, "best_provider": None, "best_latency_ms": None, "providers": [], "runs": 0, "status": "pending", "error": None, } # ── Agent task TTL ──────────────────────────────────────────────────────────── _AGENT_TASK_TTL_MS = 2 * 60 * 60 * 1000 # 2 ore _AGENT_TASK_MAX = 200 # ── MemoryManager singleton ──────────────────────────────────────────────────── _mem_manager: Any = None _mem_manager_inited: bool = False _mem_manager_lock: Any = None # asyncio.Lock creato lazily (il loop async potrebbe non esistere a import-time) def _get_mem_manager_lock() -> Any: """Lazy asyncio.Lock — N-1-FIX: protegge da race condition su init() concorrenti.""" global _mem_manager_lock if _mem_manager_lock is None: _mem_manager_lock = _asyncio_mod.Lock() return _mem_manager_lock async def _get_mem_manager_async() -> Any: """N-1-FIX: versione async con asyncio.Lock per evitare race condition su request concorrenti. Usare in tutti i contesti async — evita la creazione di N istanze MemoryManager in parallelo su boot con 3+ worker uvicorn che arrivano contemporaneamente quando _mem_manager è None.""" global _mem_manager, _mem_manager_inited if _mem_manager is not None and _mem_manager_inited: return _mem_manager async with _get_mem_manager_lock(): if _mem_manager is None: try: from memory.manager import MemoryManager _mem_manager = MemoryManager() await _mem_manager.init() _mem_manager_inited = True except Exception: _mem_manager = None elif not _mem_manager_inited: try: await _mem_manager.init() _mem_manager_inited = True except Exception: _mem_manager_inited = True # evita retry infiniti return _mem_manager def _get_mem_manager() -> Any: """ S442-FIX2: init più robusto. - _mem_manager_inited viene impostato a True anche su RuntimeError (nessun loop in corso) così da non ritentare il get_running_loop() ad ogni request (era un retry silenzioso infinito). - Se il loop non era disponibile al primo call (es. startup sync), asyncio.ensure_future() viene usato come fallback al successivo call in contesto async. """ global _mem_manager, _mem_manager_inited if _mem_manager is not None: if not _mem_manager_inited: try: import asyncio as _asyncio_inner _loop_inner = _asyncio_inner.get_running_loop() _loop_inner.create_task(_mem_manager.init()) _mem_manager_inited = True except RuntimeError: # Nessun loop in esecuzione ora — segniamo come inizializzato per evitare # retry infiniti. Il init() verrà tentato al prossimo call in contesto async. _mem_manager_inited = True return _mem_manager try: from memory.manager import MemoryManager import asyncio as _asyncio _mem_manager = MemoryManager() try: loop = _asyncio.get_running_loop() loop.create_task(_mem_manager.init()) _mem_manager_inited = True except RuntimeError as _exc: # Chiamata in contesto sync (es. import-time) — init rimandato al primo call async. # _mem_manager_inited resta False → verrà ritentato al prossimo call in loop. _logger.debug("[state] silenced %s", type(_exc).__name__) # noqa: BLE001 except Exception: _mem_manager = None return _mem_manager # ── Executor singleton ───────────────────────────────────────────────────────── _executor: Any = None def _get_executor() -> Any: global _executor if _executor is not None: return _executor try: from agents.executor import Executor _executor = Executor(memory=_get_mem_manager()) except Exception: _executor = None return _executor # ── AIClient singleton (S388) ────────────────────────────────────────────────── # Un'unica istanza condivisa tra tutte le request → nessuna re-istanziazione di # OpenAI() a ogni call. _client_cache interno all'istanza riusa i connection pool. _ai_client: Any = None def _get_ai_client() -> Any: global _ai_client if _ai_client is not None: return _ai_client try: from models.ai_client import AIClient _ai_client = AIClient() except Exception: _ai_client = None return _ai_client # ── Planner singleton ────────────────────────────────────────────────────────── _planner: Any = None def _get_planner() -> Any: global _planner if _planner is not None: return _planner try: from agents.planner import Planner _planner = Planner(llm_client=_get_ai_client()) except Exception: _planner = None return _planner # ── Prune helpers ───────────────────────────────────────────────────────────── def _prune_checkpoints() -> None: now = int(time.time() * 1000) _snap_cp = list(_task_checkpoints.items()) expired = [k for k, v in _snap_cp if now - v.get('savedAt', 0) > _CHECKPOINT_TTL_MS] for k in expired: _task_checkpoints.pop(k, None) if len(_task_checkpoints) > _CHECKPOINT_MAX: oldest = sorted(list(_task_checkpoints.items()), key=lambda x: x[1].get('savedAt', 0)) for k, _ in oldest[:len(_task_checkpoints) - _CHECKPOINT_MAX]: _task_checkpoints.pop(k, None) def _prune_agent_tasks() -> None: now = int(time.time() * 1000) _snap_at = list(_agent_tasks.items()) expired = [ k for k, v in _snap_at if v.get('status') in ('SUCCESS', 'ERROR', 'CANCELLED') and now - v.get('created_at', 0) > _AGENT_TASK_TTL_MS ] for k in expired: _agent_tasks.pop(k, None) if len(_agent_tasks) > _AGENT_TASK_MAX: oldest = sorted(list(_agent_tasks.items()), key=lambda x: x[1].get('created_at', 0)) for k, _ in oldest[:len(_agent_tasks) - _AGENT_TASK_MAX]: _agent_tasks.pop(k, None) def _prune_loop_registry() -> None: """Remove completed loop entries older than TTL to free memory.""" now = time.time() stale = [ k for k, v in list(_loop_registry.items()) if v.get('done') and now - v.get('finished_at', 0.0) > _LOOP_REGISTRY_TTL_S ] for k in stale: _loop_registry.pop(k, None) # ── Shared Pydantic models ──────────────────────────────────────────────────── class ReasonLoopIn(BaseModel): goal: str context: list[dict] = [] max_steps: int = 8 # S456-X5: project memory context injected by frontend (projectMemory.getContext()) project_context: str = "" # S456-X4: top failure patterns from frontend selfLearning engine learning_hints: list[str] = [] # BG-4: session identifier for cross-session handoff restore session_id: Optional[str] = None negative_constraints: Optional[str] = "" # P35 @field_validator('goal', mode='before') @classmethod def validate_goal(cls, v: object) -> str: if not isinstance(v, str) or not v.strip(): raise ValueError('goal must be a non-empty string') return v.strip() @field_validator('context', mode='before') @classmethod def coerce_context(cls, v: object) -> list: if v is None or v == '' or v == 'null': return [] if isinstance(v, list): return v if isinstance(v, str): return [] return [] @field_validator('project_context', mode='before') @classmethod def coerce_project_context(cls, v: object) -> str: if not isinstance(v, str): return "" return v.strip()[:2000] # cap a 2000 chars per evitare prompt bloat @field_validator('learning_hints', mode='before') @classmethod def coerce_learning_hints(cls, v: object) -> list: if not isinstance(v, list): return [] return [str(h)[:300] for h in v[:5]] # S606: 200→300 — hint completo class AgentTaskIn(BaseModel): goal: str context: list[dict] = [] max_steps: int = 8 taskId: Optional[str] = None # S456-X5/X4: stesso payload di ReasonLoopIn per il path tasks project_context: str = "" learning_hints: list[str] = [] # BG-4: session identifier for cross-session handoff restore session_id: Optional[str] = None # P16-F3: passo da cui riprendere (resume task promosso dalla coda) resume_from_step: Optional[int] = None # P17-F5: Expertise Persona — hint semantico per selezionare LLM/stile agente # Valori: "auto"|"researcher"|"coder"|"architect"|"reasoner"|"analyst"|None persona: Optional[str] = None negative_constraints: Optional[str] = "" # P35: vincoli negativi dal frontend @field_validator('goal', mode='before') @classmethod def validate_goal(cls, v: object) -> str: if not isinstance(v, str) or not v.strip(): raise ValueError('goal must be a non-empty string') return v.strip() @field_validator('context', mode='before') @classmethod def coerce_context(cls, v: object) -> list: if v is None or v == '' or v == 'null': return [] if isinstance(v, list): return v if isinstance(v, str): return [] return [] @field_validator('project_context', mode='before') @classmethod def coerce_project_context(cls, v: object) -> str: if not isinstance(v, str): return "" return v.strip()[:2000] @field_validator('learning_hints', mode='before') @classmethod def coerce_learning_hints(cls, v: object) -> list: if not isinstance(v, list): return [] return [str(h)[:300] for h in v[:5]] # S606: 200→300