api/agent_task_routes.py

"""agent_task_routes.py — Task CRUD: crea, lista, cancella, stato, stream SSE.

Estratto da agent.py (split 2026-06-30).
Route coperte:
  POST   /api/agent/tasks
  GET    /api/agent/tasks
  DELETE /api/agent/tasks/{task_id}
  GET    /api/agent/tasks/{task_id}/status
  GET    /api/agent/tasks/{task_id}/stream   (SSE principale, S359 persist)
"""
from __future__ import annotations
import os, asyncio, json, uuid, time, re
import re as _re_persona
from fastapi import APIRouter, HTTPException, Request, Body
from fastapi.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,
)
from .speculative import fire_speculative_tools
from .prewarm import fire_predictive_prewarm  # S950
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,
)
from ._agent_helpers import (
    _RE_SURROGATES, _ss, _log_task_exc,
    _PERSONA_KEYWORD_MAP, _PERSONA_CLIENT_CACHE,
    _build_persona_kw_map, _classify_persona_server, _get_persona_llm_client,
)
router = APIRouter()

@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)
    # S950: Predictive Pre-warming cluster 4x4
    fire_predictive_prewarm(body.goal)
    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',
        }
    )