api/agent.py

71753
"""backend/api/agent.py — Agent tasks, SSE streaming, checkpoints, loops, kernel (S359, S369).

S358: stream_agent_task() usa _loop_registry per evitare re-run al reconnect SSE.
S359: persistenza su Supabase di task metadata + event buffer.
  - Writes: fire-and-forget, non bloccano mai l'SSE.
  - Reads: lazy restore SOLO quando la memoria è vuota (dopo restart backend).
  - Scenario restart HF Space → client riconnette → replay eventi da Supabase.
    * Task SUCCESS/ERROR → replay completo + chiusura immediata.
    * Task era RUNNING   → replay parziale + evento task_interrupted (no token sprecati).
"""
import os, asyncio, json, uuid, time, re

# UTF-8 surrogate fix — Groq occasionally returns lone surrogates in emoji/special chars
# json.dumps raises UnicodeEncodeError for surrogates → SSE stream crashes, loop never completes
_RE_SURROGATES = re.compile(r"[�-�]", re.UNICODE)
def _ss(s: object) -> str:
    """GAP-SURR-FIX: strip lone UTF-16 surrogates + round-trip encode/decode.
    I provider (es. Groq) possono spezzare emoji multi-byte su chunk consecutivi.
    La regex rimuove surrogati isolati; il round-trip cattura byte invalidi residui."""
    if not isinstance(s, str):
        return s
    try:
        cleaned = _RE_SURROGATES.sub("", s)
        cleaned = cleaned.encode("utf-8", errors="replace").decode("utf-8", errors="replace")
    except Exception:
        cleaned = s
    return cleaned
from fastapi import APIRouter, HTTPException, Request, Body
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, field_validator
from typing import Literal
from .state import (
    _agent_tasks, _task_checkpoints, _loop_registry, _run_stream_tasks,
    _prune_agent_tasks, _prune_checkpoints, _prune_loop_registry,
    _get_mem_manager, _get_mem_manager_async, _get_executor, _get_planner, _get_ai_client,
    ReasonLoopIn, AgentTaskIn,
    write_ahead_task_created,  # WRITE-AHEAD: persist immediato alla creazione task
)
from .speculative import fire_speculative_tools
try:
    from .quality_guardian import run_quality_check as _run_quality_check
except Exception:
    _run_quality_check = None
import logging
_logger = logging.getLogger("api.agent")

from .persistence import (
    sb_upsert_task, sb_update_status, sb_append_event,
    sb_restore_task, sb_get_events, sb_delete_task_events,
    sb_list_tasks, sb_save_checkpoint, sb_get_checkpoint,
    sb_restore_handoff_context, sb_upsert_handoff, sb_delete_handoff,  # BG-4
)

def _log_task_exc(task):  # GAP-2.6: log silently-dropped exceptions in fire-and-forget tasks
    if not task.cancelled():
        exc = task.exception()
        if exc:
            _logger.warning("[agent] background task raised %s: %s", type(exc).__name__, exc)
try:
    from .telegram_notify import notify_task_done as _tg_done, notify_task_error as _tg_error, notify_task_start as _tg_start, notify_task_step as _tg_step
