sync: 149 file da Baida98/AI@37832425 (2026-07-01 10:26 UTC)

REBUILD_TRIGGER

@@ -0,0 +1 @@
+rebuild

agents/context_manager.py

@@ -18,8 +18,11 @@ from __future__ import annotations
 import asyncio
 import hashlib
 import re
+import logging
 from typing import Any
 
+_logger = logging.getLogger("agente_ai.context_manager")
+
 _FUNC_RE   = re.compile(
     r'^(?:export\s+)?(?:async\s+)?(?:function\s+(\w+)|const\s+(\w+)\s*=\s*(?:async\s*)?\()',
     re.MULTILINE)
@@ -50,89 +53,66 @@ _RANK_ENTRY_STEMS = {'main', 'index', 'app', '__init__', 'config', 'settings', '
 _CAMEL_SPLIT_RE = re.compile(r'([a-z])([A-Z])')
 
 # ── FIX-SYN-EXPAND: tabella sinonimi tecnici IT↔EN (15 cluster) ───────────────
-# Struttura: ogni entry è un frozenset di termini equivalenti.
-# _expand_tokens() aggiunge tutti i sinonimi di ogni token del goal prima del matching.
-# Scelta design: sinonimi statici (zero LLM, zero latency) coprono l'80% dei task reali.
-# I cluster coprono i domini più frequenti nello sviluppo software.
 _SYN_CLUSTERS: list[frozenset[str]] = [
-    # Auth / Sicurezza
     frozenset({'auth', 'autenticazione', 'authentication', 'login', 'signin',
                'guard', 'middleware', 'jwt', 'token', 'session', 'oauth',
                'passport', 'credential', 'permission', 'role', 'accesso'}),
-    # Pagamenti
     frozenset({'payment', 'pagamento', 'stripe', 'checkout', 'invoice',
                'billing', 'subscription', 'abbonamento', 'fattura', 'webhook',
                'price', 'plan', 'tier'}),
-    # Database / ORM
     frozenset({'database', 'db', 'schema', 'model', 'migration', 'migrazione',
                'orm', 'repository', 'query', 'drizzle', 'prisma', 'postgres',
                'sqlite', 'mysql', 'table', 'tabella', 'record'}),
-    # API / Network
     frozenset({'api', 'endpoint', 'route', 'rotta', 'router', 'server',
                'request', 'response', 'richiesta', 'risposta', 'http',
                'rest', 'graphql', 'fetch', 'axios', 'client'}),
-    # UI / Frontend
     frozenset({'component', 'componente', 'ui', 'interface', 'interfaccia',
                'button', 'form', 'modal', 'layout', 'page', 'pagina',
                'style', 'css', 'theme', 'tema', 'render', 'view'}),
-    # State Management
     frozenset({'state', 'stato', 'store', 'redux', 'zustand', 'context',
                'provider', 'hook', 'reducer', 'action', 'dispatch',
                'observable', 'signal', 'reactive'}),
-    # File / Storage
     frozenset({'file', 'upload', 'caricamento', 'storage', 'bucket',
                'download', 'attachment', 'allegato', 'blob', 'stream',
                'filesystem', 'directory', 'path', 'percorso'}),
-    # Testing
     frozenset({'test', 'testing', 'spec', 'unit', 'integration', 'e2e',
                'mock', 'stub', 'fixture', 'assert', 'expect', 'coverage',
                'vitest', 'jest', 'pytest'}),
-    # Build / Deploy
     frozenset({'build', 'deploy', 'deployment', 'bundle', 'webpack', 'vite',
                'esbuild', 'compile', 'dist', 'production', 'staging',
                'pipeline', 'ci', 'cd', 'docker', 'container'}),
-    # Email / Notifiche
     frozenset({'email', 'mail', 'smtp', 'notification', 'notifica', 'alert',
                'push', 'telegram', 'slack', 'webhook', 'message', 'messaggio',
                'sendgrid', 'resend', 'mailer'}),
-    # AI / ML
     frozenset({'ai', 'llm', 'model', 'prompt', 'embedding', 'rag',
                'vector', 'semantic', 'chat', 'completion', 'inference',
                'openai', 'gemini', 'groq', 'anthropic', 'agent', 'agente'}),
-    # Errori / Debug
     frozenset({'error', 'errore', 'exception', 'eccezione', 'bug', 'fix',
                'debug', 'log', 'logging', 'trace', 'stack', 'crash',
                'fallback', 'retry', 'recover', 'handler', 'catch'}),
-    # Configurazione
     frozenset({'config', 'configurazione', 'configuration', 'settings',
                'impostazioni', 'env', 'environment', 'variable', 'variabile',
                'secret', 'segreto', 'dotenv', 'constant', 'costante'}),
-    # Performance / Cache
     frozenset({'cache', 'performance', 'performanza', 'speed', 'velocità',
                'optimize', 'ottimizzazione', 'lazy', 'memo', 'debounce',
                'throttle', 'batch', 'compress', 'compressione'}),
-    # Sicurezza / Validazione
     frozenset({'validation', 'validazione', 'validate', 'sanitize',
                'sanitizzazione', 'schema', 'zod', 'yup', 'joi',
                'csrf', 'xss', 'injection', 'escape', 'secure'}),
-    # Monitoring / Observability (B-GAP-D: cluster mancante — task metriche/dashboard non rankati)
     frozenset({'metrics', 'metric', 'monitoring', 'monitoraggio', 'observability',
                'prometheus', 'grafana', 'dashboard', 'telemetry', 'telemetria',
                'tracing', 'trace', 'health', 'healthcheck', 'uptime', 'alerting',
                'datadog', 'sentry', 'newrelic', 'audit', 'report'}),
-    # Scheduling / Background Jobs (B-GAP-D: cluster mancante — task cron/queue/worker)
     frozenset({'cron', 'scheduler', 'pianificatore', 'schedule', 'queue', 'coda',
                'worker', 'job', 'background', 'celery', 'bull', 'bullmq',
                'agenda', 'delayed', 'periodic', 'retry', 'backoff', 'redis',
                'task', 'processo', 'process', 'daemon'}),
-    # WebSocket / Realtime (B-GAP-D: cluster mancante — task ws/sse/pubsub)
     frozenset({'websocket', 'ws', 'socket', 'socketio', 'realtime', 'real_time',
                'sse', 'server_sent', 'pubsub', 'publish', 'subscribe', 'broadcast',
                'channel', 'canale', 'room', 'event', 'listener', 'emitter',
                'live', 'push', 'poll', 'long_polling', 'signalr', 'liveview'}),
 ]
 
-# Indice inverso: token → frozenset di sinonimi (costruito una volta a import)
 _SYN_INDEX: dict[str, frozenset[str]] = {}
 for _cluster in _SYN_CLUSTERS:
     for _term in _cluster:
@@ -140,20 +120,6 @@ for _cluster in _SYN_CLUSTERS:
 
 
 def _expand_tokens(tokens: list[str]) -> list[str]:
-    """
-    FIX-SYN-EXPAND: espande ogni token del goal con i sinonimi IT/EN del suo cluster.
-
-    Esempio:
-      ["autenticazione", "aggiungi"] → ["autenticazione", "aggiungi",
-        "auth", "login", "guard", "middleware", "jwt", ...]
-
-    Garanzie:
-    - Ordine stabile: token originali prima, sinonimi dopo (preserva priorità)
-    - Nessun duplicato (usa set interno)
-    - Nessun token < 3 chars, nessuna stopword aggiunta
-    - Zero latency (<0.1ms per 20 token), zero LLM calls
-    - Mai rilancia eccezioni
-    """
     try:
         seen: set[str] = set(tokens)
         expanded = list(tokens)
@@ -165,25 +131,18 @@ def _expand_tokens(tokens: list[str]) -> list[str]:
                         seen.add(syn)
                         expanded.append(syn)
         return expanded
-    except Exception:
+    except Exception as e:
+        _logger.debug("[context_manager] _expand_tokens failed: %s", e)
         return tokens
 
 
 def _split_camel_snake(text: str) -> list[str]:
-    """
-    Spezza camelCase/PascalCase/snake_case in token lowercase (min 3 chars).
-
-    Esempi:
-      "contextManager"    → ["context", "manager"]
-      "rank_files_by_relevance" → ["rank", "files", "relevance"]
-      "UnifiedAgentLoop"  → ["unified", "agent", "loop"]
-    Usato per fuzzy prefix bonus in rank_files_by_relevance.
-    """
     try:
         snake = _CAMEL_SPLIT_RE.sub(r'\1_\2', text)
         parts = re.split(r'[_\-./]', snake)
         return [p.lower() for p in parts if len(p) >= 3]
-    except Exception:
+    except Exception as e:
+        _logger.debug("[context_manager] _split_camel_snake failed: %s", e)
         return []
 
 
@@ -193,27 +152,6 @@ def rank_files_by_relevance(
     k: int = 5,
     min_score: float = 0.0,
 ) -> list[str]:
-    """
-    FIX-SKEL-RAG + FIX-SYN-EXPAND: Seleziona i top-K file più rilevanti per il goal.
-
-    Score composito (normalizzato su max(len(base_tokens), 1)):
-      path_hits    * 2.0  — keyword del goal (espansi) nel path
-      symbol_hits  * 1.5  — keyword nei nomi funzione/classe (skeleton RAG)
-      content_hits * 1.0  — keyword nei primi 600 chars del contenuto
-      prefix_bonus * 0.4  — goal token è prefisso di un split-token path/symbol (fuzzy)
-      entry_boost  +0.15  — file entry-point/config noti
-      lang_boost   +0.20  — il linguaggio del file è nel goal
-
-    FIX-SYN-EXPAND:
-    - I token del goal vengono espansi con sinonimi IT/EN prima del matching.
-    - Normalizzazione su len(base_tokens) originali (non espansi) per evitare score
-      inflazionati su file che matchano solo sinonimi lontani.
-    - "autenticazione" → matcha authGuard.ts, middleware.ts, jwt.ts anche senza
-      keyword nel path — copertura semantica senza embeddings.
-
-    Ritorna lista di path ordinata score-desc (top-K, score > min_score).
-    Mai rilancia eccezioni — fallback ai primi K file non ranked.
-    """
     if not all_files or not goal:
         return []
     try:
@@ -225,11 +163,8 @@ def rank_files_by_relevance(
         if not base_tokens:
             return [f.get('path', '') for f in all_files[:k] if f.get('path')]
 
-        # FIX-SYN-EXPAND: espandi con sinonimi tecnici IT/EN
         tokens = _expand_tokens(base_tokens)
-
         goal_lower = goal.lower()
-        # Normalizzatore: usa len(base_tokens) non len(tokens) per evitare score inflazionati
         n = max(len(base_tokens), 1)
         scores: list[tuple[float, str]] = []
 
@@ -243,49 +178,43 @@ def rank_files_by_relevance(
             path_lower    = path.lower()
             content_lower = content.lower()
 
-            # FIX-SKEL-RAG: estrai firme funzione/classe
             sigs = _extract_signatures(content, lang)
             symbols_lower = ' '.join(s.split(':', 1)[-1].lower() for s in sigs)
 
-            # Score primario — matching su token espansi
             path_hits    = sum(1 for t in tokens if t in path_lower)
             symbol_hits  = sum(1 for t in tokens if t in symbols_lower)
             content_hits = sum(1 for t in tokens if t in content_lower)
             score        = (path_hits * 2.0 + symbol_hits * 1.5 + content_hits) / n
 
-            # Fuzzy prefix bonus (su token base, non espansi — evita falsi positivi)
             filename_stem = re.sub(r'\.[^.]+$', '', path_lower.rsplit('/', 1)[-1])
             split_path    = _split_camel_snake(filename_stem)
             split_syms    = [t for s in sigs for t in _split_camel_snake(s.split(':', 1)[-1])]
             all_split     = split_path + split_syms
             prefix_hits   = sum(
-                1 for gt in base_tokens          # usa base_tokens: fuzzy su originali
+                1 for gt in base_tokens
                 for st in all_split
                 if st != gt and st.startswith(gt)
             )
             if prefix_hits:
                 score += (prefix_hits * 0.4) / n
 
-            # Entry-point boost
             if filename_stem in _RANK_ENTRY_STEMS:
                 score += 0.15
 
-            # Language boost
             if lang and lang in goal_lower:
                 score += 0.20
 
             if score > min_score:
                 scores.append((score, path))
 
-        # Ordinamento stabile: score desc, poi path asc
         scores.sort(key=lambda x: (-x[0], x[1]))
         return [p for _, p in scores[:k] if p]
-    except Exception:
+    except Exception as e:
+        _logger.error("[context_manager] rank_files_by_relevance critical failure: %s", e)
         return [f.get('path', '') for f in all_files[:k] if f.get('path')]
 
 
 def _extract_signatures(content: str, language: str) -> list[str]:
-    """Estrae nomi di funzioni/classi per lo skeleton."""
     try:
         lang = (language or '').lower()
         sigs: list[str] = []
@@ -302,12 +231,12 @@ def _extract_signatures(content: str, language: str) -> list[str]:
             for m in _PY_CLS_RE.finditer(content):
                 sigs.append(f'class:{m.group(1)}')
         return sigs[:15]
-    except Exception:
+    except Exception as e:
+        _logger.debug("[context_manager] _extract_signatures failed: %s", e)
         return []
 
 
 def build_file_skeleton(path: str, content: str, language: str) -> str:
-    """Costruisce una riga skeleton per un singolo file."""
     sigs       = _extract_signatures(content, language)
     line_count = content.count('\n') + 1
     sigs_str   = ', '.join(sigs[:8]) if sigs else '(no symbols)'
@@ -315,11 +244,6 @@ def build_file_skeleton(path: str, content: str, language: str) -> str:
 
 
 async def build_project_skeleton(files: list[dict[str, Any]]) -> str:
-    """
-    Costruisce lo skeleton compatto da una lista di file VFS.
-    Ogni dict ha: path, content, language.
-    Ritorna stringa multiriga per iniezione nel contesto agente.
-    """
     if not files:
         return ''
     try:
@@ -330,98 +254,6 @@ async def build_project_skeleton(files: list[dict[str, Any]]) -> str:
             language = f.get('language', '') or ''
             lines.append(build_file_skeleton(path, content, language))
         return '\n'.join(lines)
-    except Exception:
-        return ''
-
-
-async def compress_cold_file(path: str, content: str, language: str,
-                              tester_llm: Any | None = None) -> str:
-    """
-    Comprime un file 'cold' al suo riepilogo essenziale.
-    Usa Groq-8b via CONTEXT role per velocità. Fallback a skeleton.
-    Max 8s. Mai rilancia eccezioni.
-    """
-    # S573: hash() è PYTHONHASHSEED-salted → chiave diversa a ogni restart HF Space
-    # → nessun riuso della cache tra restart. hashlib.sha256 è stabile e deterministica.
-    _h = hashlib.sha256(content[:500].encode("utf-8", errors="replace")).hexdigest()[:16]
-    cache_key = f'{path}:{_h}'
-    if cache_key in _SUMMARY_CACHE:
-        return _SUMMARY_CACHE[cache_key]
-
-    skeleton = build_file_skeleton(path, content, language)
-
-    if tester_llm and len(content) > 500:
-        try:
-            msgs = [
-                {"role": "system", "content":
-                 "Riassumi il file in max 2 righe: scopo, symbols chiave, deps. "
-                 "Solo facts. Formato: [SCOPO] | [SYMBOLS] | [DEPS]"},
-                {"role": "user", "content":
-                 f"File: {path}\n```{language}\n{content[:2000]}\n```"},
-            ]
-            summary = await asyncio.wait_for(
-                # S587: 120→200 — formato [SCOPO]|[SYMBOLS]|[DEPS] può superare 120 tok
-                tester_llm.chat(msgs, temperature=0.0, max_tokens=200),
-                timeout=7.0,
-            )
-            if summary and not summary.startswith('[LLM'):
-                result = f'  {path}: {summary[:300]}'  # S604: 180→300 — summary file LLM spesso 2-3 righe
-                if len(_SUMMARY_CACHE) >= _MAX_SUMMARY_CACHE:
-                    oldest = next(iter(_SUMMARY_CACHE))
-                    del _SUMMARY_CACHE[oldest]
-                _SUMMARY_CACHE[cache_key] = result
-                return result
-        except Exception:
-            pass  # S364: fallback a skeleton
-
-    return skeleton
-
-
-async def get_context_for_goal(
-    goal: str,
-    active_files: list[str],
-    all_files: list[dict[str, Any]],
-    tester_llm: Any | None = None,
-    top_k: int = 5,
-) -> str:
-    """
-    Contesto intelligente per l'agente:
-    - File in active_files: full content (max 1500 chars ciascuno)
-    - Altri file: skeleton compatto
-    - Output max: ~4000 chars
-
-    S752-A + FIX-SKEL-RAG + FIX-SYN-EXPAND: se active_files è vuoto o None, usa
-    rank_files_by_relevance() (con synonym expansion) per selezionare i top_k file
-    più rilevanti per il goal. File con score == 0 esclusi automaticamente.
-    """
-    if not all_files:
-        return ''
-    try:
-        if not active_files and goal:
-            active_files = rank_files_by_relevance(goal, all_files, k=top_k)
-
-        active_set = set(active_files)
-        parts: list[str] = []
-        budget = 4000
-
-        for f in all_files:
-            path = f.get('path', '')
-            if path not in active_set:
-                continue
-            content  = (f.get('content', '') or '')[:1500]
-            language = f.get('language', '') or ''
-            chunk    = f'[ACTIVE FILE: {path}]\n```{language}\n{content}\n```'
-            parts.append(chunk)
-            budget -= len(chunk)
-            if budget <= 0:
-                break
-
-        cold_files = [f for f in all_files if f.get('path', '') not in active_set]
-        if cold_files and budget > 500:
-            skeleton = await build_project_skeleton(cold_files)
-            if skeleton:
-                parts.append(skeleton)
-
-        return '\n\n'.join(parts) if parts else ''
-    except Exception:
-        return ''
+    except Exception as e:
+        _logger.error("[context_manager] build_project_skeleton failed: %s", e)
+        return ""

agents/executor.py

@@ -25,14 +25,11 @@ _MIN_CALLS_FOR_CIRCUIT  = 3      # minimo di chiamate prima che il circuit possa
 _RECOVERY_INTERVAL      = 5      # ogni N chiamate con circuit open → tenta il tool primario
 
 # ─── S-ORCH-8GAP FIX-GAP2: Adaptive Timeout Tracker ─────────────────────────
-# Sliding window (last 5 durations) per tool — calcola P90 adattivo.
-# Strategia iPhone: rete variabile → se tool è stato lento di recente,
-# aumenta timeout; se è stato veloce, non sprecare tempo.
 class _AdaptiveTimeoutTracker:
     """Tracked P90 per-tool timeout con sliding window di 5 call."""
     _WINDOW     = 5
-    _MIN        = 4.0   # mai sotto 4s — tool veloci non vanno sotto
-    _MAX        = 55.0  # mai sopra 55s — iPhone connection timeout ~60s
+    _MIN        = 4.0   # mai sotto 4s
+    _MAX        = 55.0  # mai sopra 55s
     _MULTIPLIER = 1.5   # P90 * 1.5 = headroom conservativo
 
     def __init__(self) -> None:
@@ -44,10 +41,9 @@ class _AdaptiveTimeoutTracker:
         self._times[tool_name].append(elapsed)
 
     def adaptive_timeout(self, tool_name: str, base_timeout: float) -> float:
-        """Ritorna timeout adattivo: P90 * 1.5 se dati sufficienti, else base."""
         times = self._times.get(tool_name)
         if not times or len(times) < 2:
-            return base_timeout  # dati insufficienti → usa base invariato
+            return base_timeout
         sorted_t = sorted(times)
         p90_idx  = min(int(len(sorted_t) * 0.9), len(sorted_t) - 1)
         adaptive = sorted_t[p90_idx] * self._MULTIPLIER
@@ -56,17 +52,15 @@ class _AdaptiveTimeoutTracker:
 _timeout_tracker = _AdaptiveTimeoutTracker()
 
 
-# ─── Helper: ottieni session_id dal ContextVar (impostato da unified_loop.py) ─
 def _get_session_id() -> str:
     try:
         from tools.registry import _agent_session_id_var
         return _agent_session_id_var.get()
-    except Exception:
+    except Exception as e:
+        _logger.debug("[executor] _get_session_id failed: %s", e)
         return "default"
 
 
-# ─── Executor ────────────────────────────────────────────────────────────────
-
 class Executor:
     def __init__(
         self,
@@ -77,33 +71,21 @@ class Executor:
         self.llm         = llm_client or AIClient()
         self.memory      = memory
         self.max_retries = max_retries
-        # GAP-SKILL-SYNC v2: contatore chiamate per recovery credit (per-tool)
         self._circuit_recovery_counts: dict[str, int] = {}
 
-    # Backward-compat: vecchia firma aveva ollama=OllamaClient, memory=MemoryManager
     @classmethod
     def from_ollama(cls, ollama=None, memory=None, max_retries: int = 2) -> "Executor":
         return cls(memory=memory, max_retries=max_retries)
 
-    # ── Circuit breaker helper ────────────────────────────────────────────────
-
     def _is_circuit_open(self, tool_name: str, session_id: str) -> bool:
-        """True se il circuit breaker deve aprirsi per questo tool in questa sessione.
-
-        Condizioni (tutte necessarie):
-          1. Wilson score < CIRCUIT_OPEN_THRESHOLD (0.15)
-          2. >= MIN_CALLS_FOR_CIRCUIT (3) chiamate nella sessione
-          3. Il tool ha fallback disponibili in TOOL_REGISTRY
-        Recovery credit: ogni RECOVERY_INTERVAL chiamate, il circuit si chiude
-        temporaneamente per un tentativo di recovery.
-        """
         tool = TOOL_REGISTRY.get(tool_name, {})
         if not tool.get("fallbacks"):
-            return False  # senza fallback il circuit non può aprirsi
+            return False
         try:
             from agents.skill_tracker import get_skill_tracker
             stats = get_skill_tracker().get_stats(session_id).get(tool_name)
-        except Exception:
+        except Exception as e:
+            _logger.debug("[executor] _is_circuit_open failed to get skill tracker: %s", e)
             return False
         if not stats:
             return False
@@ -111,19 +93,13 @@ class Executor:
             return False
         if stats["wilson_score"] >= _CIRCUIT_OPEN_THRESHOLD:
             return False
-        # Recovery credit: conta le chiamate e apri una finestra ogni RECOVERY_INTERVAL
         count = self._circuit_recovery_counts.get(tool_name, 0) + 1
         self._circuit_recovery_counts[tool_name] = count
         if count % _RECOVERY_INTERVAL == 0:
-            _logger.info(
-                "[executor] recovery credit: riprovo %s (circuit call #%d)",
-                tool_name, count,
-            )
-            return False   # consenti un tentativo di recovery
+            _logger.info("[executor] recovery credit: riprovo %s (circuit call #%d)", tool_name, count)
+            return False
         return True
 
-    # ── Fallback execution ─────────────────────────────��──────────────────────
-
     async def _try_fallbacks(
         self,
         primary_name: str,
@@ -131,11 +107,6 @@ class Executor:
         timeout:      float,
         session_id:   str,
     ) -> "dict | None":
-        """Tenta i fallback definiti in TOOL_REGISTRY ordinati per Wilson score.
-
-        Registra ogni tentativo nel skill_tracker sotto il nome del fallback.
-        Ritorna il primo risultato con successo, o None se tutti falliscono.
-        """
         tool      = TOOL_REGISTRY.get(primary_name, {})
         fallbacks = tool.get("fallbacks", [])
         if not fallbacks:
@@ -144,28 +115,25 @@ class Executor:
         try:
             from agents.skill_tracker import get_skill_tracker
             sorted_fbs = get_skill_tracker().get_sorted_fallbacks(session_id, fallbacks)
-        except Exception:
-            sorted_fbs = fallbacks   # ordinamento originale come fallback del fallback
+        except Exception as e:
+            _logger.debug("[executor] _try_fallbacks failed to sort: %s", e)
+            sorted_fbs = fallbacks
 
         for fb_name in sorted_fbs:
             fb_tool = TOOL_REGISTRY.get(fb_name)
             if not fb_tool or not fb_tool.get("_fn"):
                 continue
-            _logger.info(
-                "[executor] %s fallita — provo fallback %s (Wilson-sorted)",
-                primary_name, fb_name,
-            )
+            _logger.info("[executor] %s fallita — provo fallback %s", primary_name, fb_name)
             try:
                 _t0     = _time_mod.monotonic()
                 _fb_to  = _timeout_tracker.adaptive_timeout(fb_name, timeout)
                 result  = await asyncio.wait_for(fb_tool["_fn"](**inputs), timeout=_fb_to)
                 _timeout_tracker.record(fb_name, _time_mod.monotonic() - _t0)
-                # Registra il successo del fallback nel skill_tracker
                 try:
                     from agents.skill_tracker import get_skill_tracker
                     get_skill_tracker().record(session_id, fb_name, True)
-                except Exception:
-                    pass
+                except Exception as e:
+                    _logger.debug("[executor] record success failed: %s", e)
                 return {
                     "success":            True,
                     "tool":               fb_name,
@@ -179,19 +147,17 @@ class Executor:
                 try:
                     from agents.skill_tracker import get_skill_tracker
                     get_skill_tracker().record(session_id, fb_name, False)
-                except Exception:
-                    pass
+                except Exception as e:
+                    _logger.debug("[executor] record timeout failed: %s", e)
             except Exception as fb_exc:
                 _logger.debug("[executor] fallback %s errore: %s", fb_name, str(fb_exc)[:80])
                 try:
                     from agents.skill_tracker import get_skill_tracker
                     get_skill_tracker().record(session_id, fb_name, False)
-                except Exception:
-                    pass
-
-        return None   # tutti i fallback hanno fallito
+                except Exception as e:
+                    _logger.debug("[executor] record error failed: %s", e)
 
-    # ── run_tool ─────────────────────────────────────────────────────────────
+        return None
 
     async def run_tool(self, tool_name: str, inputs: dict, timeout: float = 30.0) -> dict:
         tool = TOOL_REGISTRY.get(tool_name)
@@ -204,24 +170,13 @@ class Executor:
 
         session_id = _get_session_id()
 
-        # ── GAP-SKILL-SYNC v2: circuit breaker pre-check ──────────────────────
-        # Se il tool ha un Wilson score molto basso (< 0.15) con >= 3 dati in sessione,
-        # bypassa il tool e vai direttamente al miglior fallback disponibile.
         if self._is_circuit_open(tool_name, session_id):
-            _logger.info(
-                "[executor] circuit OPEN per %s — routing diretto a fallback (Wilson < %.2f)",
-                tool_name, _CIRCUIT_OPEN_THRESHOLD,
-            )
+            _logger.info("[executor] circuit OPEN per %s — routing a fallback", tool_name)
             fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
             if fb_result:
                 return fb_result
-            # Tutti i fallback falliti: procedi con il tool primario (ultima spiaggia)
-            _logger.warning(
-                "[executor] tutti i fallback di %s hanno fallito — provo comunque il tool primario",
-                tool_name,
-            )
+            _logger.warning("[executor] tutti i fallback di %s falliti — provo primario", tool_name)
 
-        # ── Esecuzione normale con retry ──────────────────────────────────────
         fn = tool.get("_fn")
         if fn is None:
             return {"success": False, "error": "Tool non ha funzione di esecuzione", "output": None}
@@ -229,31 +184,21 @@ class Executor:
         last_error: str = "max_retries"
         for attempt in range(self.max_retries + 1):
             try:
-                # S-ORCH-8GAP FIX-GAP2: usa timeout adattivo basato su P90 ultime 5 chiamate
                 _adaptive_to = _timeout_tracker.adaptive_timeout(tool_name, timeout)
                 _t0          = _time_mod.monotonic()
                 result       = await asyncio.wait_for(fn(**inputs), timeout=_adaptive_to)
                 _timeout_tracker.record(tool_name, _time_mod.monotonic() - _t0)
                 if self.memory:
-                    # S577→S600: inputs 100→500 — parity con altri handler
-                    await self.memory.save_episode(
-                        "tool",
-                        f"{tool_name}: {str(inputs)[:500]}",  # S589: 200→300→500
-                        str(result)[:500],
-                        True,
-                    )
+                    try:
+                        await self.memory.save_episode("tool", f"{tool_name}: {str(inputs)[:500]}", str(result)[:500], True)
+                    except Exception as e:
+                        _logger.debug("[executor] memory save failed: %s", e)
                 return {"success": True, "tool": tool_name, "output": result, "attempt": attempt + 1}
 
             except asyncio.TimeoutError:
-                # FIX-GAP2: registra il timeout come durata massima per shrink futuro
                 _timeout_tracker.record(tool_name, timeout * 1.2)
                 last_error = f"Timeout dopo {timeout}s (tentativo {attempt + 1})"
                 if attempt == self.max_retries:
-                    # Ultima chance: prova i fallback ordinati per Wilson score
-                    _logger.info(
-                        "[executor] %s timeout definitivo — provo fallback Wilson-sorted",
-                        tool_name,
-                    )
                     fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
                     if fb_result:
                         return fb_result
@@ -263,11 +208,6 @@ class Executor:
             except Exception as e:
                 last_error = str(e)
                 if attempt == self.max_retries:
-                    # Ultima chance: prova i fallback ordinati per Wilson score
-                    _logger.info(
-                        "[executor] %s errore definitivo (%s) — provo fallback Wilson-sorted",
-                        tool_name, last_error[:60],
-                    )
                     fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
                     if fb_result:
                         return fb_result

agents/unified_loop.py

@@ -105,10 +105,11 @@ async def _read_bb_upstash(session_id: str) -> str:
                             _key  = _e.get("key", "")
                             _val  = str(_e.get("value", ""))[:200]
                             _out.append(f"- [{_agid}] {_key}: {_val}")
-                    except Exception:
-                        pass
+                    except Exception as _d_err:
+                        _logger.debug("[unified_loop] delegate parse silenced: %s", type(_d_err).__name__)
             return ("SCOPERTE CRITICHE DAI DELEGATI:\n" + "\n".join(_out)) if _out else ""
-    except Exception:
+    except Exception as _d_outer_err:
+        _logger.debug("[unified_loop] _get_delegate_findings silenced: %s", type(_d_outer_err).__name__)
         return ""
 
 # ── Nuovi mixin estratti (split 2026-06-30) ──────────────────────────────────
@@ -196,7 +197,8 @@ class UnifiedAgentLoop(
         try:
             from tools.registry import _agent_session_id_var as _sid_var
             _sid_token = _sid_var.set(self._run_task_id)
-        except Exception:
+        except Exception as _imp_err:
+            _logger.debug("[unified_loop] sid_var import silenced: %s", type(_imp_err).__name__)
             _sid_token = None  # fallback silente — registry usa default "agent_default"
 
         # S750-GAP-B: pre-warm sandbox backend-exec — POST /api/session in background.
@@ -269,8 +271,8 @@ class UnifiedAgentLoop(
                         "explanation": _bl_reason[:200],
                     }))
                 # Fail-open: logghiamo e proseguiamo — non blocchiamo task legittimi
-        except Exception:
-            pass  # decision_memory non disponibile — continua normalmente
+        except Exception as _dm_err:
+            _logger.debug("[unified_loop] decision_memory check silenced: %s", type(_dm_err).__name__)  # non disponibile — continua normalmente
 
         # P29-B1: gate ambiguità strutturale — _is_goal_ambiguous() era P28-B2 dead code (mai chiamata).
         # Zero LLM, <0.1ms. Lingua-aware via self._run_lang (P28-B1). Fires dopo blacklist e prima del routing.
@@ -314,7 +316,7 @@ class UnifiedAgentLoop(
             _r_amb = {"answer": _amb_answer, "timing_ms": 0, "effective_max_steps": state.max_steps}
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             return _r_amb
 
         # P29-R1: borderline ambiguity gate — goal con verbo ma oggetto pronominale vago.
@@ -431,7 +433,7 @@ class UnifiedAgentLoop(
             _r_bl = {"answer": _bl_answer, "timing_ms": 0, "effective_max_steps": state.max_steps}
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             return _r_bl
 
 
@@ -451,7 +453,7 @@ class UnifiedAgentLoop(
             # S749-D: reset ContextVar
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             # GAP-NEW-4: schedule VFS backup se ci sono file scritti nella sessione
             if self._session_files:
                 asyncio.ensure_future(self._vfs_git_backup())
@@ -475,7 +477,7 @@ class UnifiedAgentLoop(
             # S749-D: reset ContextVar
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             # GAP-NEW-4: schedule VFS backup se ci sono file scritti nella sessione
             if self._session_files:
                 asyncio.ensure_future(self._vfs_git_backup())
@@ -492,14 +494,14 @@ class UnifiedAgentLoop(
             try:
                 from api.state import record_timing as _rtcB5
                 _rtcB5("classify_ms", (_time.monotonic() - _t0_classify) * 1000)
-            except Exception:
-                pass
+            except Exception as _rt_err:
+                _logger.debug("[unified_loop] record_timing silenced: %s", type(_rt_err).__name__)
             _r = await self._run_fallback(state, on_step)
             _r["timing_ms"] = int((_time.monotonic() - _t_run) * 1000)
             _r["effective_max_steps"] = state.max_steps
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             if self._session_files:
                 asyncio.ensure_future(self._vfs_git_backup())
             return _r
@@ -555,12 +557,12 @@ class UnifiedAgentLoop(
                     try:
                         from api.state import increment_stat as _p36_stat
                         _p36_stat("p36_fast_path_hit")
-                    except Exception:
-                        pass
+                    except Exception as _is_err:
+                        _logger.debug("[unified_loop] increment_stat P36 silenced: %s", type(_is_err).__name__)
                     _logger.info("P36 fast-path: python_analyze in %dms", _p36_ms)
                     if _sid_token is not None:
                         try: _sid_var.reset(_sid_token)
-                        except Exception: pass
+                        except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
                     return {
                         "success": True,
                         "answer": _p36_answer,
@@ -605,7 +607,7 @@ class UnifiedAgentLoop(
             # S749-D: reset ContextVar
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             # GAP-NEW-4: schedule VFS backup se ci sono file scritti nella sessione
             if self._session_files:
                 asyncio.ensure_future(self._vfs_git_backup())
@@ -648,7 +650,7 @@ class UnifiedAgentLoop(
             # S749-D: reset ContextVar
             if _sid_token is not None:
                 try: _sid_var.reset(_sid_token)
-                except Exception: pass
+                except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)
             # GAP-NEW-4: schedule VFS backup se ci sono file scritti nella sessione
             if self._session_files:
                 asyncio.ensure_future(self._vfs_git_backup())
@@ -683,4 +685,4 @@ class UnifiedAgentLoop(
         # S749-D: reset ContextVar prima di uscire — libera la sandbox per il GC
         if _sid_token is not None:
             try: _sid_var.reset(_sid_token)
-            except Exception: pass
+            except Exception as _sv_err: _logger.debug("[unified_loop] sid_var.reset silenced: %s", type(_sv_err).__name__)

api/_agent_helpers.py

@@ -42,7 +42,6 @@ except Exception:
     async def _tg_start(*_a, **_kw): pass  # type: ignore[misc]
     async def _tg_step(*_a, **_kw):  pass  # type: ignore[misc]
 
-from fastapi import APIRouter
 router = APIRouter()
 
 @router.post('/run_loop', deprecated=True)

api/agent_checkpoint_routes.py

@@ -49,9 +49,6 @@ except Exception:
 
 router = APIRouter()
 
-            'Connection':        'keep-alive',
-        },
-    )
 
 
 # ── Task checkpoints ───────────────────────────────────────────────────────────
@@ -279,3 +276,4 @@ async def get_circuit_status(session_id: str):
             'circuits_closed':     [],
             'error':               str(exc),
         }
+

api/agent_loop_routes.py

@@ -12,7 +12,8 @@ Route coperte:
 from __future__ import annotations
 import os, asyncio, json, uuid, time, re
 import re as _re_persona
-from fastapi import APIRouter, HTTPException, Request, Body
+from fastapi import APIRouter
+router = APIRouter(), HTTPException, Request, Body
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, field_validator
 from typing import Literal
@@ -26,7 +27,8 @@ from .state import (
 from .speculative import fire_speculative_tools
 try:
     from .quality_guardian import run_quality_check as _run_quality_check
-except Exception:
+except Exception as _qg_err:
+    import logging as _qg_log; _qg_log.getLogger(__name__).warning("[routes] quality_guardian import failed: %s", _qg_err)
     _run_quality_check = None
 import logging
 _logger = logging.getLogger("api.agent")
@@ -55,8 +57,6 @@ async def run_loop_removed():
 
 # ── SSE run-stream ────────────────────────────────────────────────────────────
 
-router = APIRouter()
-
 @router.post('/api/agent/run-stream')
 async def agent_run_stream(body: ReasonLoopIn, request: Request):
     # S-BENCH: auth guard — consistente con /api/exec e /api/execute-shell
@@ -79,7 +79,8 @@ async def agent_run_stream(body: ReasonLoopIn, request: Request):
                     from agents.response_verifier import ResponseVerifier
                     _critic   = Critic(llm_client=client)
                     _verifier = ResponseVerifier()
-                except Exception:
+                except Exception as _cv_err:
+                    _logger.warning("[routes] Critic/Verifier init failed: %s", _cv_err)
                     _critic   = None
                     _verifier = None
                 # Resume automatico: inietta contesto checkpoint se disponibile (Case 2.5 fall-through)
@@ -147,8 +148,8 @@ async def agent_run_stream(body: ReasonLoopIn, request: Request):
                 yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id, 'abort_reason': 'system', 'abort_source': 'no_providers', 'error': f'Nessun provider AI disponibile ({_names})'})}\n\n"  # MX18-ABORT: no providers → system abort
                 yield "data: [DONE]\n\n"
                 return
-        except Exception:
-            pass  # se heartbeat non è inizializzato, prosegui normalmente
+        except Exception as _hb_err:
+            _logger.debug("[routes] heartbeat skip silenced: %s", type(_hb_err).__name__)  # non inizializzato, prosegui normalmente
 
         # S386: timeout ridotto 120→60s — risposta entro 1 minuto o errore esplicito
         timeout_secs    = float(os.getenv('AGENT_STREAM_TIMEOUT', '60'))
@@ -295,7 +296,8 @@ async def reason_loop(body: ReasonLoopIn):
             from agents.response_verifier import ResponseVerifier
             _critic   = Critic(llm_client=client)
             _verifier = ResponseVerifier()
-        except Exception:
+        except Exception as _cv_err:
+            _logger.warning("[routes] Critic/Verifier init failed: %s", _cv_err)
             _critic   = None
             _verifier = None
         loop = UnifiedAgentLoop(

api/hf_storage.py

@@ -18,6 +18,7 @@ import httpx
 import asyncio
 import logging
 from typing import Any, Optional
+from .state import get_env_secret
 
 _logger = logging.getLogger("api.hf_storage")
 
@@ -91,3 +92,4 @@ def hf_fire_and_forget(
             loop.create_task(hf_append_record(dataset_path, record, repo_id))
     except Exception:
         pass  # Fire-and-forget: mai propagare eccezioni al caller
+

api/job_queue.py

@@ -3,6 +3,7 @@ import json
 import time
 import asyncio
 import logging
+import uuid
 from typing import Optional, List, Dict, Any
 from fastapi import APIRouter, HTTPException, Request
 from pydantic import BaseModel
@@ -69,8 +70,8 @@ async def publish_load_metrics():
     data = {
         "role": _SPACE_ROLE,
         "ts": int(time.time() * 1000),
-        "active_tasks": len(asyncio.all_tasks()) - 5, # Stima approssimativa
-        "cpu": 0, # Placeholder per metrics reali
+        "active_tasks": len(asyncio.all_tasks()) - 5,  # Stima approssimativa
+        "cpu": 0,  # Placeholder per metrics reali
         "mem": 0
     }
     await _rcmd(["SET", _K_LOAD(_SPACE_ROLE), json.dumps(data), "EX", "60"])
@@ -85,31 +86,154 @@ async def _load_publisher_loop():
         await asyncio.sleep(30)
 
 async def _hands_consumer_loop():
-    """Loop consumer per nodi HANDS, MEMORY e AUDIT."""
+    """Loop consumer reale per nodi distribuiti (S42)."""
     _logger.info("[jq] consumer loop avviato per ruolo: %s", _SPACE_ROLE)
+    _consecutive_errors = 0
+    _MAX_CONSECUTIVE_ERRORS = 3  # INV-R1: dopo 3 errori infra → pausa 30s
+
     while True:
         try:
-            # S429: Tool Success Contract — verifica heartbeat consumer
+            # 1. Heartbeat consumer
             await _rcmd(["SET", _K_CONSUMER, "1", "EX", "15"])
-            
-            # TODO: Implementazione effettiva del prelievo job e esecuzione
-            # Per ora è un segnaposto per la struttura UltraVSS v7.0
-            
+
+            # 2. Prelievo job (RPOP per coda FIFO)
+            res = await _rcmd(["RPOP", _K_PENDING])
+            if res and res.get("result"):
+                job_raw = res["result"]
+                try:
+                    job_data = json.loads(job_raw)
+                except json.JSONDecodeError as _jd_err:
+                    _logger.error("[jq] JSON malformato in coda: %s | raw: %.120s", _jd_err, job_raw)
+                    continue
+                task_id = job_data.get("taskId", str(uuid.uuid4()))
+                goal = job_data.get("goal", "")
+                context_raw = job_data.get("context", {})
+                _logger.info("[jq] job ricevuto: %s | goal: %.80s", task_id, goal)
+
+                # Normalizza context → stringa
+                if isinstance(context_raw, dict):
+                    context_str = "\n".join(f"{k}: {v}" for k, v in context_raw.items() if v)
+                elif isinstance(context_raw, str):
+                    context_str = context_raw
+                else:
+                    context_str = ""
+
+                # S429 — Esecuzione REALE via UnifiedAgentLoop
+                result_payload: Dict[str, Any] = {}
+                try:
+                    from agents.unified_loop import UnifiedAgentLoop
+                    from .state import (
+                        _get_ai_client, _get_mem_manager_async,
+                        _get_executor, _get_planner,
+                    )
+                    ai_client = _get_ai_client()
+                    try:
+                        from agents.critic import Critic
+                        from agents.response_verifier import ResponseVerifier
+                        _critic   = Critic(llm_client=ai_client)
+                        _verifier = ResponseVerifier()
+                    except Exception:
+                        _critic   = None
+                        _verifier = None
+
+                    # Accumula step per tracciabilità nel risultato Redis
+                    _steps: list = []
+
+                    async def _on_step(step: dict) -> None:
+                        _steps.append({
+                            "action": step.get("action", ""),
+                            "output": str(step.get("output", ""))[:200],
+                        })
+
+                    loop_inst = UnifiedAgentLoop(
+                        llm_client=ai_client,
+                        critic=_critic,
+                        verifier=_verifier,
+                        memory=await _get_mem_manager_async(),
+                        executor=_get_executor(),
+                        planner=_get_planner(),
+                    )
+
+                    # max_steps: clampato 5–16, scalato per priorità job
+                    _priority = int(job_data.get("priority", 1))
+                    _max_steps = max(5, min(4 + _priority * 2, 16))
+
+                    raw_result = await loop_inst.run(
+                        goal=goal,
+                        context=context_str,
+                        max_steps=_max_steps,
+                        on_step=_on_step,
+                        session_id=task_id,
+                    )
+
+                    output  = raw_result.get("output", "") if isinstance(raw_result, dict) else str(raw_result)
+                    success = bool(raw_result.get("success", False)) if isinstance(raw_result, dict) else bool(output)
+                    engine  = raw_result.get("engine", "unknown") if isinstance(raw_result, dict) else "unknown"
+
+                    result_payload = {
+                        "taskId":  task_id,
+                        "status":  "completed" if success else "failed",
+                        "worker":  _SPACE_ROLE,
+                        "output":  output[:4000],
+                        "engine":  engine,
+                        "success": success,
+                        "steps":   len(_steps),
+                        "ts":      int(time.time() * 1000),
+                    }
+                    _consecutive_errors = 0
+                    _logger.info("[jq] job completato: %s | engine: %s | success: %s", task_id, engine, success)
+
+                except (ImportError, ModuleNotFoundError) as imp_err:
+                    # UnifiedAgentLoop non disponibile su questo nodo
+                    _logger.warning("[jq] agents.unified_loop non disponibile su %s: %s", _SPACE_ROLE, imp_err)
+                    result_payload = {
+                        "taskId": task_id,
+                        "status": "unavailable",
+                        "worker": _SPACE_ROLE,
+                        "error":  f"UnifiedAgentLoop non disponibile su nodo {_SPACE_ROLE}: {imp_err}",
+                        "ts":     int(time.time() * 1000),
+                    }
+
+                except Exception as exec_err:
+                    _consecutive_errors += 1
+                    _logger.error("[jq] job execution error task=%s: %s", task_id, exec_err)
+                    result_payload = {
+                        "taskId": task_id,
+                        "status": "error",
+                        "worker": _SPACE_ROLE,
+                        "error":  str(exec_err)[:500],
+                        "ts":     int(time.time() * 1000),
+                    }
+                    # INV-R1: 3 errori consecutivi infra → pausa 30s, log esplicito
+                    if _consecutive_errors >= _MAX_CONSECUTIVE_ERRORS:
+                        _logger.error(
+                            "[jq] INV-R1: %d errori consecutivi — pausa 30s (nodo: %s)",
+                            _consecutive_errors, _SPACE_ROLE,
+                        )
+                        await asyncio.sleep(30)
+
+                # 3. Salva risultato su Redis (TTL 1h) — Tool Success Contract S429
+                await _rcmd(["SET", _K_RESULT(task_id), json.dumps(result_payload), "EX", "3600"])
+                _logger.info("[jq] risultato Redis: %s → %s", task_id, result_payload.get("status"))
+
+            else:
+                # Coda vuota → reset errori consecutivi, polling leggero via REST
+                _consecutive_errors = 0
+                await asyncio.sleep(2)
+
         except Exception as e:
-            _logger.debug("[jq] consumer loop error: %s", e)
-        await asyncio.sleep(5)
+            _logger.error("[jq] consumer loop error: %s", e)
+            await asyncio.sleep(5)
 
 async def start_job_queue_consumer() -> None:
     """Punto di ingresso per main.py _on_startup()."""
     if not _redis_ok():
         _logger.warning("[jq] Redis non configurato — job queue disabilitato")
         return
-
-    asyncio.create_task(_load_publisher_loop())
-
+    _bg_tasks.append(asyncio.create_task(_load_publisher_loop()))
     # Grid Orchestrator: Consumer specializzati (HANDS, MEMORY, AUDIT)
     if _SPACE_ROLE in ("hands", "memory", "audit", "unknown"):
-        asyncio.create_task(_hands_consumer_loop())
+        _bg_tasks.append(asyncio.create_task(_hands_consumer_loop()))
     else:
         _logger.info("[jq] SPACE_ROLE=%s — consumer non avviato (solo load publisher)", _SPACE_ROLE)
 
@@ -131,11 +255,71 @@ async def jq_load(role: str):
     res = await _rcmd(["GET", _K_LOAD(role)])
     if not res or not res.get("result"):
         raise HTTPException(404, f"Metriche {role} non disponibili")
-    return json.loads(res["result"])
+    try:
+        return json.loads(res["result"])
+    except json.JSONDecodeError as _je:
+        raise HTTPException(500, f"Metriche Redis corrotte per {role}: {_je}")
 
 @router.post("/submit")
 async def jq_submit(job: JobPayload, request: Request):
+    """
+    S42 — Grid Orchestrator: sottomissione job reale su Redis.
+    Inserisce il job in coda LPUSH (FIFO con RPOP dal consumer).
+    Restituisce taskId e lunghezza coda corrente.
+    """
     if _INTERNAL_TOKEN and request.headers.get("X-Internal-Token") != _INTERNAL_TOKEN:
         raise HTTPException(401, "Unauthorized")
-    # Logica di sottomissione job...
-    return {"taskId": job.taskId, "status": "queued"}
+
+    if not _redis_ok():
+        raise HTTPException(503, "Job queue non disponibile: Redis non configurato")
+
+    # Normalizza taskId: usa quello fornito o genera uno nuovo
+    task_id = job.taskId if job.taskId else str(uuid.uuid4())
+
+    job_payload = {
+        "taskId": task_id,
+        "goal": job.goal,
+        "context": job.context or {},
+        "priority": job.priority,
+        "submittedAt": int(time.time() * 1000),
+        "submittedBy": _SPACE_ROLE,
+    }
+
+    # LPUSH → consumer usa RPOP (FIFO)
+    res = await _rcmd(["LPUSH", _K_PENDING, json.dumps(job_payload)])
+    if res is None:
+        raise HTTPException(503, "Errore Redis durante la sottomissione del job")
+
+    queue_len = int(res.get("result", 0)) if res else 0
+
+    # Segnale di wake per consumer (TTL breve)
+    await _rcmd(["SET", _K_WAKE, "1", "EX", "10"])
+
+    _logger.info("[jq] job sottomesso: %s (coda: %d)", task_id, queue_len)
+    return {
+        "taskId": task_id,
+        "status": "queued",
+        "queueLength": queue_len,
+        "ts": int(time.time() * 1000),
+    }
+
+@router.get("/result/{task_id}")
+async def jq_result(task_id: str, request: Request):
+    """
+    S429 — Tool Success Contract: recupero risultato job per taskId.
+    Restituisce il risultato se disponibile, altrimenti pending.
+    """
+    if _INTERNAL_TOKEN and request.headers.get("X-Internal-Token") != _INTERNAL_TOKEN:
+        raise HTTPException(401, "Unauthorized")
+
+    if not _redis_ok():
+        raise HTTPException(503, "Job queue non disponibile: Redis non configurato")
+
+    res = await _rcmd(["GET", _K_RESULT(task_id)])
+    if not res or not res.get("result"):
+        return {"taskId": task_id, "status": "pending", "ts": int(time.time() * 1000)}
+    try:
+        return json.loads(res["result"])
+    except json.JSONDecodeError as _je:
+        _logger.error("[jq] risultato Redis corrotto task=%s: %s", task_id, _je)
+        raise HTTPException(500, f"Risultato corrotto in Redis per {task_id}: {_je}")

api/providers.py

@@ -1,7 +1,7 @@
 """backend/api/providers.py — Health, tools, status, AI health, heartbeat (S354)."""
 import os, asyncio, time, logging
 from fastapi import APIRouter, Request
-from .state import _sb, SENSITIVE, _ai_health_cache, _AI_HEALTH_TTL, _heartbeat_state, _TIMING_STORE, _REPAIR_STATS
+from .state import _sb, SENSITIVE, _ai_health_cache, _AI_HEALTH_TTL, _heartbeat_state, _TIMING_STORE, _REPAIR_STATS, get_env_secret
 
 router = APIRouter()
 _logger    = logging.getLogger('agente_ai')
@@ -151,7 +151,7 @@ async def list_tools():
 @router.get('/api/status')
 async def status(request: Request):
     # security-fix: richiede X-Internal-Token — endpoint espone env vars
-    _tok = os.getenv('INTERNAL_TOKEN', '')
+    _tok = get_env_secret('INTERNAL_TOKEN')
     if _tok and request.headers.get('X-Internal-Token', '') != _tok:
         from fastapi import HTTPException as _HTTPEx
         raise _HTTPEx(401, 'Unauthorized')
@@ -569,7 +569,7 @@ async def auth_ping(
       (CF Worker aggiunge il token → role MACHINE se i due token coincidono)
     """
     import os as _os
-    server_token = _os.getenv('INTERNAL_TOKEN', '').strip()
+    server_token = _get_env_secret('INTERNAL_TOKEN').strip()
 
     # Leggi header sia dal parametro sia dall'oggetto request (FastAPI può passare entrambi)
     hdr_token = x_internal_token
@@ -611,3 +611,4 @@ async def status_ping():
         'supabase': _sb is not None,
         'backend': 'HuggingFace Spaces / Railway',
     }
+

api/state.py

@@ -5,6 +5,10 @@ Contains: Supabase client, in-memory stores, singleton getters, shared Pydantic
 TTL constants, prune helpers. Extracted from main.py — zero behaviour change.
 """
 import os, time, asyncio as _asyncio_mod, json as _json, re as _re
+
+def get_env_secret(name: str, default: str = "") -> str:
+    """Getter resiliente: prova SECRET_name prima di name per evitare collisioni HF."""
+    return os.getenv(f"SECRET_{name}") or os.getenv(name) or default
 from typing import Optional, Any
 from fastapi import HTTPException
 from pydantic import BaseModel, field_validator, model_validator
@@ -31,8 +35,8 @@ def safe_json_dumps(obj: object, *, ensure_ascii: bool = False, **kw) -> str:
 _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', '')
+    _SUPA_URL = get_env_secret('SUPABASE_URL')
+    _SUPA_KEY = get_env_secret('SUPABASE_KEY') or get_env_secret('SUPABASE_ANON_KEY')
     if _SUPA_URL and _SUPA_KEY:
         from supabase import create_client
         _sb = create_client(_SUPA_URL, _SUPA_KEY)
@@ -150,7 +154,11 @@ async def restore_agent_tasks_from_snap() -> int:
         )
         if not res or not res.data:
             return 0
-        snap: dict = _json.loads(res.data["value"])
+        try:
+            snap: dict = _json.loads(res.data["value"])
+        except _json.JSONDecodeError as _snap_jd:
+            _logger.warning("BOOT: GAP-STATE snapshot JSON corrotto: %s", _snap_jd)
+            return 0
         n = 0
         for task_id, data in snap.items():
             if task_id not in _agent_tasks and isinstance(data, dict):
@@ -343,13 +351,15 @@ async def _get_mem_manager_async() -> Any:
                 _mem_manager = MemoryManager()
                 await _mem_manager.init()
                 _mem_manager_inited = True
-            except Exception:
+            except Exception as _mm_err:
+                _logger.warning("[state] MemoryManager.init failed (1st): %s", _mm_err)
                 _mem_manager = None
         elif not _mem_manager_inited:
             try:
                 await _mem_manager.init()
                 _mem_manager_inited = True
-            except Exception:
+            except Exception as _mm2_err:
+                _logger.warning("[state] MemoryManager.init failed (2nd, no retry): %s", _mm2_err)
                 _mem_manager_inited = True  # evita retry infiniti
     return _mem_manager
 
@@ -387,7 +397,8 @@ def _get_mem_manager() -> Any:
             # 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:
+    except Exception as _mm3_err:
+        _logger.warning("[state] MemoryManager get failed: %s", _mm3_err)
         _mem_manager = None
     return _mem_manager
 
@@ -403,7 +414,8 @@ def _get_executor() -> Any:
     try:
         from agents.executor import Executor
         _executor = Executor(memory=_get_mem_manager())
-    except Exception:
+    except Exception as _ex_err:
+        _logger.warning("[state] Executor init failed: %s", _ex_err)
         _executor = None
     return _executor
 
@@ -576,3 +588,4 @@ class AgentTaskIn(BaseModel):
         if not isinstance(v, list):
             return []
         return [str(h)[:300] for h in v[:5]]  # S606: 200→300
+

api/state_sync.py

@@ -1,32 +1,49 @@
-"""
-api/state_sync.py — S901: Real-Time State Synchronizer (UltraVSS Beta Strategy)
-
-Router FastAPI per la sincronizzazione dello stato tramite WebSocket.
-"""
-
+import os, json, asyncio, httpx, time
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 import logging
 
 _logger = logging.getLogger("api.state_sync")
 router = APIRouter(prefix="/api/sync", tags=["sync"])
 
+# ── Helper Redis (Upstash REST) ──────────────────────────────────────────────
+async def _rcmd(cmd: list) -> Optional[dict]:
+    url = os.getenv("UPSTASH_REDIS_REST_URL")
+    tok = os.getenv("UPSTASH_REDIS_REST_TOKEN")
+    if not url or not tok: return None
+    try:
+        async with httpx.AsyncClient() as client:
+            r = await client.post(url, headers={"Authorization": f"Bearer {tok}"}, json=cmd, timeout=5.0)
+            return r.json()
+    except Exception as e:
+        _logger.error("[sync] redis error: %s", e)
+        return None
+
+# ── Polling interval per cross-node sync (secondi) ──────────────────────────
+_PUBSUB_POLL_INTERVAL = float(os.getenv("SYNC_POLL_INTERVAL", "3"))
+
 class ConnectionManager:
     def __init__(self):
         self.active_connections: Dict[str, List[WebSocket]] = {}
         self.room_states: Dict[str, Any] = {}
+        # Cursore per polling: ultimo timestamp letto per ogni room
+        self._last_seen: Dict[str, int] = {}
+        self._pubsub_task: Optional[asyncio.Task] = None
 
     async def connect(self, websocket: WebSocket, room_id: str):
         await websocket.accept()
         if room_id not in self.active_connections:
             self.active_connections[room_id] = []
         self.active_connections[room_id].append(websocket)
-        
-        if room_id in self.room_states:
-            await websocket.send_json({
-                "type": "SYNC_STATE",
-                "payload": self.room_states[room_id]
-            })
+
+        # S901: Recupero stato globale da Redis
+        res = await _rcmd(["GET", f"sync:state:{room_id}"])
+        if res and res.get("result"):
+            try:
+                state = json.loads(res["result"])
+            except json.JSONDecodeError:
+                state = {}
+            await websocket.send_json({"type": "SYNC_STATE", "payload": state})
 
     def disconnect(self, websocket: WebSocket, room_id: str):
         if room_id in self.active_connections:
@@ -35,25 +52,96 @@ class ConnectionManager:
             if not self.active_connections[room_id]:
                 del self.active_connections[room_id]
 
-    async def broadcast(self, message: dict, room_id: str, exclude: WebSocket = None):
+    async def broadcast(self, message: dict, room_id: str, exclude: WebSocket = None, remote: bool = False):
+        """
+        Broadcast locale e remoto (via Redis).
+        remote=True indica che il messaggio arriva da Pub/Sub, quindi non ri-pubblicare.
+        """
         if room_id not in self.active_connections:
             return
-        
+
+        # Aggiornamento stato locale/globale
         if message.get("type") == "UPDATE_STATE":
             self.room_states[room_id] = message.get("payload")
+            if not remote:
+                # Persistenza stato su Redis (TTL 1h)
+                await _rcmd(["SET", f"sync:state:{room_id}", json.dumps(message.get("payload")), "EX", "3600"])
 
+        # Pubblicazione su Redis per altri nodi (lista circolare con timestamp)
+        if not remote:
+            envelope = {
+                "ts": int(time.time() * 1000),
+                "room": room_id,
+                "msg": message,
+            }
+            await _rcmd(["LPUSH", f"sync:events:{room_id}", json.dumps(envelope)])
+            # Mantieni solo gli ultimi 50 eventi per room (evita memory leak Redis)
+            await _rcmd(["LTRIM", f"sync:events:{room_id}", "0", "49"])
+
+        # Invio ai client connessi a QUESTA istanza
         for connection in self.active_connections[room_id]:
             if connection != exclude:
                 try:
                     await connection.send_json(message)
-                except Exception as e:
-                    _logger.error(f"Error broadcasting to client in room {room_id}: {e}")
+                except Exception as _sj_err:
+                    _logger.debug("[sync] send_json silenced (WS dead?): %s", type(_sj_err).__name__)
+
+    async def start_pubsub_listener(self):
+        """
+        S901 Fix: Polling leggero su Redis per sincronizzazione cross-nodo.
+        Upstash REST non supporta SUBSCRIBE persistente; usiamo LRANGE + cursore timestamp.
+        Intervallo configurabile via SYNC_POLL_INTERVAL (default: 3s).
+        """
+        _logger.info("[sync] pubsub polling listener avviato (interval=%.1fs)", _PUBSUB_POLL_INTERVAL)
+        while True:
+            try:
+                # Itera su tutte le room attive
+                for room_id in list(self.active_connections.keys()):
+                    if not self.active_connections.get(room_id):
+                        continue
+
+                    res = await _rcmd(["LRANGE", f"sync:events:{room_id}", "0", "9"])
+                    if not res or not res.get("result"):
+                        continue
+
+                    events: list = res["result"]
+                    last_seen = self._last_seen.get(room_id, 0)
+                    new_events = []
+
+                    for raw in events:
+                        try:
+                            envelope = json.loads(raw)
+                            ts = envelope.get("ts", 0)
+                            if ts > last_seen:
+                                new_events.append(envelope)
+                        except Exception:
+                            pass
+
+                    if new_events:
+                        # Ordina per timestamp crescente
+                        new_events.sort(key=lambda e: e.get("ts", 0))
+                        for envelope in new_events:
+                            msg = envelope.get("msg", {})
+                            # Broadcast locale senza ri-pubblicare su Redis (remote=True)
+                            await self.broadcast(msg, room_id, remote=True)
+                        # Aggiorna cursore
+                        self._last_seen[room_id] = new_events[-1].get("ts", last_seen)
+
+            except Exception as e:
+                _logger.error("[sync] pubsub polling error: %s", e)
+
+            await asyncio.sleep(_PUBSUB_POLL_INTERVAL)
 
 manager = ConnectionManager()
 
 @router.websocket("/ws/{room_id}")
 async def websocket_endpoint(websocket: WebSocket, room_id: str):
     await manager.connect(websocket, room_id)
+
+    # Avvia il listener di polling se non già attivo
+    if manager._pubsub_task is None or manager._pubsub_task.done():
+        manager._pubsub_task = asyncio.create_task(manager.start_pubsub_listener())
+
     try:
         while True:
             data = await websocket.receive_json()

main.py

@@ -24,7 +24,8 @@ from fastapi.responses import JSONResponse
 
 # Gap 2.2: Structured JSON logging — setup PRIMA di qualsiasi altro import
 from api.structured_log import setup_structured_logging as _setup_structured_log, router as _logs_router
-from api.integrity_manager import router as _integrity_router  # P41
+from api.integrity_manager import router as _integrity_router
+from api.state import get_env_secret  # P41
 _setup_structured_log()
 import logging as _boot_logger; _boot_logger.getLogger('agente_ai').info('BOOT: importing FastAPI...')
 
@@ -33,7 +34,7 @@ _logger = logging.getLogger('agente_ai')
 
 # S274-SEC3: INTERNAL_TOKEN — genera casuale al boot se non configurato.
 _GENERATED_TOKEN = _secrets_mod.token_hex(32)
-if not os.getenv('INTERNAL_TOKEN'):
+if not get_env_secret('INTERNAL_TOKEN'):
     os.environ['INTERNAL_TOKEN'] = _GENERATED_TOKEN
     _logger.critical('BOOT: INTERNAL_TOKEN not set — ephemeral token generato per questa sessione.')
     _logger.critical('BOOT: Ogni restart cambia il token → CF Worker riceve 401 finché il secret non è aggiornato!')
@@ -430,3 +431,4 @@ else:
     _logger.warning('BOOT: no frontend at %s', _STATIC_DIR)
 
 _logger.info('BOOT: main.py v%s ready — %s routes registered ✓', app.version, len(app.routes))
+

models/ai_client.py

@@ -484,8 +484,8 @@ class AIClient:
         # RF-3: attivo se CF_API_TOKEN + CF_ACCOUNT_ID entrambi presenti.
         # Endpoint OpenAI-compatible: /accounts/{id}/ai/v1
         # Free: 10K req/giorno, nessun account separato — usa il token Cloudflare.
-        cf_token = os.getenv("CF_API_TOKEN")
-        cf_account = os.getenv("CF_ACCOUNT_ID")
+        cf_token = os.getenv("CF_API_TOKEN") or os.getenv("CLOUDFLARE_API_TOKEN")
+        cf_account = os.getenv("CF_ACCOUNT_ID") or os.getenv("CLOUDFLARE_ACCOUNT_ID")
         if cf_token and cf_account and not os.getenv("DISABLE_CF_PROVIDER"):
             providers.append(ProviderConfig(
                 name="cloudflare",