"""unified_loop_delegate.py — DelegateMixin: debug riflessivo, replan, delega in-loop. Estratto da unified_loop.py per ridurre il file principale. Contiene: _reflective_debug(goal, errors): BGAP-GUARD diagnosi breve da errori tool _budget_replan_check(state, step): BGAP-1 replan probabilistico su budget critico _DELEGATE_RESEARCH_RE: regex riconoscimento sub-goal tipo ricerca _run_in_loop_delegate(sub_goal): GAP-1 micro-agente specializzato in-loop Invariante B1: nessun corpo duplicato con unified_loop.py. MRO garantisce che DelegateMixin._budget_replan_check sovrascriva HelpersMixin (DelegateMixin precede HelpersMixin nella lista basi di UnifiedAgentLoop). """ from __future__ import annotations import asyncio import logging import re from typing import Any from agents.unified_loop_types import StepCallback, UnifiedLoopState, _maybe_await _logger = logging.getLogger("agente_ai") class DelegateMixin: async def _reflective_debug( self, goal: str = "", errors: Any = None, **kwargs: Any ) -> str: """Reflective debug: analizza errori e propone diagnosi in max 2 frasi. Chiamato dopo tool failures per arricchire state.context con ipotesi fix. Fail-open: non blocca mai il loop in caso di errore LLM.""" try: _ctx = f"Goal: {str(goal)[:200]}\nErrori: {'; '.join(str(e)[:300] for e in (errors if isinstance(errors, list) else [errors])[:3])}" # S573: 150→300 _fast = self._get_fast_llm() _diag = await asyncio.wait_for( _fast.chat([{"role": "user", "content": f"Diagnosi breve (max 2 frasi):\n{_ctx}"}], max_tokens=300), # S586: 120->180->300 timeout=5.0, ) return (str(_diag) if _diag else "").strip()[:300] except Exception: pass # fail-open return "" # ── BGAP-1: Probabilistic Re-planning Trigger ──────────────────────────── async def _budget_replan_check( self, state: Any, step_count: int, on_step: Any = None ) -> str: """BGAP-1: probabilistic re-planning trigger. Guards: skip se _n_err < 2 OR _budget_ratio < 0.6. Usa _get_fast_llm() con max_tokens=120. Fail-open.""" _n_err = len(state.errors) if getattr(state, 'errors', None) else 0 if _n_err < 2: return '' _budget_ratio = step_count / max(state.max_steps, 1) if _budget_ratio < 0.6: return '' # dedup guard [GAP-1-REPLAN]: skip se già replanned in questo loop if '[GAP-1-REPLAN]' in (state.context or ''): return '' try: _fast_llm = self._get_fast_llm() _prompt = ( f'Task ha avuto {_n_err} errori e usato {_budget_ratio:.0%} del budget. ' f'Suggerisci UN approccio alternativo in max 2 frasi. Goal: {state.goal[:500]}' # S597: 200->300->500 ) _hint = await asyncio.wait_for( _fast_llm.chat([{'role': 'user', 'content': _prompt}], max_tokens=120), timeout=5.0, ) return (str(_hint) if _hint else '').strip()[:200] except Exception: pass # fail-open totale return '' # ── GAP-1: Delega Dinamica In-Loop ───────────────────────────────────── _DELEGATE_RESEARCH_RE = re.compile( r'\b(cerca|research|trova|web|url|leggi|analisi|analizza|documenta|' r'news|notizie|fetch|scrape|pagina|sito|http)\b', re.IGNORECASE, ) async def _run_in_loop_delegate(self, sub_goal: str, timeout: float = 40.0) -> dict: """GAP-1: Delega Dinamica In-Loop. Lancia un micro-agente specializzato per sub_goal DURANTE il loop principale. Architettura: - Stesso executor del parent → accesso ai tool reali (write_file, run_python, ...) - LLM selezionato per ruolo → RESEARCHER, CODER o REASONER in base al goal - _is_delegate_child = True → blocca ricorsione (max 1 livello di delega) - max_steps = 4 → micro-agente leggero, non un loop completo - output troncato a 4000 chars → evita context-window explosion nel parent """ # P18: defensive anti-recursion guard at entry point if getattr(self, '_is_delegate_child', False): _logger.debug("[delegate] anti-recursion guard triggered at _run_in_loop_delegate entry") return {"output": "[DELEGATE] Ricorsione bloccata: _is_delegate_child=True.", "steps": [], "goal_met": False} try: from models.role_router import RoleRouter as _RR_d, Role as _Role_d # Seleziona LLM specializzato in base al tipo di sotto-obiettivo if self._DELEGATE_RESEARCH_RE.search(sub_goal[:300]): _sub_llm = _RR_d.get_client(_Role_d.RESEARCHER) # Gemini 2.5-flash elif self._CODE_RE.search(sub_goal[:300]): _sub_llm = _RR_d.get_client(_Role_d.CODER) # Llama 4 Scout else: _sub_llm = _RR_d.get_client(_Role_d.REASONER) # Cerebras 120B except Exception: _sub_llm = self.llm # fallback: usa LLM del parent # Crea loop figlio: stessi executor/planner/memory, LLM specializzato _sub_loop = UnifiedAgentLoop( llm_client=_sub_llm, planner=self.planner, executor=self.executor, critic=None, # no critic — micro-agente leggero memory=self.memory, verifier=None, # no verifier — massima velocità ) # Anti-ricorsione: il figlio non può delegare ulteriormente _sub_loop._is_delegate_child = True # Propaga session_id per isolare sandbox backend-exec _sub_loop._run_task_id = self._run_task_id + "_d" # GAP-6: condividi dict mutabile _session_files con il parent loop # Prima: delegate inizializzava _session_files={} -> file scritti non visibili al parent # Ora: stessa referenza -> parent vede automaticamente tutti i file scritti dal delegate _sub_loop._session_files = self._session_files # P17-F1: buffer output parziale via on_step — sopravvive al timeout _partial_steps: list[dict] = [] async def _capture_partial(step: dict) -> None: if step.get("output") or step.get("explanation"): _partial_steps.append(step) try: _res = await asyncio.wait_for( _sub_loop.run(sub_goal, max_steps=4, on_step=_capture_partial), timeout=timeout, ) _out = (_res.get("output") or "")[:4000] _logger.info( "GAP-1 delegate OK [%s] steps=%d: %s", _res.get("engine", "?"), len(_res.get("steps", [])), sub_goal[:60], ) return { "success": _res.get("success", False), "output": _out, "engine": _res.get("engine", "delegate"), "steps": len(_res.get("steps", [])), } except asyncio.TimeoutError: # P17-F1: esponi stato parziale invece di stringa vuota # _session_files già condiviso con parent → parent vede file scritti _partial_files = list(getattr(_sub_loop, "_session_files", {}).keys()) _partial_out = " ".join( (s.get("output") or s.get("explanation") or "")[:300] for s in _partial_steps[-3:] ).strip()[:1500] _logger.warning( "GAP-1 delegate timeout (%.0fs, %d steps, %d files): %s", timeout, len(_partial_steps), len(_partial_files), sub_goal[:60], ) # S-PARTIAL: emetti evento SSE partial_output al frontend PRIMA di restituire # così l'utente vede il chip "⚠ output parziale — riprendo" in tempo reale if on_step: await _maybe_await(on_step({ "event": "partial_output", "action": "partial_output", "visibility": "progress", "partial": True, "steps_done": len(_partial_steps), "partial_files": _partial_files, "partial_output": _partial_out, "output": _partial_out, "explanation": f"Output parziale dopo {timeout:.0f}s — l'agente sta recuperando", "status": "warning", })) return { "success": False, "output": _partial_out, "error": f"delegate timeout ({timeout:.0f}s) — risultato parziale", "partial": True, "partial_files": _partial_files, "steps_done": len(_partial_steps), } except Exception as _de: _logger.warning("GAP-1 delegate error: %s", _de) return {"success": False, "output": "", "error": str(_de)[:200]} # ── S362: Role routing helpers ───────────────────────────────────────────── # S427: ampliato con verbi IT/EN mancanti + framework/pattern aggiuntivi. # Stesso set di goal_verifier._CODE_RE + keyword tecnologiche per routing CODER LLM.