Terminal / api /state.py
Baida—-
deploy: full backend sync for Node A [Kernel Restore]
8b87770 verified
Raw
History Blame
26 kB
"""
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)
# Leggi SUPABASE_URL_B (canonico multi-instance) con fallback legacy SUPABASE_URL_2
_SUPA_URL2 = os.getenv("SUPABASE_URL_B") or os.getenv("SUPABASE_URL_2", "")
_SUPA_KEY2 = os.getenv("SUPABASE_KEY_B") or 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