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