except Exception:
    async def _tg_done(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_error(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_start(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_step(*_a, **_kw):  pass  # type: ignore[misc]

router = APIRouter()


# ── Deprecated run_loop ────────────────────────────────────────────────────────

@router.post('/run_loop', deprecated=True)
async def run_loop():
    """Deprecated — use /agent/task instead."""
    from fastapi.responses import JSONResponse
    return JSONResponse(status_code=410, content={"detail": {"error": "Gone", "migration": "/api/agent/tasks"}})


# ─── P17-F5: Persona helpers ──────────────────────────────────────────────────
import re as _re_persona

_PERSONA_KEYWORD_MAP: dict = {}

def _build_persona_kw_map() -> dict:
    import re
    return {
        'researcher': re.compile(
            r'\b(cerca|ricerca|research|trova|notizie|news|url|leggi|articolo|wikipedia|'
            r'google|fonte|source|scrape|fetch|sito|pagina|web|http|verifica|fact.?check)\b',
            re.IGNORECASE
        ),
        'coder': re.compile(
            r'\b(codice|code|funzione|function|bug|script|implementa|python|javascript|'
            r'typescript|refactor|debug|test|classe|class|api|endpoint|sql|database|html|'
            r'css|react|app|applicazione|programma|sviluppa)\b',
            re.IGNORECASE
        ),
        'reasoner': re.compile(
            r'\b(analizza|pianifica|strategia|decide|ragiona|valuta|confronta|'
            r'piano|roadmap|architettura|valutazione|decisione|ottimale|consiglia)\b',
            re.IGNORECASE
        ),
        'analyst': re.compile(
            r'\b(dati|statistiche|grafico|dataset|csv|dataframe|pandas|matplotlib|'
            r'metriche|kpi|trend|visualizza|dashboard|excel|tabella|percentuale|distribuzione)\b',
            re.IGNORECASE
        ),
    }

def _classify_persona_server(goal: str) -> str:
    """P17-F5: classifica la persona dal goal via regex scoring. Zero LLM — zero latency."""
    global _PERSONA_KEYWORD_MAP
    if not _PERSONA_KEYWORD_MAP:
        _PERSONA_KEYWORD_MAP = _build_persona_kw_map()
    if not goal or len(goal) < 4:
        return ''
    best, best_score = '', 0
    for persona_id, pattern in _PERSONA_KEYWORD_MAP.items():
        score = len(pattern.findall(goal))
        if score > best_score:
            best_score, best = score, persona_id
    return best if best_score >= 1 else ''

_PERSONA_CLIENT_CACHE: dict = {}

def _get_persona_llm_client(persona: str, default_client: object) -> object:
    """P17-F5: ritorna il client LLM persona-appropriate via role_router.
    Fallback silente su default_client se la chiave API manca o role_router fallisce.
    Cache in-process — zero overhead dopo il primo accesso.""";
    if not persona:
        return default_client
    if persona in _PERSONA_CLIENT_CACHE:
        return _PERSONA_CLIENT_CACHE[persona]
    _ROLE_MAP = {'researcher': 'RESEARCHER', 'analyst': 'RESEARCHER',
                 'coder': 'CODER', 'reasoner': 'REASONER', 'architect': 'ARCHITECT'}
    role_name = _ROLE_MAP.get(persona.lower())
    if not role_name:
        return default_client
    try:
        from models.role_router import RoleRouter, Role as _Role
        role = getattr(_Role, role_name, None)
        if role is None:
            return default_client
        client = RoleRouter.get_client(role)
        _PERSONA_CLIENT_CACHE[persona] = client
        return client
    except Exception:
        return default_client


async def run_loop_removed():
    """S352: endpoint rimosso. Usare POST /api/agent/tasks + GET /api/agent/tasks/{id}/stream."""
    raise HTTPException(
        status_code=410,
        detail={
            "error": "Gone",
            "message": "Endpoint rimosso. Usare POST /api/agent/tasks + GET /api/agent/tasks/{id}/stream",
            "migration": "/api/agent/tasks",
        },
    )


# ── SSE run-stream ────────────────────────────────────────────────────────────

@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
    _itok = os.getenv('INTERNAL_TOKEN', '')
    if _itok and request.headers.get('X-Internal-Token') != _itok:
        raise HTTPException(401, 'Unauthorized')
    async def generate():
        queue: asyncio.Queue = asyncio.Queue()

        async def step_cb(step: dict) -> None:
            await queue.put(step)

        async def run_loop() -> None:
            try:
                from agents.unified_loop import UnifiedAgentLoop
                # S388: usa singleton _get_ai_client() — nessuna re-istanziazione OpenAI() per request
                client = _get_ai_client()
                try:
                    from agents.critic import Critic
                    from agents.response_verifier import ResponseVerifier
                    _critic   = Critic(llm_client=client)
                    _verifier = ResponseVerifier()
                except Exception:
                    _critic   = None
                    _verifier = None
                # Resume automatico: inietta contesto checkpoint se disponibile (Case 2.5 fall-through)
                _resume_ctx = getattr(body, '_resume_context', None)
                _resume_max = getattr(body, '_resume_max_steps', None) or body.max_steps
                context_str = '\n'.join(m.get('content', '') for m in body.context) if body.context else ''
                # Bug-5-FIX: resume context iniettato DOPO che context_str è definito (era NameError)
                if _resume_ctx:
                    context_str = f"[RIPRESA AUTOMATICA]\n{_resume_ctx}\n\n{context_str}".strip()

                loop = UnifiedAgentLoop(
                    llm_client=client, critic=_critic, verifier=_verifier,
                    memory=await _get_mem_manager_async(), executor=_get_executor(), planner=_get_planner(),
                )
                # S456-X5: prepend project context (projectMemory.getContext() dal frontend)
                if body.project_context:
                    context_str = f"[PROGETTO CORRENTE]\n{body.project_context}\n\n{context_str}".strip()
                # S456-X4: inject top failure patterns appresi dal selfLearning frontend
                if body.learning_hints:
                    # S591: learning_hints[:3]→[:5] — più pattern appresi nel context
                    hints_str = "\n".join(f"- {h}" for h in body.learning_hints[:5])
                    context_str = f"{context_str}\n\n[PATTERN DI ERRORE APPRESI]\n{hints_str}".strip()
                # P35: vincoli negativi dal frontend (agentConstraints.ts → VFS /.agent/constraints.json)
                _neg_c = getattr(body, 'negative_constraints', '') or ''
                if _neg_c:
                    context_str = f"[VINCOLI OPERATIVI APPRESI — NON VIOLARE]\n{_neg_c}\n\n{context_str}".strip()
                result = await loop.run(
                    goal=body.goal, context=context_str,
                    max_steps=body.max_steps, on_step=step_cb,
                    session_id=getattr(body, "session_id", "") or "",
                )
                await queue.put({
                    '__done__': True,
                    'result':   result.get('output',  ''),
                    'engine':   result.get('engine',  'fallback'),
                    'success':  result.get('success', False),
                })
            except Exception as exc:
                # GAP-A1: log incident in registry (fire-and-forget, non-blocking)
                try:
                    from api.incident_registry import log_incident as _log_inc
                    asyncio.create_task(_log_inc(
                        task_id=body.goal[:32].replace(' ', '_'),
                        goal=body.goal, error=str(exc), source="agent",
                    )).add_done_callback(_log_task_exc)
                except Exception as _exc:
                    _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
                await queue.put({'__error__': str(exc)})

        task    = asyncio.create_task(run_loop())
        task_id = body.goal[:32].replace(' ', '_')
        # ABORT-1: registra task + queue per permettere cancellazione via POST /api/agent/abort
        _run_stream_tasks[task_id] = {"task": task, "queue": queue}
        yield "retry: 3000\n\n"
        yield f"data: {json.dumps({'type': 'task_start', 'taskId': task_id})}\n\n"

        # S386: fast-fail — se tutti i provider sono down (heartbeat lo sa già),
        # non aspettare 120s di tentativi: rispondi subito con errore chiaro.
        try:
            from api.state import _heartbeat_state
            _providers = _heartbeat_state.get("providers", [])
            if _providers and not any(p.get("ok") for p in _providers):
                task.cancel()
                _names = ", ".join(p["name"] for p in _providers)
                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

        # S386: timeout ridotto 120→60s — risposta entro 1 minuto o errore esplicito
        timeout_secs    = float(os.getenv('AGENT_STREAM_TIMEOUT', '60'))
        heartbeat_secs  = 15.0
        elapsed         = 0.0
        try:
            while True:
                try:
                    item    = await asyncio.wait_for(queue.get(), timeout=heartbeat_secs)
                    elapsed = 0.0
                except asyncio.TimeoutError:
                    elapsed += heartbeat_secs
                    if elapsed >= timeout_secs:
                        yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id, 'abort_reason': 'timeout', 'abort_source': 'stream_timeout'})}\n\n"  # MX18-ABORT: timeout → task_aborted
                        break
                    yield 'data: {"type":"ping"}\n\n'
                    continue
                # ABORT-2: segnale abort dall'endpoint POST /api/agent/abort
                if "__abort__" in item:
                    _ar  = item.get('abort_reason', 'user_stop')   # MX18-ABORT: dynamic reason
                    _src = item.get('abort_source', 'backend_abort_queue')
                    yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id, 'abort_reason': _ar, 'abort_source': _src})}\n\n"  # MX16+MX18-ABORT
                    break
                if '__error__' in item:
                    yield f"data: {json.dumps({'type': 'task_error', 'taskId': task_id, 'error': _ss(item['__error__'])})}\n\n"
                    break
                # S420: streaming token — emetti subito al frontend senza accumulare
                if item.get('action') == 'text_chunk':
                    yield f"data: {json.dumps({'type': 'text_chunk', 'token': _ss(item.get('token', '')), 'taskId': task_id})}\n\n"
                    continue
                # S758-P4.1: tool_use — chip pre-esecuzione (agent_run_stream path)
                _rs_act = item.get('action', '')
                _rs_st  = item.get('status', '')
                if ((_rs_act == 'tool_start' and _rs_st == 'running') or
                        (_rs_act.startswith('executor:') and _rs_st == 'started')):
                    _rs_tool = _rs_act.replace('executor:', '') if _rs_act.startswith('executor:') else _rs_act
                    yield f"data: {json.dumps({'type': 'tool_use', 'taskId': task_id, 'tool': _rs_tool, 'name': _rs_tool, 'label': item.get('title', _rs_tool.replace('_', ' ').capitalize())})}\n\n"
                if '__done__' in item:
                    yield f"data: {json.dumps({'type': 'task_done', 'taskId': task_id, 'result': _ss(item['result']), 'engine': item['engine'], 'success': item['success']})}\n\n"
                    break
                # S393 Priority 1: Narrative Streaming — arricchisce step_done con explanation
                _NARR_QUICK = {
                    'llm': 'Elaborazione risposta AI',
                    'direct_tools': 'Strumenti diretti',
                    'web_search': 'Ricerca web', 'get_weather': 'Dati meteo',
                    'read_page': 'Lettura pagina', 'calculate': 'Calcolo matematico',
                    'generate_image': 'Generazione immagine AI',
                    'execution_validator_fix': 'Auto-correzione codice (S393)',
                    'tool_governor_skip': 'Tool già eseguito — risultato riutilizzato',
                    # S661: label narrative per tool aggiunti in S648-S659 — prima usavano
                    # _act_q.replace('_',' ').capitalize() → "Apply patch", "Call api" (generico)
                    'apply_patch':   'Applico patch al file…',
                    'call_api':      'Chiamo API REST…',
                    'send_email':    'Invio email…',
                    'create_pdf':    'Genero documento PDF…',
                    'web_research':  'Ricerca multi-fonte…',
                    'write_file':    'Scrivo file…',
                    'read_file':     'Leggo file…',
                    'execute_shell': 'Eseguo comando shell…',
                    'analyze_image': 'Analizzo immagine…',
                    'run_python':    'Eseguo Python (Pyodide)…',
                    # S-GAP1: narrative fasi strategiche
                    'plan':             'Analizzo la richiesta e preparo un piano di esecuzione…',
                    'reflective_debug': 'Ho incontrato un ostacolo — ricalcolo una strategia più efficiente…',
                    'fallback':         'Adotto un approccio alternativo per completare il task…',
                    'smolagents':       'Orchestro gli strumenti necessari…',
                }
                _act_q = item.get('action', '')
                if 'explanation' not in item:
                    item['explanation'] = _NARR_QUICK.get(_act_q, _act_q.replace('_', ' ').capitalize())
                if 'title' not in item:
                    item['title'] = item['explanation']

                # S403: SSE Visibility Guard — classifica ogni step event:
                # "internal"  → mai visibile (pipeline internals: planner, llm, reflection)
                # "progress"  → visibile come progress card (tool reali, auto-fix)
                # "debug"     → visibile solo in dev mode (direct_tools, fast_path)
                # Il frontend filtra per visibility — solo "progress" mostrato all'utente.
                _STEP_VISIBILITY: dict[str, str] = {
                    # Internal pipeline — never shown to user
                    'plan':              'progress',   # S-GAP1
                    'llm':               'internal',
                    'smolagents':        'internal',
                    'fallback':          'progress',   # S-GAP1
                    'reflective_debug':  'progress',   # S-GAP1
                    'fast_path':         'internal',
                    'executor':          'internal',
                    # Progress — shown as step cards (user-visible)
                    'tool_start':              'progress',
                    'execution_validator_fix': 'progress',
                    'goal_verifier':           'progress',
                    'web_search':              'progress',
                    'get_weather':             'progress',
                    'read_page':               'progress',
                    'calculate':               'progress',
                    'generate_image':          'progress',
                    'run_python':              'progress',
                    'tool_governor_skip':      'progress',
                    # S660: tool aggiunti in S648-S659 mancanti da _STEP_VISIBILITY →
                    # fallback rule: _act_q.startswith('tool_') era False per questi →
                    # classificati 'debug' → nascosti all'utente durante esecuzione.
                    'apply_patch':   'progress',
                    'call_api':      'progress',
                    'send_email':    'progress',
                    'create_pdf':    'progress',
                    'web_research':  'progress',
                    'write_file':    'progress',
                    'read_file':     'progress',
                    'execute_shell': 'progress',
                    'analyze_image': 'progress',
                    # Debug — shown only when devMode active
                    'direct_tools':     'debug',
                    # S-LOOP2: fase esecuzione avanzata — visibili come progress card
                    'reasoning_core':    'progress',  # S-LOOP2: ReasoningCore multi-step
                    'browser_verifier':  'progress',  # S-LOOP2: Browser Goal Verification live
                }
                # Fallback: azioni sconosciute con "tool_" prefix → progress; resto → debug
                _vis = _STEP_VISIBILITY.get(_act_q)
                if _vis is None:
                    _vis = 'progress' if _act_q.startswith('tool_') or _act_q.startswith('executor:') else 'debug'
                item['visibility'] = _vis

                yield f"data: {json.dumps({'type': 'step_done', 'step': item, 'taskId': task_id})}\n\n"
        finally:
            task.cancel()
            # ABORT-3: cleanup registro — libera memoria e impedisce abort su task già terminati
            _run_stream_tasks.pop(task_id, None)
            yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


