""" backend/api/global_state_sync.py — Global State Sync Layer (S766-GRID-2) Sincronizzazione della memoria e dello stato tra i profili A, B, C, D. Permette a ogni daemon di leggere la memoria degli altri profili per evitare duplicazioni e per mantenere una visione coerente dello stato globale. Architettura: - Supabase Federation: Legge da tutti i database (A, B, C, D) e unifica i risultati - Memory Merge: Combina i risultati mantenendo la coerenza - Conflict Resolution: In caso di conflitto, usa timestamp e versione per decidere """ import os import asyncio import logging from typing import Optional, Dict, List, Any from datetime import datetime, timedelta import json _logger = logging.getLogger("global_state_sync") # ── Configurazione ───────────────────────────────────────────────────────── SUPABASE_URLS = { "A": os.getenv("SUPABASE_URL", ""), "B": os.getenv("SUPABASE_URL_B", ""), "C": os.getenv("SUPABASE_URL_C", ""), "D": os.getenv("SUPABASE_URL_D", ""), } SUPABASE_KEYS = { "A": os.getenv("SUPABASE_KEY", ""), "B": os.getenv("SUPABASE_KEY_B", ""), "C": os.getenv("SUPABASE_KEY_C", ""), "D": os.getenv("SUPABASE_KEY_D", ""), } GLOBAL_STATE_SYNC_ENABLED = os.getenv("GLOBAL_STATE_SYNC_ENABLED", "true").lower() == "true" class SupabaseClient: """Client per accedere a un singolo database Supabase.""" def __init__(self, url: str, key: str, profile: str): self.url = url self.key = key self.profile = profile self.base_url = f"{url}/rest/v1" async def query(self, table: str, filters: Optional[Dict] = None) -> List[Dict]: """ Esegue una query su una tabella. Esempio: query("agent_memory", {"session_id": "xyz"}) """ import httpx url = f"{self.base_url}/{table}" headers = { "apikey": self.key, "Authorization": f"Bearer {self.key}", "Content-Type": "application/json", } try: async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers, timeout=10.0) if response.status_code == 200: return response.json() else: _logger.warning(f"Supabase {self.profile} query failed: {response.status_code}") return [] except Exception as exc: _logger.error(f"Supabase {self.profile} error: {exc}") return [] class GlobalStateSync: """Sincronizzazione dello stato globale tra i profili.""" def __init__(self): self.clients = {} self._enabled = GLOBAL_STATE_SYNC_ENABLED for profile, url in SUPABASE_URLS.items(): key = SUPABASE_KEYS.get(profile, "") if url and key: self.clients[profile] = SupabaseClient(url, key, profile) async def get_unified_memory(self, session_id: str) -> Dict[str, Any]: """ Recupera la memoria unificata per una sessione da tutti i profili. Combina i risultati e risolve i conflitti. """ if not self._enabled or not self.clients: return {} tasks = [] for profile, client in self.clients.items(): tasks.append( self._fetch_profile_memory(client, session_id) ) results = await asyncio.gather(*tasks, return_exceptions=True) # Unifica i risultati unified = {} for profile, result in zip(self.clients.keys(), results): if isinstance(result, dict): unified[profile] = result return self._merge_memories(unified) async def _fetch_profile_memory(self, client: SupabaseClient, session_id: str) -> Dict: """Recupera la memoria da un singolo profilo.""" try: rows = await client.query( "agent_memory", {"session_id": session_id} ) if rows: return { "profile": client.profile, "data": rows[0], # Prendi il primo risultato "fetched_at": datetime.now().isoformat(), } return {} except Exception as exc: _logger.error(f"Error fetching memory from {client.profile}: {exc}") return {} def _merge_memories(self, unified: Dict[str, Dict]) -> Dict[str, Any]: """ Unisce le memorie da più profili. Regole di conflitto: - Se un campo ha timestamp più recente, usa quello - Se è un array, unisci senza duplicati - Se è un oggetto, fai merge ricorsivo """ merged = { "profiles": list(unified.keys()), "merged_at": datetime.now().isoformat(), "data": {}, } if not unified: return merged # Estrai i dati da tutti i profili all_data = {} for profile, result in unified.items(): if result and "data" in result: all_data[profile] = result["data"] # Merge semplice: priorità al profilo A, poi B, C, D for profile in ["A", "B", "C", "D"]: if profile in all_data: merged["data"].update(all_data[profile]) return merged async def get_health_status(self) -> Dict[str, str]: """Restituisce lo stato di connettività di tutti i profili Supabase.""" status = {} for profile, client in self.clients.items(): try: rows = await client.query("agent_memory", {}) status[profile] = "online" if rows is not None else "offline" except Exception: status[profile] = "offline" return status # ── Singleton globale ────────────────────────────────────────────────────── _global_state_sync_instance: Optional[GlobalStateSync] = None def get_global_state_sync() -> GlobalStateSync: """Restituisce l'istanza globale del GlobalStateSync.""" global _global_state_sync_instance if _global_state_sync_instance is None: _global_state_sync_instance = GlobalStateSync() return _global_state_sync_instance