"""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', } )