# ── Reason loop / Unified loop ─────────────────────────────────────────────────

@router.post('/api/reason/loop')
async def reason_loop(body: ReasonLoopIn):
    try:
        from agents.unified_loop import UnifiedAgentLoop
        # S388: singleton — riusa il client già inizializzato
        client = _get_ai_client()
        try:
            from agents.critic import Critic
            from agents.response_verifier import ResponseVerifier
            _critic   = Critic(llm_client=client)
            _verifier = ResponseVerifier()
        except Exception:
            _critic   = None
            _verifier = None
        loop = UnifiedAgentLoop(
            llm_client=client, critic=_critic, verifier=_verifier,
            memory=await _get_mem_manager_async(), executor=_get_executor(), planner=_get_planner(),
        )
        context_str = '\n'.join(m.get('content', '') for m in body.context) if body.context else ''
        # N-2-FIX: accumula step intermedi tramite on_step — inclusi nel response JSON per debug frontend
        _steps_log: list[dict] = []
        async def _on_step(step_data: dict) -> None:
            _steps_log.append({
                'action': step_data.get('action', ''),
                'output': str(step_data.get('output', ''))[:400],  # S577: 200→400
            })
        result = await loop.run(goal=body.goal, context=context_str, max_steps=body.max_steps, on_step=_on_step, session_id=getattr(body, "session_id", "") or "")
        if isinstance(result, dict):
            output_text = result.get('output', '') or result.get('answer', '') or ''
            engine_used = result.get('engine', 'ambiguity-gate' if result.get('answer') else 'unknown')
            errors_list = result.get('errors', [])
        else:
            output_text = str(result)
            engine_used = 'unknown'
            errors_list = []
        return {
            'ok':     bool(output_text and output_text.strip()),
            'success': bool(output_text and output_text.strip()),  # alias compat frontend
            'output':  output_text,                                  # alias compat frontend
            'result': output_text,
            'source': 'backend_loop',
            'engine': engine_used,
            'errors': errors_list,
            'steps':  _steps_log,  # N-2-FIX: step intermedi per debug/telemetria frontend
        }
    except Exception as e:
        _logger.error("[reason/loop] Error: %s", e)
        return {
            'ok':     False,
            'result': f'Backend reasoning non disponibile: {e}. Il loop browser continua normalmente.',
            'source': 'fallback',
            'steps':  [],
        }


@router.post('/api/unified/loop')
async def unified_loop(body: ReasonLoopIn):
    """Alias di /api/reason/loop — compatibilità con tutte le versioni frontend."""
    return await reason_loop(body)


# ── Agent kernel ───────────────────────────────────────────────────────────────

@router.get('/api/agent-kernel/status')
async def agent_kernel_status():
    gh_token = os.getenv('GITHUB_TOKEN') or os.getenv('GH_TOKEN', '')
    return {
        'dispatch_available': bool(gh_token),
        'workflow_url': 'https://github.com/Baida98/AI/actions/workflows/agent-kernel.yml',
        'mobile_url':   'https://github.com/Baida98/AI/actions',
        'secrets_needed': ['OPENROUTER_API_KEY', 'GROQ_API_KEY', 'GEMINI_API_KEY', 'HF_TOKEN', 'NVIDIA_API_KEY'],
        'usage': 'Vai su GitHub Actions → Agent Kernel — no PC → Run workflow → inserisci il goal',
    }


# S442-FIX3: modello Pydantic per agent_kernel_dispatch.
# Prima: body: dict grezzo → mode non validato, goal controllato solo dopo estrazione.
# Ora: validazione in ingresso → 422 chiaro invece di 500 a runtime.
class AgentKernelDispatchIn(BaseModel):
    goal: str
    mode: Literal["plan", "execute", "analyze"] = "plan"

    @field_validator('goal', mode='before')
    @classmethod
    def validate_goal(cls, v: object) -> str:
        if not isinstance(v, str) or not str(v).strip():
            raise ValueError('goal must be a non-empty string')
        return str(v).strip()


@router.post('/api/agent-kernel/dispatch')
async def agent_kernel_dispatch(body: AgentKernelDispatchIn):
    gh_token = os.getenv('GITHUB_TOKEN') or os.getenv('GH_TOKEN', '')
    if not gh_token:
        raise HTTPException(503, detail={
            'error': 'no_github_token',
            'message': 'GITHUB_TOKEN non configurato nel backend.',
        })
    goal = body.goal
    mode = body.mode
    import httpx as _httpx
    try:
        async with _httpx.AsyncClient(timeout=15) as _hc:
            _resp = await _hc.post(
                'https://api.github.com/repos/Baida98/AI/actions/workflows/agent-kernel.yml/dispatches',
                json={'ref': 'main', 'inputs': {'goal': goal, 'mode': mode, 'commit_memory': 'true'}},
                headers={
                    'Authorization':       f'Bearer {gh_token}',
                    'Accept':              'application/vnd.github+json',
                    'X-GitHub-Api-Version': '2022-11-28',
                },
            )
        if _resp.status_code >= 400:
            raise HTTPException(_resp.status_code, detail=_resp.text[:500])
        return {'ok': True, 'status': _resp.status_code, 'goal': goal, 'mode': mode}
    except _httpx.HTTPError as e:
        raise HTTPException(502, detail=str(e)[:500])


# ── Agent tasks (FASE 2.1 + S359 persistence) ──────────────────────────────────

@router.post('/api/agent/tasks')
async def create_agent_task(body: AgentTaskIn):
    """
    Crea o recupera un task agent.

    S359: se task_id non è in memoria ma esiste su Supabase (backend ha riavviato),
    il task viene ripristinato dallo store persistente invece di essere riavviato.
    Questo preserva lo stato SUCCESS/ERROR precedente senza sprecare token.
    """
    _prune_agent_tasks()
    task_id = body.taskId or str(uuid.uuid4())

    # Already in memory → return immediately (normal path, includes S358 reconnect)
    if task_id in _agent_tasks:
        return {'taskId': task_id, 'status': _agent_tasks[task_id]['status']}

    # S359: try Supabase lazy restore (only hit network after backend restart)
    restored = await sb_restore_task(task_id)
    if restored:
        # Put restored metadata back into memory so stream_agent_task can use it.
        # Use context from the incoming request (not persisted to save space).
        restored['context'] = body.context
        _agent_tasks[task_id] = restored
        return {'taskId': task_id, 'status': restored['status'], 'restored': True}

    # Brand new task
    created_at = int(time.time() * 1000)
    _agent_tasks[task_id] = {
        'id':              task_id,
        'status':          'QUEUED',
        'goal':            body.goal,
        'context':         body.context,
        'max_steps':       body.max_steps,
        'created_at':      created_at,
        'project_context':   body.project_context,  # S456-X5
        'learning_hints':    body.learning_hints,   # S456-X4
        'resume_from_step':  body.resume_from_step, # P16-F3: passo resume dalla coda
        'persona':           body.persona,           # P17-F5: expertise persona hint
        'session_id':        body.session_id or '',   # P17-F2: BB session key (normalize None→'')
    }
    # WRITE-AHEAD: persiste il task su Supabase immediatamente, prima del checkpoint
    # periodico (15-60s). Finestra di perdita per la fase di creazione → zero.
    asyncio.create_task(write_ahead_task_created(task_id, body.goal)).add_done_callback(_log_task_exc)
    # BG-4: restore cross-session handoff context (async, non-blocking)
    if body.session_id:
        _hctx = await sb_restore_handoff_context(body.session_id)
        if _hctx:
            _agent_tasks[task_id]['_handoff_context'] = _hctx
            asyncio.create_task(sb_delete_handoff(body.session_id)).add_done_callback(_log_task_exc)
    # Persist asynchronously — never block the response
    asyncio.create_task(
        sb_upsert_task(task_id, body.goal, 'QUEUED', body.max_steps, body.context, created_at)
    ).add_done_callback(_log_task_exc)
    # S361: Speculative Tool Firing — pre-fires read-only tools in parallel
    # while the main model processes. Results cached for _run_direct_tools to consume.
    asyncio.create_task(fire_speculative_tools(task_id, body.goal)).add_done_callback(_log_task_exc)
    return {'taskId': task_id, 'status': 'QUEUED'}


