diff --git "a/api/agent.py" "b/api/agent.py" new file mode 100644--- /dev/null +++ "b/api/agent.py" @@ -0,0 +1,2019 @@ +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 + + +def _sanitize_for_json(obj: object) -> object: + """BUG-SSE-SURR: ricorsivamente applica _ss() su valori stringa prima + di json.dumps() — previene UnicodeEncodeError da surrogati Groq/LLM.""" + if isinstance(obj, str): + return _ss(obj) + if isinstance(obj, dict): + return {k: _sanitize_for_json(v) for k, v in obj.items()} + if isinstance(obj, list): + return [_sanitize_for_json(x) for x in obj] + return obj + +from fastapi import APIRouter, Depends, HTTPException, Request, Body +from fastapi.responses import StreamingResponse +from .auth_guard import require_role, AuthRole +from pydantic import BaseModel, TypeAdapter, ValidationError, field_validator +from typing import Literal + +# P0 JSON contract: expose structured JSON only after schema validation. +_STRUCTURED_JSON_ADAPTER = TypeAdapter(dict[str, object] | list[object]) + +def _validated_response_json(output: object) -> dict[str, object] | list[object] | None: + if not isinstance(output, str) or not output.strip(): + return None + try: + parsed = json.loads(output) + return _STRUCTURED_JSON_ADAPTER.validate_python(parsed) + except (json.JSONDecodeError, ValidationError, TypeError, ValueError): + return None + +from .state import ( + _agent_tasks, _task_ai_clients, _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, +) +from .speculative import fire_speculative_tools +from .vfs_sync import build_vfs_sync_complete +from .task_tool_policy import build_task_tool_policy +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 +) +try: + from .kernel import kernel as _kernel # ARCH-K2.2: Brain→Kernel abstraction + _KERNEL_AVAILABLE = True +except Exception: + _kernel = None # type: ignore[assignment] + _KERNEL_AVAILABLE = False + +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() + + +class _LLMAdmission: + """Bounded in-process admission control for LLM loops.""" + def __init__(self) -> None: + self.max_concurrency = max(1, int(os.getenv('LLM_MAX_CONCURRENCY', '2'))) + self.max_queue = max(0, int(os.getenv('LLM_MAX_QUEUE', '8'))) + self._condition = asyncio.Condition() + self._active = 0 + self._queued = 0 + + async def reserve(self) -> bool: + async with self._condition: + if self._active + self._queued >= self.max_concurrency + self.max_queue: + return False + self._queued += 1 + return True + + async def acquire(self) -> None: + async with self._condition: + try: + while self._active >= self.max_concurrency: + await self._condition.wait() + self._queued = max(0, self._queued - 1) + self._active += 1 + except BaseException: + self._queued = max(0, self._queued - 1) + self._condition.notify(1) + raise + + async def cancel(self) -> None: + async with self._condition: + self._queued = max(0, self._queued - 1) + self._condition.notify(1) + + async def release(self) -> None: + async with self._condition: + self._active = max(0, self._active - 1) + self._condition.notify(1) + + async def snapshot(self) -> dict[str, int]: + async with self._condition: + return { + 'active': self._active, + 'queued': self._queued, + 'max_concurrency': self.max_concurrency, + 'max_queue': self.max_queue, + } + + +_LLM_ADMISSION = _LLMAdmission() + + +async def _record_phase(phase: str, duration_ms: float = 0.0, + outcome: str = 'ok', error_class: str | None = None) -> None: + try: + from .agent_telemetry import record_runtime_phase + await record_runtime_phase(phase, duration_ms, outcome, error_class) + except Exception: + pass + + +def _queue_full_error() -> HTTPException: + return HTTPException( + status_code=429, + detail={ + 'error': 'llm_queue_full', + 'message': 'Coda LLM satura. Riprova tra pochi secondi.', + 'retry_after_seconds': 5, + }, + headers={'Retry-After': '5'}, + ) + + +@router.get('/api/agent/queue-status') +async def agent_queue_status(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): + return await _LLM_ADMISSION.snapshot() + + +def _attach_byok_client(task_id: str, credentials: object) -> None: + """Create a task-scoped LLM client without persisting or logging credentials.""" + if credentials is None: + return + try: + runtime_config = credentials.as_runtime_config() + if not runtime_config: + return + from models.ai_client import AIClient + _task_ai_clients[task_id] = AIClient(byok_credentials=runtime_config) + except Exception as exc: + # Never include the payload or credential values in diagnostics. + _logger.warning("[agent] unable to initialise BYOK task client: %s", type(exc).__name__) + + +# ── Deprecated run_loop ──────────────────────────────────────────────────────── + +@router.post('/run_loop', deprecated=True) +async def run_loop(): + """Deprecated — use /agent/task instead.""" + from fastapi import HTTPException + raise HTTPException(status_code=410, detail="run_loop is deprecated. Use /agent/task.") + + +# ─── 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, + role: AuthRole = Depends(require_role(AuthRole.MACHINE)), # GAP-1-fix: era fail-open +): + await _record_phase('auth') + if not await _LLM_ADMISSION.reserve(): + await _record_phase('queue', outcome='rejected', error_class='queue_full') + raise _queue_full_error() + await _record_phase('queue') + async def generate(): + queue: asyncio.Queue = asyncio.Queue() + + async def step_cb(step: dict) -> None: + if str(step.get('action', '')).startswith(('tool', 'executor:')): + await _record_phase('tool', outcome='ok') + 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() + _provider_started = time.perf_counter() + 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 _record_phase('provider', (time.perf_counter() - _provider_started) * 1000) + await queue.put({ + '__done__': True, + 'result': result.get('output', ''), + 'engine': result.get('engine', 'fallback'), + 'success': result.get('success', False), + 'response_json': _validated_response_json(result.get('output', '')), + }) + 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)}) + + async def _admitted_run_loop(): + await _LLM_ADMISSION.acquire() + try: + await run_loop() + finally: + await _LLM_ADMISSION.release() + task = asyncio.create_task(_admitted_run_loop()) + task.add_done_callback(_log_task_exc) # BUG-CB-1 + task_id = str(uuid.uuid4()) + # 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_error', 'taskId': task_id, 'error': f'Nessun provider AI disponibile al momento ({_names}). Riprova tra qualche minuto.'})}\n\n" + 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_error', 'error': 'stream timeout'})}\n\n" + break + yield 'data: {"type":"ping"}\n\n' + continue + # ABORT-2: segnale abort dall'endpoint POST /api/agent/abort + if "__abort__" in item: + yield f"data: {json.dumps({'type': 'task_aborted', 'taskId': task_id})}\n\n" + 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': _ss(str(item.get('title', _rs_tool.replace('_', ' ').capitalize())))})}\n\n" # BUG-SSE-SURR + if '__done__' in item: + _final_res = _ss(item.get('result', '')) + # FIX-SILENZIO: output vuoto + fallimento → messaggio esplicito invece del silenzio + if not _final_res and not item.get('success', True): + _err_detail = _ss(item.get('error', '')) + _final_res = (f"\u26a0\ufe0f {_err_detail}" if _err_detail + else "\u26a0\ufe0f Tutti i provider AI sono temporaneamente indisponibili (rate limit). Riprova tra qualche minuto.") + _done_payload = { + 'type': 'task_done', 'taskId': task_id, 'result': _final_res, + 'engine': item.get('engine', 'fallback'), 'success': item.get('success', False), + } + if item.get('response_json') is not None: + _done_payload['response_json'] = _sanitize_for_json(item['response_json']) + yield f"data: {json.dumps(_done_payload)}\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': _sanitize_for_json(item), 'taskId': task_id})}\n\n" # BUG-SSE-SURR + 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", "Connection": "keep-alive"}) + + +# ── Reason loop / Unified loop ───────────────────────────────────────────────── + +@router.post('/api/reason/loop') +async def reason_loop(body: ReasonLoopIn, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + 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 + }) + if not await _LLM_ADMISSION.reserve(): + raise _queue_full_error() + await _LLM_ADMISSION.acquire() + _provider_started = time.perf_counter() + try: + 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 "") + except Exception as exc: + await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000, 'error', type(exc).__name__) + raise + else: + await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000) + finally: + await _LLM_ADMISSION.release() + if isinstance(result, dict): + output_text = result.get('output', '') or '' + engine_used = result.get('engine', 'unknown') + errors_list = result.get('errors', []) + else: + output_text = str(result) + engine_used = 'unknown' + errors_list = [] + response_json = _validated_response_json(output_text) + 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 + **({'response_json': response_json} if response_json is not None else {}), + } + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix: alias che chiama reason_loop direttamente — Depends NON si propagano + """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(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + 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 + # ARCH-K2.2: registra il dispatch nella Queue del Kernel prima di triggerare GitHub Actions + if _KERNEL_AVAILABLE and _kernel is not None: + _dispatch_id = str(uuid.uuid4()) + asyncio.create_task(_kernel.submit_task( + payload={ + 'type': 'gh_dispatch', + 'goal': goal, + 'mode': mode, + 'dispatch_id': _dispatch_id, + 'metadata': {'workflow': 'agent-kernel.yml'}, + }, + priority='HIGH', + )).add_done_callback(_log_task_exc) + asyncio.create_task(_kernel.publish_event( + topic='agent.kernel.dispatched', + payload={'goal': goal[:200], 'mode': mode}, + )).add_done_callback(_log_task_exc) + 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) ────────────────────────────────── + +async def _create_task_internal(task_id: str, goal: str, job: dict) -> dict: + """ + Versione interna di create_agent_task per uso da job_queue (GAP-1-fix). + Non richiede FastAPI body né dipendenze auth — chiamabile direttamente. + """ + _prune_agent_tasks() + if task_id in _agent_tasks: + return {"taskId": task_id, "status": _agent_tasks[task_id]["status"]} + created_at = int(time.time() * 1000) + _agent_tasks[task_id] = { + "id": task_id, + "status": "QUEUED", + "goal": goal, + "context": job.get("context", {}), + "max_steps": job.get("max_steps", 20), + "created_at": created_at, + "session_id": job.get("session_id", ""), + } + asyncio.create_task( + sb_upsert_task(task_id, goal, "QUEUED", job.get("max_steps", 20), job.get("context", {}), created_at) + ).add_done_callback(_log_task_exc) + if _KERNEL_AVAILABLE and _kernel is not None: + asyncio.create_task(_kernel.submit_task( + payload={ + "task_id": task_id, + "goal": goal, + "max_steps": job.get("max_steps", 20), + "source": "job_queue", + "metadata": {"job_queue": True}, + }, + priority="NORMAL", + session_id=job.get("session_id", ""), + )).add_done_callback(_log_task_exc) + return {"taskId": task_id, "status": "QUEUED"} + + +@router.post('/api/agent/tasks') +async def create_agent_task(body: AgentTaskIn, request: Request, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """ + 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. + """ + await _record_phase('auth') + _prune_agent_tasks() + raw_idempotency_key = (body.idempotency_key or request.headers.get('Idempotency-Key') or '').strip() + if len(raw_idempotency_key) > 200: + raise HTTPException(400, detail={'error': 'invalid_idempotency_key'}) + task_id = body.taskId or ( + str(uuid.uuid5(uuid.NAMESPACE_URL, f'baida98-ai:{body.session_id or ""}:{raw_idempotency_key}')) + if raw_idempotency_key else str(uuid.uuid4()) + ) + _idempotency_claimed = False + if raw_idempotency_key and task_id not in _agent_tasks: + # Claim before the first await: concurrent retries in this worker see the + # same task immediately and cannot create a second provider loop/write. + _agent_tasks[task_id] = { + 'id': task_id, 'status': 'CREATING', + 'idempotency_key': raw_idempotency_key, + } + _idempotency_claimed = True + + # Already in memory → return immediately (normal path, includes S358 reconnect) + if task_id in _agent_tasks and not _idempotency_claimed: + if task_id not in _task_ai_clients: + _attach_byok_client(task_id, body.provider_credentials) + 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 + _attach_byok_client(task_id, body.provider_credentials) + return {'taskId': task_id, 'status': restored['status'], 'restored': True} + + # Brand new task. La policy è calcolata al confine HTTP, prima di ogni + # tool speculativo, pianificazione o chiamata al loop. + _tool_policy = build_task_tool_policy(body.goal) + 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→'') + 'idempotency_key': raw_idempotency_key, + 'forbid_tools': _tool_policy.forbid_tools, + 'literal_response': _tool_policy.literal_response, + 'allow_local_csv_conversion': _tool_policy.allow_local_csv_conversion, + } + # Le credenziali BYOK restano in una mappa runtime separata dai metadata task + # e non raggiungono Supabase, checkpoint o buffer SSE. + _attach_byok_client(task_id, body.provider_credentials) + # 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) + await _record_phase('persistence') + # S361: gli strumenti speculativi sono consentiti solo quando il messaggio + # utente non li vieta esplicitamente. La policy è fail-closed per questo task. + if not _tool_policy.forbid_tools: + asyncio.create_task(fire_speculative_tools(task_id, body.goal)).add_done_callback(_log_task_exc) + # ARCH-K2.2: registra il task nella Queue del Kernel e pubblica evento task.created + if _KERNEL_AVAILABLE and _kernel is not None: + asyncio.create_task(_kernel.submit_task( + payload={ + 'task_id': task_id, + 'goal': body.goal, + 'max_steps': body.max_steps, + 'persona': body.persona, + 'source': 'agent_api', + 'metadata': {'agent_api': True}, + }, + priority='NORMAL', + session_id=body.session_id, + )).add_done_callback(_log_task_exc) + asyncio.create_task(_kernel.publish_event( + topic='task.created', + payload={'task_id': task_id, 'goal': body.goal[:200], 'status': 'QUEUED'}, + )).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 = '', role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """ + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + if task_id in _agent_tasks: + _agent_tasks[task_id]['status'] = 'CANCELLED' + _task_ai_clients.pop(task_id, None) + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """ + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """ + 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. + """ + await _record_phase('auth') + # 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] + # I task restaurati da persistenza potrebbero non contenere metadata runtime. + # Ricostruire la policy dal goal mantiene il resume fail-closed. + if ( + "forbid_tools" not in task + or "literal_response" not in task + or "allow_local_csv_conversion" not in task + ): + _restored_policy = build_task_tool_policy(task.get("goal", "")) + task["forbid_tools"] = _restored_policy.forbid_tools + task["literal_response"] = _restored_policy.literal_response + task["allow_local_csv_conversion"] = _restored_policy.allow_local_csv_conversion + _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" + + # Contratto letterale: chiusura immediata prima di task_start, planner o + # tool. È il backstop per client SSE che non applicano il fast path UI. + _literal_response = task.get('literal_response') + if isinstance(_literal_response, str) and _literal_response: + _agent_tasks[task_id]['status'] = 'COMPLETED' + asyncio.create_task(sb_update_status(task_id, 'COMPLETED')).add_done_callback(_log_task_exc) + literal_event = json.dumps(_sanitize_for_json({ + 'event': 'task_done', 'taskId': task_id, 'result': _literal_response, + })) + yield f"data: {literal_event}\n\n" + yield "data: [DONE]\n\n" + return + + 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 ('COMPLETED', '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 ────────────────────────────────────────────── + if not await _LLM_ADMISSION.reserve(): + await _record_phase('queue', outcome='rejected', error_class='queue_full') + _agent_tasks[task_id]['status'] = 'RATE_LIMITED' + _rate_limited = json.dumps(_sanitize_for_json({ + 'event': 'task_error', 'taskId': task_id, + 'statusCode': 429, 'retryAfter': 5, + 'error': 'llm_queue_full', + })) + yield f'data: {_rate_limited}\n\n' + yield 'data: [DONE]\n\n' + return + await _record_phase('queue') + _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(_sanitize_for_json({'event': event, **data}))}\n\n" # BUG-SSE-SURR + # 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) + # ARCH-K2.2: pubblica lifecycle event via Kernel + if _KERNEL_AVAILABLE and _kernel is not None: + asyncio.create_task(_kernel.publish_event( + topic='task.running', + payload={'task_id': task_id, 'status': 'RUNNING'}, + )).add_done_callback(_log_task_exc) + _prune_agent_tasks() + + async def run_loop() -> None: + try: + # Contratto letterale: nessun provider, planner, tool, card o side effect. + # È emesso direttamente nello stream affinché i client SSE non possano bypassarlo. + _literal_response = task.get('literal_response') + if isinstance(_literal_response, str) and _literal_response: + _agent_tasks[task_id]['status'] = 'COMPLETED' + asyncio.create_task(sb_update_status(task_id, 'COMPLETED')).add_done_callback(_log_task_exc) + _sse('task_done', {'taskId': task_id, 'result': _literal_response}) + return + + from agents.unified_loop import UnifiedAgentLoop + # Ogni task BYOK usa il suo client effimero; gli altri mantengono + # il singleton runtime. Le credenziali non entrano nel task dict. + client = _task_ai_clients.get(task_id) or _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 + # ARTIFACT-CONTRACT: le richieste di pagina/app richiedono un + # artifact reale, non una risposta narrativa. Il frontend committa + # solo file_written + vfs_sync_complete, quindi l'agente deve + # scrivere e rileggere almeno un file prima di dichiarare successo. + _artifact_goal = bool(re.search( + r"\b(?:pagina|sito|website|webapp|app|mini[- ]app|ui|interfaccia)\b", + task.get('goal', ''), re.IGNORECASE, + )) + if _artifact_goal: + _artifact_hint = ( + "[CONTRATTO ARTIFACT UI]\n" + "Per questa richiesta devi creare davvero i file nel workspace usando write_file " + "(non limitarti a proporre codice in chat). Dopo la scrittura, usa read_file o " + "un controllo equivalente per verificare il contenuto. Dichiara completamento " + "solo se la scrittura è riuscita; in caso contrario segnala l'errore senza inventare " + "un link o un'anteprima." + ) + context_str = f"{_artifact_hint}\n\n{context_str}".strip() + # 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() + # I router persona dipendono dalle env del backend: con BYOK il + # client per task resta autorevole in ogni fase, incluso il planner. + _is_byok_task = task_id in _task_ai_clients + _persona_client = client if _is_byok_task else _get_persona_llm_client(_persona, client) + _planner = _get_planner() + if _is_byok_task: + try: + from agents.planner import Planner + _planner = Planner(llm_client=client) + except Exception as exc: + _logger.warning("[agent] BYOK planner fallback: %s", type(exc).__name__) + loop = UnifiedAgentLoop( + llm_client=_persona_client, critic=_critic, verifier=_verifier, + memory=await _get_mem_manager_async(), executor=_get_executor(), planner=_planner, + ) + step_idx = [0] + _backend_steps: list[dict] = [] # GAP-SYNC-FIX: log step per resume preciso + # P34/SYNC-1: i file arrivano nel frontend in staging; raccogliamo + # soltanto quelli con contenuto così il commit atomico può avvenire + # una sola volta dopo un task riuscito. + _vfs_written_paths: set[str] = set() + # Accumula soltanto i token realmente emessi per poter riconciliare + # il testo in streaming con l'output autorevole del finalizer. + _streamed_chunks: list[str] = [] + + 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': + _token = _ss(step_data.get('token', '')) + if _token: + _streamed_chunks.append(_token) + _sse('text_chunk', {'taskId': task_id, 'token': _token}) + return + # RECOV-P1: engineering_state event — forward projection to frontend + if _action == 'engineering_state': + _sse('engineering_state', { + 'taskId': task_id, + 'status': step_data.get('status'), + 'mode': step_data.get('mode'), + 'engineering_state': step_data.get('engineering_state'), + }) + 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). + # Per immagini dirette il backend non serializza byte nell'SSE: + # inoltra soltanto l'URL HTTPS generato internamente, che il client + # materializza come data URI nel proprio VFS prima del commit atomico. + if _action == 'file_written' and step_data.get('content'): + _vfs_evt['content'] = str(step_data['content'])[:60_000] + _vfs_written_paths.add(str(_vfs_file)[:500]) + elif _action == 'file_written': + _source_url = str(step_data.get('source_url') or '') + _mime_type = str(step_data.get('mime_type') or '') + _allowed_image_origins = ( + 'https://image.pollinations.ai/', + 'https://media.pollinations.ai/', + 'https://gen.pollinations.ai/', + ) + if (_source_url.startswith(_allowed_image_origins) and + _mime_type in {'image/jpeg', 'image/png', 'image/webp'}): + _vfs_evt['sourceUrl'] = _source_url[:2_000] + _vfs_evt['mimeType'] = _mime_type + _vfs_written_paths.add(str(_vfs_file)[:500]) + _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 + + _provider_started = time.perf_counter() + 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 '', + allow_tools=not bool(task.get('forbid_tools', False)), + allow_local_csv_conversion=bool(task.get('allow_local_csv_conversion', False)), + ) + await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000) + # ARTIFACT-CONTRACT-FALLBACK: alcuni provider restituiscono codice HTML + # nella risposta finale dopo aver narrato una scrittura, senza produrre + # il tool event `file_written`. Non committiamo mai una falsa positività: + # se il goal è UI, proviamo solo a materializzare un documento HTML + # completo già presente nell'output, usando l'executor VFS autenticato. + # Se non troviamo un documento completo o la scrittura fallisce, il task + # resta senza artifact e il frontend non riceve alcun sync inventato. + if _artifact_goal and not _vfs_written_paths and getattr(loop, 'executor', None): + _artifact_output = str(result.get('output', result) if isinstance(result, dict) else result) + _html_match = re.search(r'(?is)()', _artifact_output) + if not _html_match: + _html_match = re.search(r'(?is)(]*)?>.*?)', _artifact_output) + if _html_match: + _artifact_content = _html_match.group(1).strip() + _artifact_path = 'index.html' + _path_match = re.search(r'(?i)(?:/|\b)([\w.-]+\.html)\b', _artifact_output) + if _path_match: + _artifact_path = _path_match.group(1) + try: + _write_result = await asyncio.wait_for( + loop.executor.run_tool('write_file', { + 'path': _artifact_path, + 'content': _artifact_content, + }), + timeout=25.0, + ) + _write_ok = not (isinstance(_write_result, dict) and _write_result.get('error')) + if _write_ok: + _vfs_written_paths.add(_artifact_path) + await step_cb({ + 'action': 'file_written', 'status': 'done', + 'path': _artifact_path, 'content': _artifact_content, + 'result': f'Artifact materializzato: {_artifact_path}', + }) + _read_result = await asyncio.wait_for( + loop.executor.run_tool('read_file', {'path': _artifact_path}), + timeout=15.0, + ) + _read_ok = not (isinstance(_read_result, dict) and _read_result.get('error')) + await step_cb({ + 'action': 'read_file', + 'status': 'done' if _read_ok else 'error', + 'path': _artifact_path, + 'result': f'Verifica artifact: {_artifact_path}' if _read_ok else str(_read_result)[:300], + }) + except Exception as _artifact_exc: + _logger.warning('[agent] artifact fallback failed: %s', type(_artifact_exc).__name__) + + # Lifecycle separato: l'esecuzione primaria è completa quando VFS e risultato + # sono confermati; la verifica qualità successiva non deve tenere il task RUNNING. + _agent_tasks[task_id]['status'] = 'COMPLETED' + _agent_tasks[task_id]['quality_status'] = ( + 'QUALITY_CHECK_PENDING' if _run_quality_check else 'NOT_REQUIRED' + ) + asyncio.create_task(sb_update_status(task_id, 'COMPLETED')).add_done_callback(_log_task_exc) + # ARCH-K2.2: pubblica lifecycle event via Kernel + if _KERNEL_AVAILABLE and _kernel is not None: + asyncio.create_task(_kernel.publish_event( + topic='task.completed', + payload={'task_id': task_id, 'status': 'SUCCESS'}, + )).add_done_callback(_log_task_exc) + _result_text = str(result.get('output', result) if isinstance(result, dict) else result) + _vfs_commit = build_vfs_sync_complete( + task_id, _vfs_written_paths, result, + ) + if _vfs_commit is not None: + # P34: completa l’atomic swap frontend solo dopo che il loop ha + # confermato il task. Su errore/cancellazione lo staging rimane + # intenzionalmente non committato. + _sse('vfs_sync_complete', _vfs_commit) + if _run_quality_check: + _sse('quality_check_pending', { + 'taskId': task_id, + 'status': 'QUALITY_CHECK_PENDING', + 'primaryStatus': 'COMPLETED', + }) + _streamed_text = ''.join(_streamed_chunks) + if _streamed_text and _result_text != _streamed_text: + # Il finalizer può riparare/normalizzare la risposta dopo gli + # ultimi chunk. Notifica esplicitamente la sostituzione così il + # client non mostra un testo parziale o divergente. + _sse('text_replace', { + 'taskId': task_id, + 'text': _result_text[:8000], + 'reason': 'authoritative_finalizer_output', + }) + _sse('task_done', { + 'taskId': task_id, + 'result': _result_text[:8000], + 'status': 'COMPLETED', + 'qualityStatus': _agent_tasks[task_id].get('quality_status', 'NOT_REQUIRED'), + }) + asyncio.create_task(_tg_done(task_id, task.get('goal', ''), _result_text[:500], _task_started_ms)).add_done_callback(_log_task_exc) + + # Quality check non bloccante: aggiorna solo quality_status, mai lo stato + # primario del task e mai il commit VFS già confermato. + 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: + async def _run_quality_background() -> None: + try: + await _run_quality_check( + task_id, task['goal'], _qg_result, + on_event=lambda ev: _sse(ev.get('type', 'test_result'), ev), + ) + _agent_tasks.get(task_id, {})['quality_status'] = 'QUALITY_CHECK_DONE' + _sse('quality_check_done', { + 'taskId': task_id, + 'status': 'QUALITY_CHECK_DONE', + 'primaryStatus': 'COMPLETED', + }) + except Exception as _q_exc: + _agent_tasks.get(task_id, {})['quality_status'] = 'QUALITY_CHECK_ERROR' + _sse('quality_check_done', { + 'taskId': task_id, + 'status': 'QUALITY_CHECK_ERROR', + 'primaryStatus': 'COMPLETED', + 'error': type(_q_exc).__name__, + }) + asyncio.create_task(_run_quality_background()).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) + # ARCH-K2.2: pubblica lifecycle event via Kernel + if _KERNEL_AVAILABLE and _kernel is not None: + asyncio.create_task(_kernel.publish_event( + topic='task.cancelled', + payload={'task_id': task_id, 'status': 'CANCELLED'}, + )).add_done_callback(_log_task_exc) + _sse('task_cancelled', {'taskId': task_id}) + + except (ImportError, ModuleNotFoundError): + _agent_tasks[task_id]['status'] = 'COMPLETED' + asyncio.create_task(sb_update_status(task_id, 'COMPLETED')).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) + # ARCH-K2.2: pubblica lifecycle event via Kernel + if _KERNEL_AVAILABLE and _kernel is not None: + asyncio.create_task(_kernel.publish_event( + topic='task.failed', + payload={'task_id': task_id, 'status': 'ERROR', 'error': str(err)[:500]}, + )).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: + # Il task è terminale: rimuove l’unico riferimento alle credenziali + # BYOK, lasciando replay SSE e metadati senza segreti. + _task_ai_clients.pop(task_id, None) + 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 + + async def _admitted_run_loop(): + await _LLM_ADMISSION.acquire() + try: + await run_loop() + finally: + await _LLM_ADMISSION.release() + reg_entry['asyncio_task'] = asyncio.create_task(_admitted_run_loop()) + reg_entry['asyncio_task'].add_done_callback(_log_task_exc) # BUG-CB-2 + + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + _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, body.step, _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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + _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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + _task_checkpoints.pop(task_id, None) + return {'deleted': task_id} + + +@router.get('/api/agent/checkpoints') +async def list_checkpoints(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + _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(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix: internal diagnostics + """ + 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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """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(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """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, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix + """ + 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), + } + +@router.post('/api/agent/abort') +async def abort_agent_task(taskId: str = Body(..., embed=True), role: AuthRole = Depends(require_role(AuthRole.MACHINE))): + """ + ABORT-ENDPOINT: Permette di cancellare un task in streaming (run-stream). + Inserisce un segnale __abort__ nella coda del task, che verrà catturato + dal loop SSE per interrompere l'esecuzione e pulire le risorse. + """ + if taskId in _run_stream_tasks: + queue = _run_stream_tasks[taskId].get("queue") + if queue: + await queue.put({"__abort__": True}) + return {"ok": True, "message": f"Abort signal sent to task {taskId}"} + return {"ok": False, "message": "Task not found or already finished"}