api/state.py

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