# ── S369: List agent tasks (in-memory + Supabase merge) ─────────────────────

@router.get('/api/agent/tasks')
async def list_agent_tasks(limit: int = 50, status: str = ''):
    """
    S369 — Lista tutti i task agent: unione di in-memory (_agent_tasks) e
    Supabase (ultimi N task persistiti). In-memory ha sempre precedenza.

    Query params:
      limit  — max task da Supabase (default 50, max 200)
      status — filtra per status (es. RUNNING, SUCCESS, ERROR); vuoto = tutti
    """
    _prune_agent_tasks()
    now_ms = int(time.time() * 1000)
    limit  = min(max(limit, 1), 200)

    # 1. Task in-memory (live)
    mem_tasks = []
    for tid, t in _agent_tasks.items():
        reg   = _loop_registry.get(tid)
        is_live = reg is not None and not reg.get('done', True)
        mem_tasks.append({
            'taskId':    tid,
            'goal':      (t.get('goal') or '')[:300],  # S606: 200→300
            'status':    t.get('status', 'UNKNOWN'),
            'maxSteps':  t.get('max_steps', 8),
            'createdAt': t.get('created_at', 0),
            'ageMs':     now_ms - t.get('created_at', now_ms),
            'source':    'memory',
            'isLive':    is_live,
        })

    mem_ids = {t['taskId'] for t in mem_tasks}

    # 2. Supabase recent tasks (only if Supabase available)
    sb_tasks = []
    try:
        sb_rows = await sb_list_tasks(limit=limit, status_filter=status or None)
        for r in sb_rows:
            if r['task_id'] in mem_ids:
                continue  # already included from memory
            sb_tasks.append({
                'taskId':    r['task_id'],
                'goal':      (r.get('goal') or '')[:300],  # S606: 200→300
                'status':    r.get('status', 'UNKNOWN'),
                'maxSteps':  r.get('max_steps', 8),
                'createdAt': r.get('created_at', 0),
                'ageMs':     now_ms - r.get('created_at', now_ms),
                'source':    'supabase',
                'isLive':    False,
            })
    except Exception as _exc:
        _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001

    all_tasks = mem_tasks + sb_tasks
    # Apply status filter to in-memory tasks too
    if status:
        all_tasks = [t for t in all_tasks if t['status'] == status.upper()]

    # Sort by createdAt desc (newest first)
    all_tasks.sort(key=lambda t: t['createdAt'], reverse=True)

    return {
        'count':   len(all_tasks),
        'memory':  len(mem_tasks),
        'supabase': len(sb_tasks),
        'tasks':   all_tasks[:limit],
    }


@router.delete('/api/agent/tasks/{task_id}')
async def cancel_agent_task(task_id: str):
    if task_id in _agent_tasks:
        _agent_tasks[task_id]['status'] = 'CANCELLED'
    reg = _loop_registry.get(task_id)
    if reg and not reg.get('done'):
        at = reg.get('asyncio_task')
        if at and not at.done():
            at.cancel()
    # Persist status + clean up events
    asyncio.create_task(sb_update_status(task_id, 'CANCELLED')).add_done_callback(_log_task_exc)
    asyncio.create_task(sb_delete_task_events(task_id)).add_done_callback(_log_task_exc)
    # S361: clean speculative cache for cancelled task
    try:
        goal = _agent_tasks.get(task_id, {}).get('goal', '')
        if goal:
            from .speculative import purge_speculative
            purge_speculative(goal)
    except Exception as _exc:
        _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
    return {'cancelled': task_id}



@router.get('/api/agent/tasks/{task_id}/status')
async def get_agent_task_status(task_id: str):
    """
    Controlla lo stato di un task agent senza aprire un SSE stream.
    Usato dal frontend per recovery al boot: verifica se un task in sospeso
    e` ancora in esecuzione, completato, o scomparso dopo riavvio HF Space.
    Returns: {taskId, status, goal, source: 'memory'|'supabase'|'not_found'}
    """
    if task_id in _agent_tasks:
        t = _agent_tasks[task_id]
        return {'taskId': task_id, 'status': t.get('status', 'UNKNOWN'),
                'goal': (t.get('goal') or '')[:300], 'source': 'memory'}
    restored = await sb_restore_task(task_id)
    if restored:
        return {'taskId': task_id, 'status': restored.get('status', 'UNKNOWN'),
                'goal': (restored.get('goal') or '')[:300], 'source': 'supabase'}
    return {'taskId': task_id, 'status': 'NOT_FOUND', 'source': None}


