sync: 150 file da Baida98/AI@eec8e22f (2026-08-14 17:10 UTC) [deploy-all] (#29)

- sync: 150 file da Baida98/AI@eec8e22f (2026-08-14 17:10 UTC) [deploy-all] (de6fcf343ff73e4a471bab6e3608db6f77011c70)

agents/planner.py

@@ -210,7 +210,7 @@ class Planner:
 
     def _get_fast_llm(self) -> AIClient:
         """Gap-5: Cerebras gpt-oss-120b (2000+ tok/s) per quick-start draft.
-        Fallback: Groq llama-3.1-8b-instant se CEREBRAS_API_KEY assente."""
+        Fallback: Groq openai/gpt-oss-20b se CEREBRAS_API_KEY assente."""
         try:
             from models.role_router import RoleRouter, Role
             return RoleRouter.get_client(Role.REASONER)  # Cerebras 120B

agents/unified_loop_llm.py

@@ -65,7 +65,7 @@ class LLMSelectionMixin:
         return self._coder_llm
 
     def _get_fast_llm(self) -> Any:
-        """S-FAST: return Role.FAST client (Groq llama-3.1-8b-instant) per query semplici.
+        """S-FAST: return Role.FAST client (Groq openai/gpt-oss-20b) per query semplici.
         Caricato lazy e cachato in self._fast_llm — zero overhead dopo il primo accesso.
         Fallback silenzioso su self.llm se GROQ_API_KEY mancante o RoleRouter non disponibile."""
         if self._fast_llm is None:
@@ -284,11 +284,18 @@ class LLMSelectionMixin:
     _FORMAT_DIRECTIVE_CODE = (
         "FORMATO RISPOSTA OBBLIGATORIO — CODICE:\n"
         "• Usa SEMPRE blocchi markdown con linguaggio specificato (```python, ```typescript, ecc.)\n"
-        "• Struttura: breve spiegazione → blocco codice completo → come usarlo\n"
-        "• Ogni blocco deve essere autonomo ed eseguibile senza modifiche\n"
-        "• Aggiungi commenti inline per la logica non ovvia\n"
-        "• Se multi-file: mostra ogni file in un blocco separato con il nome come titolo\n"
-        "• Formato titolo file OBBLIGATORIO: ### src/nomefile.tsx (H3 - risparmia spazio verticale su mobile)"
+        "• Per una richiesta di singolo snippet, emetti ESATTAMENTE un blocco nel linguaggio richiesto; "
+        "non sostituirlo con pseudocodice, analisi o un blocco generico.\n"
+        "• Il blocco deve contenere la soluzione completa, autonoma ed eseguibile senza modifiche; "
+        "mantieni gli export e la firma richiesti.\n"
+        "• Prima di rispondere applica il CONTROLLO FINALE: codice compilabile, nessun placeholder/TODO, "
+        "nessun simbolo non definito, tipi espliciti.\n"
+        "• Per codice async con handler indipendenti: includi `async`, `await` e `try/catch` oppure "
+        "`Promise.allSettled` per isolare ogni errore.\n"
+        "• Per correzioni React useEffect: preserva la struttura, usa AbortController o una guardia di annullamento "
+        "e restituisci sempre cleanup (`return () => ...`).\n"
+        "• Aggiungi commenti inline solo per la logica non ovvia. Se multi-file: mostra ogni file in un blocco separato "
+        "con il nome come titolo; formato titolo: ### src/nomefile.tsx."
     )
     _FORMAT_DIRECTIVE_MARKDOWN = (
         "FORMATO RISPOSTA OBBLIGATORIO — STRUTTURATO:\n"

agents/unified_loop_prompts.py

@@ -49,8 +49,8 @@ class PromptBuilderMixin:
         "8c. OBBLIGO TypeScript: ogni snippet di codice TypeScript DEVE essere in blocchi "
         "```typescript```...```typescript. Mai inline, mai in blocchi generici. "
         "Il codice deve compilare: nessun placeholder, nessun TODO, tipi espliciti. In caso di REFACTORING: sostituisci SEMPRE nomi di variabili a lettera singola (p, m, v) con nomi semantici e descrittivi, e usa interfacce o tipi per ogni oggetto complesso.\n"
-        "8d. REASONING: Per problemi complessi, scomponi il problema in sotto-task logici. Verifica la coerenza dei risultati intermedi prima di procedere al calcolo finale.
-9. Per decisioni architetturali: dai 3 opzioni con pro/contro e raccomandazione\n"
+        "8d. REASONING: Per problemi complessi, scomponi il problema in sotto-task logici. Verifica la coerenza dei risultati intermedi prima di procedere al calcolo finale.\n"
+        "9. Per decisioni architetturali: dai 3 opzioni con pro/contro e raccomandazione\n"
         "10. NON inventare mai informazioni su te stesso: token usati, context window, "
         "versione, architettura, parametri interni. Se non lo sai con certezza, "
         "di esplicitamente 'non ho accesso a questa informazione'.\n"
@@ -306,7 +306,7 @@ class PromptBuilderMixin:
 
       # ── S200: Context-aware rule injection ──────────────────────────────────────
     # Seleziona solo le regole rilevanti per il task corrente.
-    # Con llama-3.1-8b-instant (8K context), mettere tutto nel system prompt
+    # Con openai/gpt-oss-20b (8K context), mettere tutto nel system prompt
     # causa troncamento silenzioso — le regole non vengono mai lette.
     # Soluzione: iniettare 2-4 regole contestuali ALLA FINE del user message
     # (posizione con massima attenzione del modello = "recency bias").

api/auth_guard.py

@@ -228,6 +228,38 @@ async def _resolve_role(
     return AuthRole.USER
 
 
+async def require_private_state_machine(
+    request: 'Request',
+    x_internal_token: Optional[str] = Header(None, alias="X-Internal-Token"),
+) -> AuthRole:
+    """Autorizza esclusivamente il proxy Pages dello stato privato.
+
+    Usa un token dedicato per non ruotare o esporre ``INTERNAL_TOKEN``, da cui
+    dipendono le integrazioni legacy del master B. Il token non conferisce un
+    ruolo più ampio del canale MACHINE e resta soggetto allo stesso rate limit.
+    """
+    import secrets as _sec_comp
+    private_token = _get_token("PRIVATE_STATE_INTERNAL_TOKEN")
+    if not private_token:
+        raise HTTPException(status_code=503, detail="Canale stato privato non configurato")
+    if not x_internal_token or not _sec_comp.compare_digest(x_internal_token, private_token):
+        raise HTTPException(status_code=403, detail="Permessi insufficienti per lo stato privato")
+
+    client_ip = (
+        request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
+        or request.headers.get('X-Real-IP', '')
+        or (request.client.host if request.client else None)
+    ) or None
+    allowed, retry_after = _check_rate_limit(int(AuthRole.MACHINE), x_internal_token, client_ip)
+    if not allowed:
+        raise HTTPException(
+            status_code=429,
+            detail="Rate limit stato privato superato",
+            headers={'Retry-After': str(retry_after)},
+        )
+    return AuthRole.MACHINE
+
+
 def require_role(min_role: AuthRole):
     """
     FastAPI Depends factory per autorizzazione granulare.

api/benchmark.py

@@ -373,7 +373,7 @@ async def run_benchmark(
 #
 # Per ogni categoria agente (DA / ORCH / MC / REC):
 #   1. Inietta la context rule via UnifiedLoopPrompts._pick_context_rules()
-#   2. Chiama il LLM (ARCHITECT = llama-3.3-70b-versatile) a temperatura 0.3
+#   2. Chiama il LLM (ARCHITECT = openai/gpt-oss-120b) a temperatura 0.3
 #   3. Valuta la risposta con checker regex (stessa logica di benchmark-extended.mjs)
 #   4. Produce score 0-100 per categoria + media totale
 #

api/exec.py

@@ -568,8 +568,8 @@ async def llm_fix_code(
     _FIX_CHAIN = []
     groq_key = os.getenv('GROQ_API_KEY', '')
     if groq_key:
-        _FIX_CHAIN.append(('https://api.groq.com/openai/v1', groq_key, 'llama-3.3-70b-versatile'))
-        _FIX_CHAIN.append(('https://api.groq.com/openai/v1', groq_key, 'llama-3.1-8b-instant'))
+        _FIX_CHAIN.append(('https://api.groq.com/openai/v1', groq_key, 'openai/gpt-oss-120b'))
+        _FIX_CHAIN.append(('https://api.groq.com/openai/v1', groq_key, 'openai/gpt-oss-20b'))
 
     or_key = os.getenv('OPENROUTER_API_KEY', '')
     if or_key:

api/private_state.py

@@ -0,0 +1,382 @@
+"""API private per il cutover browser → backend delle tabelle Supabase sensibili.
+
+Questi endpoint sono destinati esclusivamente alle Pages Functions, che inoltrano
+``X-Internal-Token`` al master B. Nessun client browser riceve una service-role key
+o accede direttamente alle tabelle private.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import math
+import re
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel, Field, field_validator
+
+from .auth_guard import require_private_state_machine
+from .state import get_supabase
+
+_logger = logging.getLogger("agente_ai.api.private_state")
+router = APIRouter(
+    prefix="/api/private-state",
+    tags=["private-state"],
+    dependencies=[Depends(require_private_state_machine)],
+)
+
+_RAG_LANGUAGE = "rag_chunk"
+_RAG_PREFIX = "__rag_chunk"
+_MAX_RAG_CHUNKS = 100
+_MAX_RAG_CONTENT_CHARS = 10_000
+_MAX_EMBEDDING_DIMENSIONS = 4_096
+_MAX_TASK_PAGE = 100
+
+
+def _db() -> Any:
+    """Restituisce il client service-role del backend o un errore non sensibile."""
+    client = get_supabase()
+    if client is None:
+        raise HTTPException(status_code=503, detail="Archivio privato temporaneamente non disponibile")
+    return client
+
+
+async def _call(operation):
+    """Esegue il client sincrono Supabase senza bloccare l'event loop FastAPI."""
+    try:
+        return await asyncio.to_thread(operation, _db())
+    except HTTPException:
+        raise
+    except Exception as exc:  # Non esporre dettagli backend, query o dati al browser.
+        _logger.warning("[private-state] database operation failed: %s", type(exc).__name__)
+        raise HTTPException(status_code=502, detail="Operazione sullo stato privato non riuscita") from exc
+
+
+def _json_object(value: object) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+            return parsed if isinstance(parsed, dict) else {}
+        except (TypeError, ValueError):
+            return {}
+    return {}
+
+
+def _as_epoch_ms(value: object) -> int:
+    """Normalizza i valori `timestamptz` PostgREST in millisecondi browser-safe."""
+    if isinstance(value, (int, float)):
+        return int(value)
+    if isinstance(value, datetime):
+        moment = value
+    elif isinstance(value, str):
+        try:
+            moment = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return 0
+    else:
+        return 0
+    if moment.tzinfo is None:
+        moment = moment.replace(tzinfo=timezone.utc)
+    return int(moment.timestamp() * 1_000)
+
+
+def _finite_vector(values: list[float]) -> list[float]:
+    if not values or len(values) > _MAX_EMBEDDING_DIMENSIONS:
+        raise ValueError("dimensione embedding non valida")
+    if any(not math.isfinite(value) for value in values):
+        raise ValueError("embedding contiene valori non finiti")
+    return values
+
+
+class TelegramConfigIn(BaseModel):
+    bot_token: str = Field(min_length=1, max_length=512)
+    chat_id: str = Field(min_length=1, max_length=128)
+
+
+class SkillPatternIn(BaseModel):
+    id: str = Field(min_length=1, max_length=128)
+    task_signature: str = Field(min_length=1, max_length=200)
+    tool_sequence: list[str] = Field(min_length=1, max_length=8)
+    success_count: int = Field(ge=0, le=1_000_000)
+    total_count: int = Field(ge=1, le=1_000_000)
+    last_used: int = Field(ge=0)
+    confidence: float = Field(ge=0, le=1)
+
+    @field_validator("tool_sequence")
+    @classmethod
+    def validate_tools(cls, tools: list[str]) -> list[str]:
+        clean = [tool.strip()[:120] for tool in tools if isinstance(tool, str) and tool.strip()]
+        if not clean:
+            raise ValueError("tool_sequence non valida")
+        return clean
+
+
+class RagChunkIn(BaseModel):
+    id: str = Field(min_length=1, max_length=128)
+    path: str = Field(min_length=1, max_length=256)
+    content: str = Field(min_length=51, max_length=_MAX_RAG_CONTENT_CHARS)
+    embedding: list[float] | None = Field(default=None, max_length=_MAX_EMBEDDING_DIMENSIONS)
+
+    @field_validator("embedding")
+    @classmethod
+    def validate_embedding(cls, value: list[float] | None) -> list[float] | None:
+        return _finite_vector(value) if value is not None else None
+
+
+class RagIndexIn(BaseModel):
+    file_id: str = Field(min_length=1, max_length=40)
+    chunks: list[RagChunkIn] = Field(min_length=1, max_length=_MAX_RAG_CHUNKS)
+
+    @field_validator("file_id")
+    @classmethod
+    def validate_file_id(cls, value: str) -> str:
+        if not re.fullmatch(r"[a-z0-9_]+", value):
+            raise ValueError("file_id non valido")
+        return value
+
+
+class RagSearchIn(BaseModel):
+    query_embedding: list[float] | None = Field(default=None, max_length=_MAX_EMBEDDING_DIMENSIONS)
+    query: str = Field(default="", max_length=2_000)
+    similarity_threshold: float = Field(default=0.22, ge=-1, le=1)
+    match_count: int = Field(default=5, ge=1, le=10)
+
+    @field_validator("query_embedding")
+    @classmethod
+    def validate_query_embedding(cls, value: list[float] | None) -> list[float] | None:
+        return _finite_vector(value) if value is not None else None
+
+    @field_validator("query")
+    @classmethod
+    def validate_query(cls, value: str) -> str:
+        if not value.strip() and value == "":
+            return ""
+        return value.strip()
+
+
+@router.get("/sessions")
+async def list_sessions(
+    max_age_ms: int = Query(default=300_000, ge=10_000, le=3_600_000),
+    limit: int = Query(default=100, ge=1, le=200),
+) -> dict[str, object]:
+    """Restituisce esclusivamente i metadati delle sessioni agente ancora vive."""
+    cutoff = (datetime.now(timezone.utc) - timedelta(milliseconds=max_age_ms)).isoformat()
+
+    def operation(client: Any):
+        return client.table("agent_tasks").select("task_id,context,updated_at") \
+            .eq("status", "__session__").gte("updated_at", cutoff) \
+            .order("updated_at", desc=True).limit(limit).execute()
+
+    result = await _call(operation)
+    sessions: list[dict[str, object]] = []
+    for row in result.data or []:
+        context = _json_object(row.get("context"))
+        session_id = str(context.get("sessionId") or row.get("task_id") or "").strip()
+        if not session_id:
+            continue
+        claimed = context.get("claimedFiles")
+        sessions.append({
+            "session_id": session_id,
+            "session_name": str(context.get("sessionName") or session_id)[:160],
+            "sprint": str(context["sprint"])[:120] if context.get("sprint") else None,
+            "claimed_files": [str(item)[:300] for item in claimed[:100]] if isinstance(claimed, list) else [],
+            "last_heartbeat": _as_epoch_ms(context.get("lastHeartbeat")) or _as_epoch_ms(row.get("updated_at")),
+            "current_task": str(context["currentTask"])[:500] if context.get("currentTask") else None,
+        })
+    return {"sessions": sessions}
+
+
+@router.get("/tasks")
+async def list_tasks(
+    limit: int = Query(default=20, ge=1, le=_MAX_TASK_PAGE),
+    offset: int = Query(default=0, ge=0, le=10_000),
+    status: str | None = Query(default=None, max_length=64),
+) -> dict[str, object]:
+    """Lista task non di configurazione per il TMA, con conteggi per stato."""
+    normalized_status = status.strip().upper() if status else ""
+
+    def operation(client: Any):
+        query = client.table("agent_tasks").select("task_id,goal,status,updated_at") \
+            .neq("status", "__session__").neq("status", "__config__")
+        if normalized_status:
+            query = query.eq("status", normalized_status)
+        page = query.order("updated_at", desc=True).range(offset, offset + limit - 1).execute()
+        all_statuses = client.table("agent_tasks").select("status") \
+            .neq("status", "__session__").neq("status", "__config__").limit(2_000).execute()
+        return page, all_statuses
+
+    page, all_statuses = await _call(operation)
+    counts: dict[str, int] = {}
+    for row in all_statuses.data or []:
+        key = str(row.get("status") or "UNKNOWN").upper()
+        counts[key] = counts.get(key, 0) + 1
+    tasks = [
+        {
+            "task_id": str(row.get("task_id") or ""),
+            "goal": str(row.get("goal") or "")[:1_000],
+            "status": str(row.get("status") or "UNKNOWN"),
+            "updated_at": _as_epoch_ms(row.get("updated_at")),
+        }
+        for row in page.data or []
+    ]
+    return {"tasks": tasks, "counts": counts, "offset": offset, "limit": limit}
+
+
+@router.post("/telegram-config")
+async def save_telegram_config(payload: TelegramConfigIn) -> dict[str, bool]:
+    """Salva la configurazione Telegram nel record privato del daemon."""
+    now = datetime.now(timezone.utc).isoformat()
+    row = {
+        "task_id": "__telegram_config__",
+        "goal": "__telegram_config__",
+        "status": "__config__",
+        "max_steps": 0,
+        "context": json.dumps({"botToken": payload.bot_token, "chatId": payload.chat_id}),
+        "updated_at": now,
+    }
+
+    def operation(client: Any):
+        return client.table("agent_tasks").upsert(row, on_conflict="task_id").execute()
+
+    await _call(operation)
+    return {"ok": True}
+
+
+@router.get("/skill-patterns")
+async def list_skill_patterns(limit: int = Query(default=100, ge=1, le=100)) -> dict[str, object]:
+    """Carica pattern cloud per il merge con lo storage locale Dexie."""
+
+    def operation(client: Any):
+        return client.table("skill_patterns").select(
+            "id,task_signature,tool_sequence,success_count,total_count,last_used,confidence"
+        ).order("confidence", desc=True).limit(limit).execute()
+
+    result = await _call(operation)
+    return {"patterns": result.data or []}
+
+
+@router.put("/skill-patterns/{pattern_id}")
+async def upsert_skill_pattern(pattern_id: str, payload: SkillPatternIn) -> dict[str, bool]:
+    """Sincronizza un pattern già validato dal layer locale del browser."""
+    if pattern_id != payload.id:
+        raise HTTPException(status_code=400, detail="Identificatore pattern non coerente")
+
+    row = payload.model_dump()
+
+    def operation(client: Any):
+        return client.table("skill_patterns").upsert(row, on_conflict="id").execute()
+
+    await _call(operation)
+    return {"ok": True}
+
+
+@router.post("/rag/index")
+async def index_rag(payload: RagIndexIn) -> dict[str, int]:
+    """Sostituisce i chunk RAG di un file senza esporre `vfs_files` al browser."""
+    prefix = f"{_RAG_PREFIX}/{payload.file_id}/"
+    rows: list[dict[str, object]] = []
+    now = int(time.time() * 1000)
+    for chunk in payload.chunks:
+        if not chunk.path.startswith(prefix) or not chunk.id.startswith(f"rag-{payload.file_id}-"):
+            raise HTTPException(status_code=400, detail="Chunk RAG non coerente con il file")
+        row: dict[str, object] = {
+            "id": chunk.id,
+            "user_id": "default",
+            "path": chunk.path,
+            "content": chunk.content,
+            "language": _RAG_LANGUAGE,
+            "created_at": now,
+            "updated_at": now,
+        }
+        if chunk.embedding:
+            row["embedding"] = "[" + ",".join(str(value) for value in chunk.embedding) + "]"
+            row["embedding_vec"] = chunk.embedding
+        rows.append(row)
+
+    def operation(client: Any):
+        client.table("vfs_files").delete().eq("language", _RAG_LANGUAGE).like("path", prefix + "%").execute()
+        return client.table("vfs_files").upsert(rows).execute()
+
+    await _call(operation)
+    return {"indexed": len(rows)}
+
+
+def _parse_vector(value: object) -> list[float] | None:
+    if isinstance(value, list):
+        try:
+            return [float(item) for item in value]
+        except (TypeError, ValueError):
+            return None
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+            return [float(item) for item in parsed] if isinstance(parsed, list) else None
+        except (TypeError, ValueError):
+            return None
+    return None
+
+
+def _cosine_similarity(left: list[float], right: list[float]) -> float:
+    if len(left) != len(right) or not left:
+        return 0.0
+    numerator = sum(a * b for a, b in zip(left, right))
+    left_norm = math.sqrt(sum(a * a for a in left))
+    right_norm = math.sqrt(sum(b * b for b in right))
+    return numerator / (left_norm * right_norm) if left_norm and right_norm else 0.0
+
+
+@router.post("/rag/search")
+async def search_rag(payload: RagSearchIn) -> dict[str, object]:
+    """Ricerca pgvector con fallback server-side alla similarità coseno in memoria."""
+
+    def operation(client: Any):
+        if payload.query_embedding:
+            try:
+                rpc = client.rpc("match_rag_chunks", {
+                    "query_embedding": payload.query_embedding,
+                    "similarity_threshold": payload.similarity_threshold,
+                    "match_count": payload.match_count,
+                }).execute()
+                if isinstance(rpc.data, list):
+                    return {"mode": "pgvector", "rows": rpc.data}
+            except Exception as exc:
+                _logger.info("[private-state] rag RPC unavailable, using fallback: %s", type(exc).__name__)
+
+        result = client.table("vfs_files").select("content,embedding,path") \
+            .eq("language", _RAG_LANGUAGE).limit(300).execute()
+        query_words = {word for word in re.split(r"\W+", payload.query.lower()) if len(word) > 3}
+        scored: list[dict[str, object]] = []
+        for row in result.data or []:
+            content = str(row.get("content") or "")
+            embedding = _parse_vector(row.get("embedding"))
+            if payload.query_embedding and embedding:
+                score = _cosine_similarity(payload.query_embedding, embedding)
+            else:
+                lower = content.lower()
+                hits = sum(1 for word in query_words if word in lower)
+                score = hits / len(query_words) if query_words else 0.0
+            if score >= payload.similarity_threshold:
+                scored.append({
+                    "content": content,
+                    "similarity": score,
+                    "path": str(row.get("path") or ""),
+                })
+        scored.sort(key=lambda item: float(item["similarity"]), reverse=True)
+        return {"mode": "cosine_fallback", "rows": scored[:payload.match_count]}
+
+    result = await _call(operation)
+    rows = [
+        {
+            "content": str(row.get("content") or ""),
+            "similarity": float(row.get("similarity") or 0),
+            "path": str(row.get("path") or ""),
+        }
+        for row in result["rows"]
+        if isinstance(row, dict)
+    ]
+    return {"results": rows, "mode": result["mode"]}

api/providers.py

@@ -883,7 +883,7 @@ async def update_provider_models(role: AuthRole = Depends(require_role(AuthRole.
 
     # Mappa: modello_vecchio -> modello_nuovo
     MODEL_FIXES = [
-        ("llama-3.1-70b-versatile",    "llama-3.3-70b-versatile"),
+        ("llama-3.1-70b-versatile",    "openai/gpt-oss-120b"),
         ("llama3.1-70b",               "llama-4-scout"),
         ("llama-3.1-405b-instruct",    "meta/llama-3.3-70b-instruct"),
         ("llama-3.1-405b",             "meta-llama/llama-4-scout:free"),

api/research.py

@@ -283,7 +283,7 @@ async def _synthesize(topic: str, sources: list[dict]) -> str:
                 "https://api.groq.com/openai/v1/chat/completions",
                 headers={"Authorization": f"Bearer {groq_key}", "Content-Type": "application/json"},
                 json={
-                    "model": "llama-3.1-8b-instant",
+                    "model": "openai/gpt-oss-20b",
                     "max_tokens": 700,
                     "messages": [
                         {"role": "system", "content": "Sei un assistente che sintetizza informazioni web. Rispondi sempre in italiano. Sii conciso e preciso."},

api/speculative.py

@@ -2,7 +2,7 @@
 backend/api/speculative.py — Speculative Tool Firing (S361)
 
 Pre-fires tool calls in parallel while the main model is processing.
-Uses Groq llama-3.1-8b-instant for ultra-fast intent extraction (~200-300ms).
+Uses Groq openai/gpt-oss-20b for ultra-fast intent extraction (~200-300ms).
 Results stored in a per-goal cache, consumed by _run_direct_tools before actual execution.
 
 Architecture:
@@ -141,7 +141,7 @@ def _get_spec_groq_client() -> Any:
 
 async def _extract_tools_fast(goal: str) -> list[dict]:
     """
-    Usa Groq llama-3.1-8b-instant per estrarre tool calls in ~300ms.
+    Usa Groq openai/gpt-oss-20b per estrarre tool calls in ~300ms.
     Fallback silenzioso → [] se timeout, errore o key assente.
     """
     if not os.getenv("GROQ_API_KEY"):
@@ -154,7 +154,7 @@ async def _extract_tools_fast(goal: str) -> list[dict]:
         resp = await asyncio.wait_for(
             asyncio.to_thread(
                 client.chat.completions.create,
-                model="llama-3.1-8b-instant",
+                model="openai/gpt-oss-20b",
                 messages=[{"role": "user", "content": prompt}],
                 temperature=0.0,
                 max_tokens=400,  # S587: 256→400 — JSON array da goal[:500] supera 256 tok

api/vault.py

@@ -95,23 +95,21 @@ def _vault_encrypt(plaintext: str) -> str:
 
 
 def _vault_decrypt(ciphertext: str) -> str:
-    """Decrittografia: tenta Fernet; il fallback XOR è permesso solo per la migrazione di vecchi segreti."""
+    """Decrittografa Fernet; il formato XOR legacy è ammesso solo in sviluppo per migrazione."""
     if _fernet_instance:
         try:
             return _fernet_instance.decrypt(ciphertext.encode('ascii')).decode('utf-8')
         except Exception:
-            _vault_logger.warning('Vault: rilevato segreto legacy (XOR) — si consiglia di risalvarlo per migrare a Fernet')
-    
+            _vault_logger.warning('Vault: ciphertext non-Fernet rilevato')
+
     if os.getenv('ENV', 'production') == 'development':
         return _vault_decrypt_xor(ciphertext)
-    
-    # In produzione, se Fernet fallisce e non siamo in dev, blocchiamo i segreti non sicuri
-    # a meno che non sia strettamente necessario per la migrazione.
-    try:
-        return _vault_decrypt_xor(ciphertext)
-    except Exception as e:
-        _vault_logger.error(f'Vault: errore decrittografia segreto: {e}')
-        raise HTTPException(status_code=500, detail='Vault decryption error: invalid key or corrupted data')
+
+    _vault_logger.error('Vault: ciphertext legacy rifiutato in produzione')
+    raise HTTPException(
+        status_code=422,
+        detail='Vault decryption error: legacy ciphertext is not accepted in production',
+    )
 
 
 # ── XOR legacy (usato solo come fallback per migrazione segreti esistenti) ────
@@ -246,6 +244,8 @@ async def vault_get_token(
         raise HTTPException(status_code=404, detail=f"Chiave '{key}' non trovata nel vault")
     try:
         return {'key': key, 'value': _vault_decrypt(data[key])}
+    except HTTPException:
+        raise
     except Exception as e:
         raise HTTPException(status_code=500, detail=f'Decryption error: {e}')
 

main.py

@@ -140,6 +140,7 @@ _ROUTER_MAP = {
     "marketplace": "marketplace",
     "plugins": "plugins",
     "skills": "skills",
+    "private_state": "private_state",
     "auth": "auth_managed",
     # ── Aggiunti ROUTER-COMPLETE (29 moduli orfani rimontati) ─────────────────
     "agent_checkpoint": "agent_checkpoint",

models/ai_client.py

@@ -24,6 +24,20 @@ from api.semantic_cache import get_cached_response, set_cached_response
 import logging
 _logger = logging.getLogger("agente_ai")
 
+class ProviderUnavailableError(RuntimeError):
+    """Raised when no configured LLM provider can produce a response.
+
+    The error deliberately includes provider names only, never credentials or
+    raw upstream payloads, so callers can distinguish infrastructure failure
+    from a model answer without leaking sensitive data.
+    """
+
+    def __init__(self, providers: list[str] | tuple[str, ...]) -> None:
+        self.providers = tuple(providers)
+        detail = ", ".join(self.providers) if self.providers else "none"
+        super().__init__(f"provider_unavailable: {detail}")
+
+
 @dataclass(frozen=True)
 class ProviderConfig:
     id: int = 0
@@ -40,7 +54,7 @@ class ProviderConfig:
 # (nessun proxy CF Worker qui: questo client gira lato backend Python, non browser).
 _PROVIDER_DEFS = [
     # tier 0 — free tier veloce e affidabile
-    {"name": "groq",       "env_key": "GROQ_API_KEY",       "base_url": "https://api.groq.com/openai/v1",                    "model_env": "GROQ_MODEL",      "default_model": "llama-3.3-70b-versatile",     "tier": 0, "purpose": "reasoning"},
+    {"name": "groq",       "env_key": "GROQ_API_KEY",       "base_url": "https://api.groq.com/openai/v1",                    "model_env": "GROQ_MODEL",      "default_model": "openai/gpt-oss-120b",     "tier": 0, "purpose": "reasoning"},
     {"name": "cerebras",   "env_key": "CEREBRAS_API_KEY",   "base_url": "https://api.cerebras.ai/v1",                        "model_env": "CEREBRAS_MODEL",  "default_model": "llama-4-scout",                "tier": 0, "purpose": "reasoning"},
     {"name": "sambanova",  "env_key": "SAMBANOVA_API_KEY",  "base_url": "https://api.sambanova.ai/v1",                       "model_env": "SAMBANOVA_MODEL", "default_model": "DeepSeek-V3.2",               "tier": 0, "purpose": "reasoning"},
     # tier 1 — free tier con rate limit più stretti
@@ -67,6 +81,21 @@ class AIClient:
         providers = self._try_load_from_supabase()
         return providers if providers else self._discover_providers_from_env()
 
+    @staticmethod
+    def _runtime_model_override(row: dict) -> str:
+        """Use an explicit model environment override for a known provider endpoint.
+
+        Supabase remains the source for provider credentials and ordering; runtime
+        model-selection variables deliberately win so emergency model migrations
+        do not require reading or mutating provider secrets in the database.
+        """
+        database_model = str(row.get("default_model", ""))
+        row_base_url = str(row.get("base_url", "")).rstrip("/")
+        for definition in _PROVIDER_DEFS:
+            if row_base_url == definition["base_url"].rstrip("/"):
+                return os.getenv(definition["model_env"], database_model)
+        return database_model
+
     def _try_load_from_supabase(self) -> list[ProviderConfig]:
         url = os.getenv("SUPABASE_URL", "")
         key = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")
@@ -87,7 +116,7 @@ class AIClient:
             return [
                 ProviderConfig(
                     id=row["id"], name=row["name"], api_key=row["api_key"],
-                    base_url=row["base_url"], default_model=row["default_model"],
+                    base_url=row["base_url"], default_model=self._runtime_model_override(row),
                     tier=row["tier"], purpose=row["purpose"], profile="general",
                 )
                 for row in rows
@@ -124,7 +153,9 @@ class AIClient:
             self._client_cache[provider.name] = OpenAI(
                 api_key=provider.api_key, 
                 base_url=provider.base_url, 
-                timeout=20, 
+                # I task coding possono richiedere più di 20 s prima del primo
+                # chunk dal fallback gratuito; il budget esterno resta finito.
+                timeout=45,
                 max_retries=0
             )
         return self._client_cache[provider.name]
@@ -141,7 +172,10 @@ class AIClient:
                     temperature=temperature, 
                     max_tokens=max_tokens
                 ),
-                timeout=15
+                # Il fallback non-streaming deve avere lo stesso budget del client:
+                # 15s scartava provider sani su richieste coding che richiedono
+                # più tempo per produrre una risposta completa dopo uno stream interrotto.
+                timeout=45
             )
             return provider, response.choices[0].message.content or "", _time_mod.monotonic() - start
         except Exception as e:
@@ -188,7 +222,7 @@ class AIClient:
             pool = self.providers[:4]
         
         if not pool:
-            return "🔴 Nessun provider LLM configurato (verifica le API key nei secrets del backend)."
+            raise ProviderUnavailableError([])
         
         # 3. Esecuzione parallela (Ensemble Intelligence)
         tasks = [self._fetch_one(p, messages, temperature, max_tokens) for p in pool]
@@ -204,8 +238,8 @@ class AIClient:
 
     def _judge_best_response(self, results: List[Tuple[ProviderConfig, str, float]], target_purpose: str) -> str:
         valid = [(p, r, t) for p, r, t in results if not r.startswith("ERROR:") and len(r) > 10]
-        if not valid: 
-            return "🔴 Tutti i provider configurati hanno fallito o sono saturi. Riprovo con provider esterni..."
+        if not valid:
+            raise ProviderUnavailableError([p.name for p, _r, _t in results])
         
         def score(item):
             p, r, t = item
@@ -226,16 +260,25 @@ class AIClient:
         # Nessun provider configurato: feedback immediato all'utente invece di
         # cadere silenziosamente nel loop vuoto e dare un messaggio generico.
         if not self.providers:
-            yield (
-                "⚠️ Nessun provider LLM configurato. "
-                "Imposta almeno una delle seguenti variabili d'ambiente: "
-                "GROQ_API_KEY, CEREBRAS_API_KEY, SAMBANOVA_API_KEY, "
-                "OPENROUTER_API_KEY, HF_TOKEN, GEMINI_API_KEY."
-            )
-            return
+            raise ProviderUnavailableError([])
         
-        # Nello streaming proviamo i provider in ordine di tier e performance
-        for provider in self.providers:
+        # Un client di ruolo può contenere un solo provider specializzato.
+        # Dopo il suo primario, integra la flotta runtime non duplicata: un limite
+        # temporaneo di quel provider non deve rendere indisponibile l'intero task.
+        providers = list(self.providers)
+        try:
+            for fallback in self._load_providers():
+                if not any(
+                    current.name == fallback.name
+                    and current.base_url == fallback.base_url
+                    for current in providers
+                ):
+                    providers.append(fallback)
+        except Exception as exc:
+            _logger.debug("Streaming fleet expansion skipped: %s", type(exc).__name__)
+
+        # Nello streaming proviamo il primario di ruolo, poi i fallback runtime.
+        for provider in providers:
             client = self._client_for(provider)
             try:
                 stream = await asyncio.to_thread(
@@ -257,7 +300,7 @@ class AIClient:
                 _logger.warning(f"Streaming fallito su {provider.name}: {e}")
                 continue
         
-        yield "🔴 Errore critico: tutti i provider configurati sono falliti o non disponibili."
+        raise ProviderUnavailableError([provider.name for provider in providers])
 
 
 

models/role_router.py

@@ -38,9 +38,9 @@ _logger = logging.getLogger("models.role_router")
 class Role(str, Enum):
     FAST       = "fast"        # greetings, math semplice, identity — openai/gpt-oss-20b
     ARCHITECT  = "architect"   # planning, ragionamento complesso — llama-4-scout (10M ctx)
-    CODER      = "coder"       # coding, debug — llama-3.3-70b-versatile
-    TESTER     = "tester"      # test gen, debug hints — llama-3.3-70b-versatile
-    CONTEXT    = "context"     # summarization, context compression — llama-3.3-70b-versatile
+    CODER      = "coder"       # coding, debug — openai/gpt-oss-120b
+    TESTER     = "tester"      # test gen, debug hints — openai/gpt-oss-120b
+    CONTEXT    = "context"     # summarization, context compression — openai/gpt-oss-120b
     DEFAULT    = "default"     # AIClient() primary
     RESEARCHER = "researcher"  # web research + document synthesis — gemini-2.0-flash-exp
     REASONER   = "reasoner"    # throughput massimo — Cerebras llama-4-scout (2000+ tok/s)
@@ -85,7 +85,7 @@ class RoleRouter:
 
     @staticmethod
     def _fast_client() -> Any:
-        """Groq llama-3.3-70b-versatile — 344ms TTFT, 100% benchmark qualità.
+        """Groq GPT-OSS 20B per query brevi e a bassa latenza.
         Usato per: greetings, calcoli semplici, identity, domande 1-liner."""
         from models.ai_client import AIClient, ProviderConfig
         groq_key = os.getenv("GROQ_API_KEY")
@@ -96,7 +96,7 @@ class RoleRouter:
             name="groq-fast",
             api_key=groq_key,
             base_url="https://api.groq.com/openai/v1",
-            default_model=os.getenv("GROQ_FAST_MODEL", "llama-3.3-70b-versatile"),
+            default_model=os.getenv("GROQ_FAST_MODEL", "openai/gpt-oss-20b"),
         )
         rest = [p for p in client.providers if p.name not in ("groq", "groq-fast", "groq-tester")]
         client.providers = [fast, *rest]
@@ -161,12 +161,12 @@ class RoleRouter:
 
     @staticmethod
     def _coder_client() -> Any:
-        """Groq llama-3.3-70b-versatile — 358ms TTFT, 100% benchmark qualità.
-        AGGIORNATO 2026-08-04: era Groq openai/gpt-oss-120b.
-        Fallback: OpenRouter llama-4-scout:free se GROQ_API_KEY mancante."""
+        """Groq GPT-OSS 120B per coding e debug.
+        Fallback: provider ordinari del router se GROQ_API_KEY manca."""
         from models.ai_client import AIClient, ProviderConfig
         groq_key = os.getenv("GROQ_API_KEY")
-        model = os.getenv("CODER_MODEL", "llama-3.3-70b-versatile")
+        nvidia_key = os.getenv("NVIDIA_API_KEY")
+        model = os.getenv("CODER_MODEL", "openai/gpt-oss-120b")
         if groq_key:
             client = AIClient()
             coder = ProviderConfig(
@@ -174,9 +174,26 @@ class RoleRouter:
                 api_key=groq_key,
                 base_url="https://api.groq.com/openai/v1",
                 default_model=model,
+                purpose="coding",
             )
-            rest = [p for p in client.providers if p.name not in ("groq", "groq-coder")]
-            client.providers = [coder, *rest]
+            dedicated_fallbacks: list[ProviderConfig] = []
+            if nvidia_key:
+                dedicated_fallbacks.append(
+                    ProviderConfig(
+                        name="nvidia-coder",
+                        api_key=nvidia_key,
+                        base_url="https://integrate.api.nvidia.com/v1",
+                        default_model=os.getenv(
+                            "NVIDIA_MODEL", "nvidia/nemotron-3-ultra-550b-a55b"
+                        ),
+                        purpose="coding",
+                    )
+                )
+            rest = [
+                provider for provider in client.providers
+                if provider.name not in ("groq", "groq-coder", "nvidia", "nvidia-coder")
+            ]
+            client.providers = [coder, *dedicated_fallbacks, *rest]
             client.provider_name = coder.name
             client.default_model = coder.default_model
             client.client = client._client_for(coder)
@@ -189,6 +206,7 @@ class RoleRouter:
                 api_key=openrouter_key,
                 base_url="https://openrouter.ai/api/v1",
                 default_model="meta-llama/llama-4-scout:free",
+                purpose="coding",
             )
             rest = [p for p in client.providers if not p.name.startswith("openrouter")]
             client.providers = [fallback, *rest]
@@ -200,48 +218,66 @@ class RoleRouter:
 
     @staticmethod
     def _researcher_client() -> Any:
-        """Gemini 2.0-flash-exp — TTFT 910ms, ottima per research/synthesis/doc analysis."""
+        """Groq GPT-OSS 120B per analisi e sintesi; la flotta restante è fallback.
+
+        Gemini può essere configurato ma ha una quota indipendente e più stretta:
+        non deve quindi bloccare i task della persona analyst/researcher quando
+        Groq è sano. L'ordine conserva tutti i provider ordinari dopo Groq.
+        """
         from models.ai_client import AIClient, ProviderConfig
-        gemini_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")
         groq_key = os.getenv("GROQ_API_KEY")
-    
-        if gemini_key:
+        if groq_key:
             client = AIClient()
             researcher = ProviderConfig(
-                name="gemini-researcher",
-                api_key=gemini_key,
-                base_url="https://generativelanguage.googleapis.com/v1beta/openai",
-                default_model=os.getenv("GEMINI_MODEL", "gemini-2.0-flash-exp"),
+                name="groq-researcher",
+                api_key=groq_key,
+                base_url="https://api.groq.com/openai/v1",
+                default_model=os.getenv(
+                    "GROQ_RESEARCH_MODEL",
+                    os.getenv("GROQ_MODEL", "openai/gpt-oss-120b"),
+                ),
             )
-            rest = [p for p in client.providers if not p.name.startswith("gemini")]
+            rest = [
+                provider for provider in client.providers
+                if provider.name not in ("groq", "groq-researcher")
+            ]
             client.providers = [researcher, *rest]
             client.provider_name = researcher.name
             client.default_model = researcher.default_model
             client.client = client._client_for(researcher)
             return client
-        elif groq_key:
+        return AIClient()
+
+    @staticmethod
+    def _reasoner_client() -> Any:
+        """Priorità a Groq GPT-OSS per reasoning/MMLU, con flotta runtime come fallback.
+
+        Gemini è soggetto a quote RPM e non deve essere il percorso iniziale per
+        risposte deterministiche a scelta multipla. Cerebras resta un fallback
+        compatibile quando Groq non è configurato."""
+        from models.ai_client import AIClient, ProviderConfig
+        groq_key = os.getenv("GROQ_API_KEY")
+        if groq_key:
             client = AIClient()
-            groq_compound = ProviderConfig(
-                name="groq-compound-researcher",
+            reasoner = ProviderConfig(
+                name="groq-reasoner",
                 api_key=groq_key,
                 base_url="https://api.groq.com/openai/v1",
-                default_model=os.getenv("GROQ_COMPOUND_MODEL", "groq/compound"),
+                default_model=os.getenv(
+                    "GROQ_REASONER_MODEL",
+                    os.getenv("GROQ_MODEL", "openai/gpt-oss-120b"),
+                ),
             )
-            rest = [p for p in client.providers if p.name not in ("groq", "groq-compound-researcher")]
-            client.providers = [groq_compound, *rest]
-            client.provider_name = groq_compound.name
-            client.default_model = groq_compound.default_model
-            client.client = client._client_for(groq_compound)
+            rest = [
+                provider for provider in client.providers
+                if provider.name not in ("groq", "groq-reasoner")
+            ]
+            client.providers = [reasoner, *rest]
+            client.provider_name = reasoner.name
+            client.default_model = reasoner.default_model
+            client.client = client._client_for(reasoner)
             return client
-        return AIClient()
 
-    @staticmethod
-    def _reasoner_client() -> Any:
-        """Cerebras llama-4-scout — 207ms TTFT, 100% qualità (bench 2026-08-04).
-        REASONING MODEL: genera "reasoning" field prima del "content".
-        Richiede max_tokens≥500 per output non-vuoto su task non-triviali.
-        Fallback: _coder_client (Groq 70B) se CEREBRAS_API_KEY mancante."""
-        from models.ai_client import AIClient, ProviderConfig
         cerebras_key = os.getenv("CEREBRAS_API_KEY")
         if not cerebras_key:
             return RoleRouter._coder_client()
@@ -252,7 +288,7 @@ class RoleRouter:
             base_url="https://api.cerebras.ai/v1",
             default_model=os.getenv("CEREBRAS_MODEL", "llama-4-scout"),
         )
-        rest = [p for p in client.providers if not p.name.startswith("cerebras")]
+        rest = [provider for provider in client.providers if not provider.name.startswith("cerebras")]
         client.providers = [reasoner, *rest]
         client.provider_name = reasoner.name
         client.default_model = reasoner.default_model
@@ -306,7 +342,7 @@ class RoleRouter:
 
     @staticmethod
     def _tester_client() -> Any:
-        """Groq llama-3.3-70b-versatile — fast, sufficiente per test gen e debug hints."""
+        """Groq GPT-OSS 20B per test rapidi e debug hints."""
         from models.ai_client import AIClient, ProviderConfig
         groq_key = os.getenv("GROQ_API_KEY")
         if not groq_key:
@@ -316,7 +352,8 @@ class RoleRouter:
             name="groq-tester",
             api_key=groq_key,
             base_url="https://api.groq.com/openai/v1",
-            default_model=os.getenv("GROQ_FAST_MODEL", "llama-3.3-70b-versatile"),
+            default_model=os.getenv("GROQ_FAST_MODEL", "openai/gpt-oss-20b"),
+            purpose="coding",
         )
         rest = [p for p in client.providers if p.name not in ("groq", "groq-tester")]
         client.providers = [tester, *rest]

tests/test_ai_client_provider_unavailability.py

@@ -0,0 +1,89 @@
+import asyncio
+import unittest
+from unittest.mock import patch
+
+from models.ai_client import AIClient, ProviderConfig, ProviderUnavailableError
+
+
+class _FailingCompletions:
+    def create(self, **_kwargs):
+        raise RuntimeError("quota exhausted")
+
+
+class _FailingChat:
+    completions = _FailingCompletions()
+
+
+class _FailingClient:
+    chat = _FailingChat()
+
+
+class _ClientWithFailingProviders(AIClient):
+    def __init__(self):
+        self.providers = [
+            ProviderConfig(name="primary", api_key="x", base_url="https://example.invalid", default_model="model-a"),
+            ProviderConfig(name="fallback", api_key="y", base_url="https://example.invalid", default_model="model-b"),
+        ]
+        self._client_cache = {}
+        self._rr_indices = {}
+
+    def _client_for(self, _provider):
+        return _FailingClient()
+
+
+class ProviderUnavailableTests(unittest.IsolatedAsyncioTestCase):
+    async def test_chat_raises_structured_error_when_every_provider_fails(self):
+        client = _ClientWithFailingProviders()
+
+        with self.assertRaises(ProviderUnavailableError) as raised:
+            await client.chat([{"role": "user", "content": "hello"}], max_tokens=8)
+
+        self.assertEqual(raised.exception.providers, ("primary", "fallback"))
+        self.assertNotIn("api_key", str(raised.exception).lower())
+
+    async def test_stream_chat_raises_structured_error_when_every_provider_fails(self):
+        client = _ClientWithFailingProviders()
+
+        with self.assertRaises(ProviderUnavailableError) as raised:
+            async for _ in client.stream_chat([{"role": "user", "content": "hello"}], max_tokens=8):
+                pass
+
+        self.assertEqual(raised.exception.providers, ("primary", "fallback"))
+        self.assertNotIn("api_key", str(raised.exception).lower())
+
+    async def test_stream_chat_expands_a_role_specific_provider_pool(self):
+        client = _ClientWithFailingProviders()
+        client.providers = [
+            ProviderConfig(name="gemini-role", api_key="x", base_url="https://example.invalid", default_model="gemini")
+        ]
+        runtime_fallback = ProviderConfig(
+            name="nvidia", api_key="y", base_url="https://fallback.invalid", default_model="nemotron"
+        )
+
+        with patch.object(client, "_load_providers", return_value=[runtime_fallback]):
+            with self.assertRaises(ProviderUnavailableError) as raised:
+                async for _ in client.stream_chat([{"role": "user", "content": "hello"}], max_tokens=8):
+                    pass
+
+        self.assertEqual(raised.exception.providers, ("gemini-role", "nvidia"))
+
+
+class RuntimeModelOverrideTests(unittest.TestCase):
+    def test_groq_runtime_model_overrides_database_model(self):
+        row = {
+            "base_url": "https://api.groq.com/openai/v1",
+            "default_model": "llama-3.3-70b-versatile",
+        }
+        with patch.dict("os.environ", {"GROQ_MODEL": "openai/gpt-oss-120b"}, clear=False):
+            self.assertEqual(
+                AIClient._runtime_model_override(row),
+                "openai/gpt-oss-120b",
+            )
+
+    def test_unknown_provider_keeps_database_model(self):
+        row = {"base_url": "https://example.invalid/v1", "default_model": "custom-model"}
+        self.assertEqual(AIClient._runtime_model_override(row), "custom-model")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_coding_output_contract.py

@@ -0,0 +1,27 @@
+import unittest
+
+from agents.unified_loop_llm import LLMSelectionMixin
+
+
+class CodingOutputContractTests(unittest.TestCase):
+    def test_code_directive_requires_extractable_single_snippet(self):
+        directive = LLMSelectionMixin._FORMAT_DIRECTIVE_CODE
+
+        self.assertIn("ESATTAMENTE un blocco", directive)
+        self.assertIn("linguaggio richiesto", directive)
+        self.assertIn("compilabile", directive)
+        self.assertIn("export", directive)
+
+    def test_code_directive_covers_async_and_react_safety(self):
+        directive = LLMSelectionMixin._FORMAT_DIRECTIVE_CODE
+
+        self.assertIn("async", directive)
+        self.assertIn("await", directive)
+        self.assertIn("try/catch", directive)
+        self.assertIn("Promise.allSettled", directive)
+        self.assertIn("AbortController", directive)
+        self.assertIn("return () =>", directive)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_private_state_contract.py

@@ -0,0 +1,46 @@
+import math
+import unittest
+
+from api.auth_guard import require_private_state_machine
+from api.private_state import RagSearchIn, SkillPatternIn, TelegramConfigIn, _as_epoch_ms, _cosine_similarity, _parse_vector, router
+
+
+class PrivateStateContractTests(unittest.TestCase):
+    def test_router_is_namespaced_under_private_state(self):
+        self.assertEqual(router.prefix, "/api/private-state")
+        paths = {route.path for route in router.routes}
+        self.assertIn("/api/private-state/sessions", paths)
+        self.assertIn("/api/private-state/tasks", paths)
+        self.assertIn("/api/private-state/rag/index", paths)
+        self.assertIn("/api/private-state/rag/search", paths)
+        self.assertEqual(router.dependencies[0].dependency, require_private_state_machine)
+
+    def test_rag_search_allows_lexical_fallback_without_embedding(self):
+        payload = RagSearchIn(query="contesto progetto", query_embedding=None)
+        self.assertEqual(payload.query, "contesto progetto")
+        self.assertIsNone(payload.query_embedding)
+
+    def test_rag_search_rejects_non_finite_embedding(self):
+        with self.assertRaises(ValueError):
+            RagSearchIn(query_embedding=[1.0, math.inf])
+
+    def test_skill_pattern_rejects_empty_tool_sequence(self):
+        with self.assertRaises(ValueError):
+            SkillPatternIn(
+                id="pattern", task_signature="task", tool_sequence=["  "],
+                success_count=0, total_count=1, last_used=1, confidence=0,
+            )
+
+    def test_timestamp_is_normalized_for_browser_clients(self):
+        self.assertEqual(_as_epoch_ms("1970-01-01T00:00:01+00:00"), 1_000)
+        self.assertEqual(_as_epoch_ms("not-a-timestamp"), 0)
+
+    def test_private_payload_has_bounded_required_fields(self):
+        config = TelegramConfigIn(bot_token="token", chat_id="chat")
+        self.assertEqual(config.chat_id, "chat")
+        self.assertEqual(_parse_vector("[1, 2]"), [1.0, 2.0])
+        self.assertAlmostEqual(_cosine_similarity([1.0, 0.0], [1.0, 0.0]), 1.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_role_router_researcher.py

@@ -0,0 +1,70 @@
+import os
+import unittest
+from unittest.mock import patch
+
+from models.ai_client import AIClient
+from models.role_router import Role, RoleRouter
+
+
+class ResearcherRoleRoutingTests(unittest.TestCase):
+    @patch.object(AIClient, "_load_providers", return_value=[])
+    @patch.dict(
+        os.environ,
+        {
+            "GROQ_API_KEY": "test-groq-key",
+            "GROQ_MODEL": "openai/gpt-oss-120b",
+            "GEMINI_API_KEY": "test-gemini-key",
+        },
+        clear=False,
+    )
+    def test_researcher_prefers_groq_gpt_oss_when_available(self, _load_providers):
+        client = RoleRouter.get_client(Role.RESEARCHER)
+
+        self.assertEqual(client.provider_name, "groq-researcher")
+        self.assertEqual(client.default_model, "openai/gpt-oss-120b")
+        self.assertEqual(client.providers[0].name, "groq-researcher")
+        self.assertTrue(
+            all(not provider.name.startswith("gemini") for provider in client.providers)
+        )
+
+    @patch.object(AIClient, "_load_providers", return_value=[])
+    @patch.dict(
+        os.environ,
+        {
+            "GROQ_API_KEY": "test-groq-key",
+            "GROQ_MODEL": "openai/gpt-oss-120b",
+            "GEMINI_API_KEY": "test-gemini-key",
+        },
+        clear=False,
+    )
+    def test_reasoner_prefers_groq_gpt_oss_when_available(self, _load_providers):
+        client = RoleRouter.get_client(Role.REASONER)
+
+        self.assertEqual(client.provider_name, "groq-reasoner")
+        self.assertEqual(client.default_model, "openai/gpt-oss-120b")
+        self.assertEqual(client.providers[0].name, "groq-reasoner")
+
+    @patch.object(AIClient, "_load_providers", return_value=[])
+    @patch.dict(
+        os.environ,
+        {
+            "GROQ_API_KEY": "test-groq-key",
+            "CODER_MODEL": "openai/gpt-oss-120b",
+            "NVIDIA_API_KEY": "test-nvidia-key",
+            "NVIDIA_MODEL": "nvidia/nemotron-3-ultra-550b-a55b",
+        },
+        clear=False,
+    )
+    def test_coder_gpt_oss_is_in_the_coding_pool(self, _load_providers):
+        client = RoleRouter.get_client(Role.CODER)
+
+        self.assertEqual(client.provider_name, "groq-coder")
+        self.assertEqual(client.providers[0].purpose, "coding")
+        self.assertEqual(
+            [provider.name for provider in client.providers],
+            ["groq-coder", "nvidia-coder"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()