@router.get('/api/agent/tasks/{task_id}/stream')
async def stream_agent_task(task_id: str, request: Request, resume: int = 0):
    """
    SSE stream per un task agent.

    S358: reconnect-safe via _loop_registry fanout (no re-run mentre il backend gira).
    S359: lazy restore da Supabase dopo restart HF Space:
      - Task SUCCESS/ERROR → replay event buffer da Supabase → chiusura immediata.
      - Task era RUNNING   → replay buffer parziale + evento task_interrupted.
      - Task non trovato   → prova sb_restore_task prima di 404.
    """
    # S359: se task_id non è in memoria, prova il restore da Supabase
    if task_id not in _agent_tasks:
        restored = await sb_restore_task(task_id)
        if restored:
            restored['context'] = []
            _agent_tasks[task_id] = restored
        else:
            raise HTTPException(404, detail=f'Task {task_id} non trovato')

    task = _agent_tasks[task_id]
    _last_event_id = request.headers.get("Last-Event-ID") or request.headers.get("last-event-id")
    _resume_from   = int(_last_event_id) if (_last_event_id and _last_event_id.isdigit()) else resume

    sub_q: asyncio.Queue[str | None] = asyncio.Queue()

    async def generate():
        yield "retry: 3000\n\n"

        reg = _loop_registry.get(task_id)

        is_done_reconnect = reg is not None and reg.get('done', False)
        is_reconnect      = reg is not None and not reg.get('done', False)

        # ── Case 1: loop già finito in questa sessione → replay buffer in-memory ──
        if is_done_reconnect:
            for evt_str in reg['event_buffer'][_resume_from:]:
                yield evt_str
            yield "data: [DONE]\n\n"
            return

        # ── Case 2: loop attivo in questa sessione → reconnect SSE (S358) ─────────
        if is_reconnect:
            join_idx = len(reg['event_buffer'])
            reg['subscriber_queues'].append(sub_q)
            try:
                for evt_str in reg['event_buffer'][_resume_from:join_idx]:
                    yield evt_str
                while True:
                    if _agent_tasks.get(task_id, {}).get('status') == 'CANCELLED':
                        break
                    try:
                        item = await asyncio.wait_for(sub_q.get(), timeout=15.0)
                        if item is None:
                            break
                        yield item
                    except asyncio.TimeoutError:
                        yield ': heartbeat\n\n'
            finally:
                try:
                    reg['subscriber_queues'].remove(sub_q)
                except ValueError as _exc:
                    _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
            yield "data: [DONE]\n\n"
            return

        # ── Case 2.5 (S359): backend riavviato → prova Supabase event buffer ──────
        sb_events = await sb_get_events(task_id)
        if sb_events:
            task_status = task.get('status', 'UNKNOWN')
            terminal    = task_status in ('SUCCESS', 'ERROR', 'CANCELLED')
            # Replay buffer from resume point
            for evt_str in sb_events[_resume_from:]:
                yield evt_str
            if terminal:
                # Task già completato → niente da fare, client ha tutto
                yield "data: [DONE]\n\n"
                return
            else:
                # Task era in esecuzione quando il backend è crashato — prova resume automatico
                _cp_sb = _task_checkpoints.get(task_id) or await sb_get_checkpoint(task_id)
                _can_resume = (
                    _cp_sb is not None and
                    len(_cp_sb.get('plan', [])) >= 1 and
                    len(_cp_sb.get('logs', [])) >= 2
                )
                if _can_resume:
                    # GAP-SYNC-FIX: usa _backend_steps se disponibili (context preciso per resume)
                    _bsteps = _cp_sb.get('_backend_steps', [])
                    if _bsteps:
                        _steps_text = '\n'.join(
                            f"  Passo {s['step']}: {s['action']} → {s['result'][:80]}"
                            for s in _bsteps[-8:]
                        )
                        _rctx = (
                            f"[RESUME AUTOMATICO] Step già completati dal backend:\n{_steps_text}\n"
                            f"Riprendi dal passo {_cp_sb.get('step', 0)+1} senza ripetere quelli già eseguiti."
                        )
                    else:
                        # Fallback: context semantico (piano + log riassuntivi)
                        _rctx = (
                            f"Piano già definito: {' | '.join((_cp_sb.get('plan') or [])[:5])}\n"
                            f"Log fin qui: {' | '.join((_cp_sb.get('logs') or [])[-5:])}\n"
                            f"Riprendi dal passo {_cp_sb.get('step', 0)} senza ripetere gli step già fatti."
                        )
                    task['_resume_context']   = _rctx
                    task['_resume_max_steps'] = max(1, task.get('max_steps', 8) - _cp_sb.get('step', 0))
                    # Fall through a Case 3 — NON fare return
                else:
                    # Nessun checkpoint utile → fallback onesto (comportamento precedente)
                    interrupted_evt = json.dumps({
                        'event':   'task_interrupted',
                        'taskId':  task_id,
                        'reason':  'backend_restarted',
                        'message': 'Il backend si è riavviato durante l\'esecuzione. '
                                   'Premi "Riprova" per rieseguire il task.',
                    })
                    yield f"data: {interrupted_evt}\n\n"
                    _agent_tasks[task_id]['status'] = 'ERROR'
                    asyncio.create_task(sb_update_status(task_id, 'ERROR')).add_done_callback(_log_task_exc)
                    yield "data: [DONE]\n\n"
                    return
        # ── Case 3: nuova esecuzione ──────────────────────────────────────────────
        _prune_loop_registry()
        reg_entry: dict = {
            'asyncio_task':      None,
            'event_buffer':      [],
            'subscriber_queues': [sub_q],
            'done':              False,
            'finished_at':       0.0,
        }
        _loop_registry[task_id] = reg_entry
        _ctr = [0]

        def _sse(event: str, data: dict) -> None:
            """Emit one SSE frame: buffer it, fanout to all subscribers, persist async."""
            _ctr[0] += 1
            s = f"id: {_ctr[0]}\ndata: {json.dumps({'event': event, **data})}\n\n"
            # GAP-3-FIX: text_chunk bypass buffer — fanout diretto, no persist.
            # 800 token x 1 evento/token saturerebbero il cap da 500 evictando step cruciali.
            # Su reconnect iOS i token non servono replay (streaming completato o ricominciato).
            if event == 'text_chunk':
                for q in list(reg_entry['subscriber_queues']):
                    try:
                        q.put_nowait(s)
                    except Exception as _exc:
                        _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
                return
            reg_entry['event_buffer'].append(s)
            # N-5-FIX: cap buffer a 500 eventi — evita crescita illimitata su task lunghi
            if len(reg_entry['event_buffer']) > 500:
                reg_entry['event_buffer'] = reg_entry['event_buffer'][-500:]
            for q in list(reg_entry['subscriber_queues']):
                try:
                    q.put_nowait(s)
                except Exception as _exc:
                    _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
            # S359: persist event asynchronously (fire-and-forget)
            asyncio.create_task(sb_append_event(task_id, _ctr[0], s)).add_done_callback(_log_task_exc)

        _agent_tasks[task_id]['status'] = 'RUNNING'
        asyncio.create_task(sb_update_status(task_id, 'RUNNING')).add_done_callback(_log_task_exc)
        _prune_agent_tasks()

        async def run_loop() -> None:
            try:
                from agents.unified_loop import UnifiedAgentLoop
                # S388: singleton — evita OpenAI() per ogni task
                client = _get_ai_client()
                try:
                    from agents.critic import Critic
                    from agents.response_verifier import ResponseVerifier
                    _critic   = Critic(llm_client=client)
                    _verifier = ResponseVerifier()
                except Exception:
                    _critic   = None
                    _verifier = None

                context_str = '\n'.join(m.get('content', '') for m in task['context']) if task['context'] else ''
                # S456-X5/X4: inject project context + learning hints stored at task creation
                _proj_ctx = task.get('project_context', '')
                if _proj_ctx:
                    context_str = f"[PROGETTO CORRENTE]\n{_proj_ctx}\n\n{context_str}".strip()
                _hints = task.get('learning_hints', [])
                if _hints:
                    # S591: _hints[:3]→[:5] — più pattern appresi nel context (task replay)
                    hints_str = "\n".join(f"- {h}" for h in _hints[:5])
                    context_str = f"{context_str}\n\n[PATTERN DI ERRORE APPRESI]\n{hints_str}".strip()
                # P16-F3: inject resume hint if task was promoted from queue at a specific step
                _resume_step = task.get('resume_from_step')
                if _resume_step:
                    context_str = f"[RIPRESA DA PASSO {_resume_step}] Riprendi dall'iterazione {_resume_step} del task.\n\n{context_str}".strip()
                # P39-UX: Tocco Finale Manus — spiega all'agente come segnalare OAuth mancante
                _connector_hint = (
                    "[CONNETTORI OAUTH]\n"
                    "Se durante il task hai bisogno di un accesso OAuth (GitHub, Google Calendar, Instagram)\n"
                    "ma non hai il token disponibile, includi nella tua risposta finale o parziale:\n"
                    "  [CONNECTOR_NEEDED:github] oppure [CONNECTOR_NEEDED:google] oppure [CONNECTOR_NEEDED:instagram]\n"
                    "Il frontend mostrerà automaticamente un pulsante 'Connetti' all'utente."
                )
                context_str = f"{context_str}\n\n{_connector_hint}".strip() if context_str else _connector_hint
                # GAP-SYNC-FIX: inject _resume_context (set da stream_agent_task su reconnect con checkpoint)
                # Bug: _resume_context era settato su task{} ma mai letto qui → context perduto su resume.
                _resume_ctx = task.get('_resume_context', '')
                if _resume_ctx:
                    context_str = f"{_resume_ctx}\n\n{context_str}".strip()
                # P17-F5: inject Expertise Persona hint se specificato
                _PERSONA_HINTS = {
                    "researcher": (
                        "[PERSONA: RICERCATORE ESPERTO]\n"
                        "- Priorizza sempre la ricerca web aggiornata prima di rispondere\n"
                        "- Cita fonti specifiche (URL, titolo, data) per ogni claim importante\n"
                        "- Struttura le risposte: Sommario → Dettaglio → Fonti\n"
                        "- Verifica incrociando più fonti prima di concludere\n"
                        "- Strumenti preferiti: web_search, read_page, fetch_url, research"
                    ),
                    "coder": (
                        "[PERSONA: SENIOR ENGINEER]\n"
                        "- Scrivi codice production-ready: tipizzato, documentato, con error handling\n"
                        "- Esegui il codice per verificare il funzionamento prima di rispondere\n"
                        "- Preferisci soluzioni robuste e testate su approcci creativi ma fragili\n"
                        "- Documenta funzioni e classi con docstring/JSDoc\n"
                        "- Strumenti preferiti: run_python, write_file, read_file, pip_install"
                    ),
                    "architect": (
                        "[PERSONA: ARCHITECT]\n"
                        "- Priorizza analisi, design di sistema e decisioni strategiche\n"
                        "- Struttura l'architettura in componenti chiari e mantenibili\n"
                        "- Considera scalabilità, manutenibilità e trade-off tecnici\n"
                        "- Documenta le decisioni architetturali e il loro razionale"
                    ),
                    "reasoner": (
                        "[PERSONA: RAGIONATORE STRATEGICO]\n"
                        "- Usa ragionamento step-by-step esplicito: mostra il processo di pensiero\n"
                        "- Analizza ogni prospettiva prima di concludere\n"
                        "- Struttura la risposta: Analisi → Pro/Contro → Raccomandazione\n"
                        "- Considera le implicazioni di lungo termine delle scelte"
                    ),
                    "analyst": (
                        "[PERSONA: ANALISTA DATI]\n"
                        "- Usa Python per elaborare e analizzare dati quando disponibili\n"
                        "- Produci visualizzazioni chiare (grafici, tabelle) ove possibile\n"
                        "- Interpreta i risultati con rigore: distingui correlazione da causalità\n"
                        "- Struttura i report: Executive Summary → Metodologia → Risultati → Conclusioni\n"
                        "- Strumenti preferiti: run_python, web_search, vision"
                    ),
                }
                _persona = task.get('persona') or ''
                # P17-F5-IMPROVED: server-side classification se persona vuota/auto
                _persona_auto = False
                if not _persona:
                    _persona = _classify_persona_server(task.get('goal', ''))
                    if _persona:
                        _persona_auto = True
                        task['persona'] = _persona  # persist per history/resume
                _persona_hint = _PERSONA_HINTS.get(_persona.lower().strip(), '')
                if _persona_hint:
                    context_str = f"{_persona_hint}\n\n{context_str}".strip()
                # P17-F5: emit persona_classified SSE event — UI badge feedback
                if _persona:
                    _persona_conf = 0.85 if not _persona_auto else 0.78
                    _sse('persona_classified', {
                        'taskId':     task_id,
                        'persona':    _persona,
                        'confidence': _persona_conf,
                        'auto':       _persona_auto,
                    })
                # BG-4: inject cross-session handoff context if available
                _hctx = task.get("_handoff_context", "")
                if _hctx:
                    context_str = f"{_hctx}\n\n{context_str}".strip()
                # P17-F5: route primary LLM to persona-appropriate client
                _persona_client = _get_persona_llm_client(_persona, client)
                loop = UnifiedAgentLoop(
                    llm_client=_persona_client, critic=_critic, verifier=_verifier,
                    memory=await _get_mem_manager_async(), executor=_get_executor(), planner=_get_planner(),
                )
                step_idx = [0]
                _backend_steps: list[dict] = []  # GAP-SYNC-FIX: log step per resume preciso

                async def step_cb(step_data: dict) -> None:
                    step_idx[0] += 1
                    _action = step_data.get('action', f'Step {step_idx[0]}')
                    # S420: streaming token — emetti direttamente senza passare dal buffer step
                    if _action == 'text_chunk':
                        _sse('text_chunk', {'taskId': task_id, 'token': _ss(step_data.get('token', ''))})
                        return

                    # S363-Blueprint: Narrative Streaming — explanation lookup for ALL step_done events
                    # S376: _STEP_NARRATIONS espanso — aggiunge 12 tool mancanti
                    # Il fallback `_action.replace('_', ' ').capitalize()` è troppo generico
                    # per tool composti — narrativa esplicita migliora la UX del LiveStreamBlock
                    _STEP_NARRATIONS = {
                        'plan': 'Analisi del goal e creazione piano di azione',
                        'llm': 'Elaborazione risposta AI',
                        'fallback': 'Completamento task',
                        'smolagents': 'Esecuzione agente autonomo con strumenti',
                        'web_search': 'Cerco informazioni aggiornate sul web',
                        'read_page': 'Leggo il contenuto della pagina web',
                        'fetch_url': 'Recupero dati dall\'URL richiesto',
                        'fetch_url_content': 'Scarico il contenuto dell\'URL',
                        'run_code': 'Eseguo il codice nel sandbox',
                        'write_file': 'Scrivo il file nel progetto',
                        'read_file': 'Leggo il file dal VFS',
                        'delete_file': 'Rimuovo il file dal progetto',
                        'create_file': 'Creo il file nel progetto',
                        'list_files': 'Elenco i file del progetto',
                        'search_github': 'Cerco codice e repository su GitHub',
                        'search_github_code': 'Cerco snippet di codice su GitHub',
                        'search_wikipedia': 'Consulto Wikipedia per informazioni',
                        'get_weather': 'Recupero le previsioni meteo',
                        'get_news': 'Carico le ultime notizie',
                        'get_currency': 'Consulto il tasso di cambio',
                        'get_location': 'Rilevo la posizione geografica',
                        'calculate': 'Calcolo l\'espressione matematica',
                        'math_eval': 'Valuto l\'espressione matematica',
                        'generate_image': 'Genero l\'immagine con AI (Pollinations)',
                        'remember': 'Salvo informazioni in memoria',
                        'recall': 'Recupero informazioni dalla memoria',
                        'direct_tools': 'Utilizzo strumenti diretti',
                        'critic_retry': 'Auto-correzione risposta (Quality Gate)',
                        'execution_validator_fix': 'Auto-fix codice rilevato (ExecutionValidator)',
                        '__thinking__': 'Ragionamento interno in corso',
                        '__plan__': 'Pianificazione step successivo',
                        '__verify__': 'Verifica e validazione risposta',
                        'reflective_debug': 'Analisi root cause errore (Chain-of-Verification)',
                        'lint_result': 'Validazione sintattica file',
                        'lint_code': 'Analisi statica del codice',
                        'project_skeleton': 'Mappa aggiornata del progetto',
                        'tool_governor_skip': 'Tool già eseguito — risultato riutilizzato',
                        'severity_retry': 'Retry adattivo per tipologia errore (S376)',
                        # S-LOOP2: narrations per fasi avanzate
                        'reasoning_core':   'Ragionamento multi-step (ReasoningCore attivo)',
                        'browser_verifier':  'Verifica app live in tempo reale (Playwright)',
                    }
                    _tool_key_narr = _action.replace('executor:', '') if _action.startswith('executor:') else _action
                    _narration = _STEP_NARRATIONS.get(_tool_key_narr,
                        _action.replace('executor:', '').replace('_', ' ').capitalize())
                    # P16-B4: propaga 'truncated' dal loop (finish_reason==length) → frontend
                    _step_truncated = bool(step_data.get('truncated', False))
                    _sse('step_done', {
                        'taskId': task_id,
                        'step': {
                            'name':        _action,
                            'index':       step_idx[0],
                            'status':      step_data.get('status', 'done'),
                            'result':      str(step_data.get('result', step_data.get('output', '')))[:500],
                            'explanation': _narration,  # S363-Blueprint: narrative field
                            'truncated':   _step_truncated,  # P16-B4: segnala max_tokens raggiunto
                        },
                    })
                    # P39-UX: rileva [CONNECTOR_NEEDED:provider] nel result → emetti SSE connector_needed
                    import re as _re_cn
                    _cn_result = str(step_data.get('result', step_data.get('output', '')))
                    _cn_matches = _re_cn.findall(r'\[CONNECTOR_NEEDED:([\w]+)\]', _cn_result)
                    for _cn_prov in _cn_matches:
                        _PROVIDER_LABELS = {'github': 'GitHub', 'google': 'Google Calendar', 'instagram': 'Instagram'}
                        _cn_label = _PROVIDER_LABELS.get(_cn_prov.lower(), _cn_prov.capitalize())
                        _sse('connector_needed', {
                            'taskId':   task_id,
                            'provider': _cn_prov.lower(),
                            'label':    _cn_label,
                            'message':  f"Per completare il task ho bisogno di accedere a {_cn_label}. Connettiti con un tap.",
                        })
                    # GAP-SYNC-FIX: accumula step results per resume preciso (checkpoint backend-side)
                    _backend_steps.append({
                        'step':   step_idx[0],
                        'action': _action,
                        'result': str(step_data.get('result', step_data.get('output', '')))[:150],
                        'ok':     step_data.get('status', 'done') not in ('error', 'failed'),
                    })
                    # Ogni 2 step: persisti il log su Supabase (non saturare Supabase su loop lunghi)
                    if step_idx[0] % 2 == 0:
                        asyncio.create_task(
                            sb_save_checkpoint(task_id, step_idx[0], {
                                '_backend_steps': _backend_steps[-10:],  # ultime 10 step
                                'step':           step_idx[0],
                            })
                        ).add_done_callback(_log_task_exc)
                    # TG-STEP: notifica step intermedio rilevante (fire-and-forget, rate-limited 30s)
                    asyncio.create_task(_tg_step(task_id, _action, _narration)).add_done_callback(_log_task_exc)
                    # S362: emit vfs_update when a file operation is detected
                    # SYNC-1: file_written (da unified_loop GAP-1) incluso + content forwarding
                    _VFS_ACTIONS = ('write_file', 'file_write', 'create_file', 'delete_file', 'file_delete', 'file_written')
                    if _action in _VFS_ACTIONS or step_data.get('file_path'):
                        # S581: 120→200 — path file spesso 120-200 chars
                        # S596: 200→400 — result/output può contenere path completo di progetto
                        # S604: 400→500 — parity con altri campi step
                        # SYNC-1: file_written porta path in 'path', non 'file_path'
                        _vfs_file = (step_data.get('path') or
                                     step_data.get('file_path') or
                                     step_data.get('result', '')[:500] or
                                     step_data.get('output', '')[:500])
                        _vfs_op = 'delete' if 'delete' in _action else 'write'
                        _vfs_evt: dict = {'taskId': task_id, 'file': str(_vfs_file)[:500], 'op': _vfs_op}
                        # SYNC-1: includi content nel SSE event per file_written (≤60KB)
                        # Frontend scrive direttamente nel VFS locale senza fetch aggiuntivo
                        if _action == 'file_written' and step_data.get('content'):
                            _vfs_evt['content'] = str(step_data['content'])[:60_000]
                        _sse('vfs_update', _vfs_evt)

                    # S363-UI: thought event — emitted when planner completes
                    if _action == 'plan' and step_data.get('status') == 'done':
                        _plan_obj = step_data.get('result', step_data.get('output', ''))
                        _thought = (_plan_obj.get('goal', '') if isinstance(_plan_obj, dict) else str(_plan_obj))[:400]  # S604: 280→400
                        if _thought:
                            _sse('thought', {'taskId': task_id, 'text': _thought,
                                             'complexity': _plan_obj.get('complexity') if isinstance(_plan_obj, dict) else None})
                        # S367: plan_update — structured subtask list for live plan tracking UI
                        if isinstance(_plan_obj, dict) and _plan_obj.get('subtasks'):
                            _sse('plan_update', {
                                'taskId': task_id,
                                'subtasks': [
                                    {
                                        'id':          s.get('id', _si + 1),
                                        'description': s.get('description', '')[:200],  # S581: 80→200
                                        'tool':        s.get('tool', ''),
                                        'status':      'pending',
                                    }
                                    for _si, s in enumerate(_plan_obj['subtasks'])
                                ],
                                'goal': _plan_obj.get('goal', ''),
                            })

                    # S367: subtask_done — mark individual subtask complete for live checkbox update
                    if step_data.get('subtask_id') and step_data.get('status') == 'done':
                        _sse('plan_update', {
                            'taskId':      task_id,
                            'subtask_done': step_data['subtask_id'],
                        })

                    # S363-UI: action event — tool execution phase
                    _TOOL_EXPLAINS_S363 = {
                        'web_search': 'Cerco informazioni in rete',
                        'get_weather': 'Recupero dati meteo',
                        'get_news': 'Carico notizie recenti',
                        'search_wikipedia': 'Consulto Wikipedia',
                        'fetch_url': 'Leggo la pagina web',
                        'search_github': 'Cerco su GitHub',
                        'run_code': 'Eseguo il codice',
                        'write_file': 'Scrivo il file',
                        'read_file': 'Leggo il file',
                        'direct_tools': 'Eseguo strumenti diretti',
                    }
                    _tool_key = _action.replace('executor:', '') if _action.startswith('executor:') else _action
                    if _action.startswith('executor:') or _tool_key in _TOOL_EXPLAINS_S363:
                        _sse('action', {
                            'taskId': task_id,
                            'log': _tool_key.upper().replace('_', ' ')[:30],
                            'explain': _TOOL_EXPLAINS_S363.get(_tool_key, f'Esecuzione: {_tool_key}'),
                        })
                    # S758-P4.1: tool_use — chip pre-esecuzione (stream_agent_task path)
                    _is_pre_exec = (
                        (_action == 'tool_start' and step_data.get('status') == 'running') or
                        (_action.startswith('executor:') and step_data.get('status') == 'started')
                    )
                    if _is_pre_exec:
                        _sse('tool_use', {
                            'taskId': task_id,
                            'tool':   _tool_key,
                            'name':   _tool_key,
                            'label':  (step_data.get('title') or
                                       _TOOL_EXPLAINS_S363.get(_tool_key,
                                       _tool_key.replace('_', ' ').capitalize())),
                            'args':   {},
                        })
                    # S758-P4.1: task_thinking — chip ragionamento LLM
                    if (_action in ('__thinking__', 'reflective_debug') and
                            step_data.get('status') in ('started', 'running', 'running_deep')):
                        _sse('task_thinking', {
                            'taskId':  task_id,
                            'message': (step_data.get('explanation') or step_data.get('title') or
                                       "L’agente sta elaborando…"),
                        })

                _sse('task_start', {'taskId': task_id, 'goal': task['goal']})
                _task_started_ms = int(time.time() * 1000)  # NOTIFY-BOT: elapsed tracking
                asyncio.create_task(_tg_start(task_id, task['goal'])).add_done_callback(_log_task_exc)
                _sse('step_start', {'taskId': task_id, 'step': {'name': 'Analisi goal', 'index': 0}})

                # S364: inject project skeleton into context from VFS (Gap 4)
                if task.get('conversation_id'):
                    try:
                        from api.project_manifest import build_manifest_from_vfs, get_skeleton
                        await asyncio.wait_for(
                            build_manifest_from_vfs(task['conversation_id']),
                            timeout=3.0,
                        )
                        _skeleton = await get_skeleton(task['conversation_id'])
                        if _skeleton:
                            context_str = (_skeleton + '\n\n' + context_str).strip()
                    except Exception:
                        pass  # S364: skeleton injection is optional

                result = await loop.run(
                    goal=task['goal'],
                    context=context_str,
                    max_steps=task.get('_resume_max_steps', task.get('max_steps', 8)),  # AG-BUG-1: _resume_max mai definito in questo scope
                    on_step=step_cb,
                    session_id=task.get('session_id', '') or '',
                )
                _agent_tasks[task_id]['status'] = 'SUCCESS'
                asyncio.create_task(sb_update_status(task_id, 'SUCCESS')).add_done_callback(_log_task_exc)
                _result_text = str(result.get('output', result) if isinstance(result, dict) else result)
                _sse('task_done', {'taskId': task_id, 'result': _result_text[:8000]})
                asyncio.create_task(_tg_done(task_id, task.get('goal', ''), _result_text[:500], _task_started_ms)).add_done_callback(_log_task_exc)

                # S363: fire-and-forget quality check when code detected in output
                if _run_quality_check:
                    _qg_result = str(result.get('output', result) if isinstance(result, dict) else result)
                    if len(_qg_result) > 500 and _qg_result.count('```') >= 2:  # S373: threshold raised — evita QG su snippet brevi
                        asyncio.create_task(_run_quality_check(
                            task_id, task['goal'], _qg_result,
                            on_event=lambda ev: _sse(ev.get('type', 'test_result'), ev),
                        )).add_done_callback(_log_task_exc)


            except asyncio.CancelledError:
                _agent_tasks[task_id]['status'] = 'CANCELLED'
                asyncio.create_task(sb_update_status(task_id, 'CANCELLED')).add_done_callback(_log_task_exc)
                _sse('task_cancelled', {'taskId': task_id})

            except (ImportError, ModuleNotFoundError):
                _agent_tasks[task_id]['status'] = 'SUCCESS'
                asyncio.create_task(sb_update_status(task_id, 'SUCCESS')).add_done_callback(_log_task_exc)
                _sse('step_done', {'taskId': task_id, 'step': {'name': 'Ragionamento', 'index': 0}})
                _sse('task_done', {'taskId': task_id, 'result': (
                    f'Goal ricevuto: {task["goal"]}\n\n'
                    'Il backend non ha il modulo agents.unified_loop. '
                    'Configura HuggingFace Spaces con smolagents per l\'esecuzione autonoma.'
                )})

            except Exception as err:
                _agent_tasks[task_id]['status'] = 'ERROR'
                asyncio.create_task(sb_update_status(task_id, 'ERROR')).add_done_callback(_log_task_exc)
                _logger.error('[agent/stream] %s error: %s', task_id, err, exc_info=True)
                _sse('task_error', {'taskId': task_id, 'error': str(err)[:1000]})
                asyncio.create_task(_tg_error(task_id, task.get('goal', ''), str(err))).add_done_callback(_log_task_exc)

            finally:
                reg_entry['done'] = True
                reg_entry['finished_at'] = time.time()
                for q in list(reg_entry['subscriber_queues']):
                    try:
                        q.put_nowait(None)
                    except Exception as _exc:
                        _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001

        reg_entry['asyncio_task'] = asyncio.create_task(run_loop())

        try:
            while True:
                if _agent_tasks.get(task_id, {}).get('status') == 'CANCELLED':
                    at = reg_entry.get('asyncio_task')
                    if at and not at.done():
                        at.cancel()
                    break
                try:
                    item = await asyncio.wait_for(sub_q.get(), timeout=15.0)
                    if item is None:
                        break
                    yield item
                except asyncio.TimeoutError:
                    yield ': heartbeat\n\n'
        finally:
            try:
                reg_entry['subscriber_queues'].remove(sub_q)
            except ValueError as _exc:
                _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001

        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type='text/event-stream',
        headers={
            'Cache-Control':     'no-cache',
            'X-Accel-Buffering': 'no',
            'Connection':        'keep-alive',
        },
    )


# ── Task checkpoints ───────────────────────────────────────────────────────────

class CheckpointIn(BaseModel):
    taskId:     str
    step:       int
    goal:       str
    plan:       list[str] = []
    logs:       list[str] = []
    artifacts:  list[str] = []
    retryCount: int = 0
    extra:      dict = {}


@router.post('/api/agent/tasks/{task_id}/checkpoint')
async def save_checkpoint(task_id: str, body: CheckpointIn):
    _prune_checkpoints()
    _task_checkpoints[task_id] = {
        'taskId':     task_id,
        'step':       body.step,
        'goal':       body.goal,
        'plan':       body.plan,
        'logs':       body.logs[-50:],
        'artifacts':  body.artifacts,
        'retryCount': body.retryCount,
        'extra':      body.extra,
        'savedAt':    int(time.time() * 1000),
    }
    asyncio.create_task(sb_save_checkpoint(task_id, _task_checkpoints[task_id])).add_done_callback(_log_task_exc)
    return {'saved': True, 'taskId': task_id, 'step': body.step}


@router.get('/api/agent/tasks/{task_id}/checkpoint')
async def get_checkpoint(task_id: str):
    _prune_checkpoints()
    cp = _task_checkpoints.get(task_id)
    if not cp:
        cp = await sb_get_checkpoint(task_id)
    if not cp:
        raise HTTPException(404, detail={'error': 'checkpoint_not_found', 'taskId': task_id})
    return cp


@router.delete('/api/agent/tasks/{task_id}/checkpoint')
async def delete_checkpoint(task_id: str):
    _task_checkpoints.pop(task_id, None)
    return {'deleted': task_id}


@router.get('/api/agent/checkpoints')
async def list_checkpoints():
    _prune_checkpoints()
    now = int(time.time() * 1000)
    return {
        'count': len(_task_checkpoints),
        'checkpoints': [
            {'taskId': k, 'step': v['step'], 'goal': v['goal'][:300], 'age_ms': now - v['savedAt']}  # S606: 200→300
            for k, v in _task_checkpoints.items()
        ],
    }


# ─── Sprint 5 ITEM 15: /debug/timing — telemetria timing + qualità agente ────
# Usato da TelemetryDashboard.tsx (frontend) per la sezione "Qualità agente".
# Espone: timing_stats (avg/count per fase) + repair_stats (contatori qualità).
# Non richiede auth — dati aggregati, nessun dato sensibile.
@router.get('/debug/timing')
async def get_debug_timing():
    """
    Espone timing breakdown per fase (classify/plan/coder/verifier/browser)
    e contatori qualità (goal_success, repair_success, tool_failure, req_engine).
    Formato: { timing_stats: {label: {avg, count}}, repair_stats: {key: count} }
    """
    try:
        from api.state import _TIMING_STORE, _REPAIR_STATS
        timing_stats: dict = {}
        for label, samples in _TIMING_STORE.items():
            if samples:
                avg_val = round(sum(samples) / len(samples), 1)
            else:
                avg_val = None
            timing_stats[label] = {"avg": avg_val, "count": len(samples)}
        return {
            "timing_stats": timing_stats,
            "repair_stats": dict(_REPAIR_STATS),
        }
    except Exception as exc:
        return {"timing_stats": {}, "repair_stats": {}, "error": str(exc)}


# ─── GAP-SKILL-SYNC: /api/agent/skill-stats — statistiche tool adattive ──────
# Espone i dati del SkillTracker (session-scoped success/fail per tool)
# al frontend per merge con skillRegistry Dexie — vista cross-runtime unificata.
@router.get('/api/agent/skill-stats/{session_id}')
async def get_skill_stats(session_id: str):
    """Success/fail rate + Wilson score per ogni tool nella sessione.

    Il frontend usa questa API per arricchire i dati Dexie di skillRegistry.ts
    con le stats backend: confidence reale (server-side) vs contatori browser-only.
    """
    try:
        from agents.skill_tracker import get_skill_tracker
        return {
            "session_id": session_id,
            "stats":      get_skill_tracker().get_stats(session_id),
        }
    except Exception as exc:
        return {"session_id": session_id, "stats": {}, "error": str(exc)}


@router.get('/api/agent/skill-stats')
async def list_all_skill_sessions():
    """Debug: panoramica di tutte le sessioni SkillTracker attive (tool count, call count)."""
    try:
        from agents.skill_tracker import get_skill_tracker
        return get_skill_tracker().get_all_sessions()
    except Exception as exc:
        return {"error": str(exc)}

# ── /api/agent/circuit-status/{session_id} — circuit breaker live status ──────
# Espone per ogni tool tracciato in sessione: stato circuito, Wilson score,
# recovery calls effettuate — utile per debug e monitoring real-time.
@router.get('/api/agent/circuit-status/{session_id}')
async def get_circuit_status(session_id: str):
    """
    Stato real-time del circuit breaker per ogni tool di una sessione.

    Per ogni tool tracciato, classifica il circuito come:
      - open   → Wilson score < 0.15 AND total_count >= 3 AND tool ha fallback
                 (il tool viene bypassato — routing automatico ai fallback)
      - closed → performance sufficiente o dati insufficienti per aprire il circuit

    Campi per tool:
      wilson_score:     lower bound dell'intervallo di confidenza al 95% (0–1)
      success_count:    successi registrati nella sessione
      fail_count:       fallimenti registrati nella sessione
      total_count:      chiamate totali
      success_rate:     raw rate (NON usato dal circuit — solo informativo)
      avg_latency_ms:   latenza media (ms)
      has_fallbacks:    True se TOOL_REGISTRY definisce fallback per il tool
      recovery_calls:   quante volte il recovery credit ha concesso un tentativo
      circuit_state:    "open" | "closed" | "no_data" | "insufficient_data"

    Thresholds (from executor.py):
      circuit_open_threshold: 0.15   (Wilson score sotto cui il circuit si apre)
      min_calls_for_circuit:  3      (chiamate minime prima che il circuit possa aprirsi)
      recovery_interval:      5      (ogni N call con circuit open → recovery attempt)
    """
    try:
        from agents.skill_tracker import get_skill_tracker
        from tools.registry import TOOL_REGISTRY
        from api.state import _get_executor
        from agents.executor import (
            _CIRCUIT_OPEN_THRESHOLD,
            _MIN_CALLS_FOR_CIRCUIT,
            _RECOVERY_INTERVAL,
        )

        stats = get_skill_tracker().get_stats(session_id)

        # Recovery counts vivono nell'istanza Executor singleton
        executor    = _get_executor()
        rec_counts: dict = {}
        if executor is not None:
            rec_counts = getattr(executor, '_circuit_recovery_counts', {})

        circuits_open: list[dict]   = []
        circuits_closed: list[dict] = []

        for tool_name, s in stats.items():
            has_fallbacks = bool(TOOL_REGISTRY.get(tool_name, {}).get('fallbacks'))
            recovery_calls = rec_counts.get(tool_name, 0)

            # Replica logica _is_circuit_open() di executor.py
            if s['total_count'] == 0:
                state = 'no_data'
            elif s['total_count'] < _MIN_CALLS_FOR_CIRCUIT:
                state = 'insufficient_data'
            elif s['wilson_score'] < _CIRCUIT_OPEN_THRESHOLD and has_fallbacks:
                state = 'open'
            else:
                state = 'closed'

            entry = {
                'tool':            tool_name,
                'circuit_state':   state,
                'wilson_score':    s['wilson_score'],
                'success_count':   s['success_count'],
                'fail_count':      s['fail_count'],
                'total_count':     s['total_count'],
                'success_rate':    s['success_rate'],
                'avg_latency_ms':  s['avg_latency_ms'],
                'has_fallbacks':   has_fallbacks,
                'recovery_calls':  recovery_calls,
            }
            if state == 'open':
                circuits_open.append(entry)
            else:
                circuits_closed.append(entry)

        # Ordina open per Wilson score asc (peggiori prima), closed per desc (migliori prima)
        circuits_open.sort(key=lambda x: x['wilson_score'])
        circuits_closed.sort(key=lambda x: x['wilson_score'], reverse=True)

        return {
            'session_id':            session_id,
            'total_tools_tracked':   len(stats),
            'circuits_open_count':   len(circuits_open),
            'circuits_closed_count': len(circuits_closed),
            'circuits_open':         circuits_open,
            'circuits_closed':       circuits_closed,
            'thresholds': {
                'circuit_open_threshold': _CIRCUIT_OPEN_THRESHOLD,
                'min_calls_for_circuit':  _MIN_CALLS_FOR_CIRCUIT,
                'recovery_interval':      _RECOVERY_INTERVAL,
            },
        }
    except Exception as exc:
        return {
            'session_id':          session_id,
            'total_tools_tracked': 0,
            'circuits_open_count': 0,
            'circuits_open':       [],
            'circuits_closed':     [],
            'error':               str(exc),
        }