""" dynamic_replanner.py — COG-1: Dynamic Re-planner on subtask failure. Quando il loop accumula >= 1 subtask falliti con errori reali, genera un NUOVO piano con il contesto degli errori iniettato nel goal. Architettura: - should_replan(): decision gate — zero latency, no LLM - replan(): chiama planner.create_plan() con failure context - Max 1 re-plan per run (flag _replanned=True nel piano restituito) - Timeout 20s; fallback: None → usa piano originale Integration: chiamato da unified_loop.py dopo il gather dei subtask se exec_warn contiene fallimenti reali (non solo risk:high skips). """ from __future__ import annotations import asyncio import logging import re _logger = logging.getLogger("agente_ai.replanner") _FAILURE_RE = re.compile( r"(timeout|error|errore|fallito|failed|exception|not found|non trovato" r"|AttributeError|TypeError|RuntimeError|ImportError|KeyError" r"|404|500|503|ECONNREFUSED|ConnectionError|ModuleNotFoundError)", re.IGNORECASE, ) # P26-B2: pattern errori transienti — non triggerano replan (si risolvono da soli) _TRANSIENT_RE = re.compile( r"(429|rate.?limit|too many requests|connection.?reset|connection.?refused" r"|network.*timeout|read.*timeout|ssl.*timeout|temporary.*unavailable" r"|service.*unavailable|overloaded|quota.*exceeded)", re.IGNORECASE, ) # Pattern per fallimenti critici strutturali (richiedono replan immediato) _CRITICAL_RE = re.compile( r"(ImportError|ModuleNotFoundError|SyntaxError|TypeError|AttributeError" r"|PermissionError|AssertionError|not found|ECONNREFUSED)", re.IGNORECASE, ) def should_replan(exec_warn: list[str], exec_done: list[str]) -> bool: """ Decision gate: decide se vale la pena re-pianificare. Trigger se: - Almeno 1 warning contiene pattern di failure reale (non solo skip risk:high) - exec_done ha meno successi dei fallimenti (piano non sta funzionando) P26-B2: errori transienti (429/RateLimit/timeout di rete) NON triggerano replan — si risolvono da soli e il replan sarebbe un falso positivo costoso. """ if not exec_warn: return False real_failures = [w for w in exec_warn if _FAILURE_RE.search(w)] if not real_failures: return False # P26-B2: se TUTTI i fallimenti sono transienti → no replan, lascia retry naturale transient = [w for w in real_failures if _TRANSIENT_RE.search(w)] if transient and len(transient) == len(real_failures): _logger.debug("P26-B2 should_replan=False: tutti i %d fallimenti sono transienti", len(transient)) return False # REASONING-BUG-6: singolo fallimento critico strutturale → re-plan immediato critical_failures = [w for w in real_failures if _CRITICAL_RE.search(w)] if critical_failures: return True # ImportError / SyntaxError / AttributeError → replan subito # Re-plan se fallimenti strutturali >= successi (piano non sta funzionando) structural = [w for w in real_failures if not _TRANSIENT_RE.search(w)] return len(structural) >= max(len(exec_done), 1) def _find_downstream(subtasks: list, done_descs: set) -> tuple: """P25-R1: dato il grafo requires[], ritorna (done_ids, pending_ids). - done_ids : subtask già completati (matched by description in done_descs) - pending_ids: subtask non ancora completati (da includere nel re-plan) Logica: matching fuzzy description→done_descs (substring 40 char). Pure function, zero I/O, zero LLM — usata solo per filtrare il re-plan scope. """ done_ids: set = set() for st in subtasks: desc = str(st.get("description", ""))[:40].lower() if any(desc and desc in d.lower() for d in done_descs): done_ids.add(st.get("id")) pending = [st for st in subtasks if st.get("id") not in done_ids] return done_ids, pending async def replan( planner: object, original_goal: str, exec_warn: list[str], exec_done: list[str], error_context: str = "", plan: "dict | None" = None, ) -> "dict | None": """ Genera un nuovo piano con il contesto dei fallimenti iniettato nel goal. Il goal arricchito contiene: - Subtask già completati (da NON ripetere) - Problemi riscontrati (ultimi 3 warning) - Analisi errore classificata (se disponibile) Returns: nuovo piano dict con _replanned=True, o None se fallisce. """ if not planner: return None # P25-R1: graph-aware scope — se abbiamo il piano corrente, replan solo i subtask # pendenti (non quelli già completati). Riduce il re-plan al sottoinsieme necessario. _scope_hint = "" if plan and plan.get("subtasks"): _done_descs = set(exec_done) _, _pending = _find_downstream(plan["subtasks"], _done_descs) if _pending and len(_pending) < len(plan["subtasks"]): _ids = [st.get("id") for st in _pending] _scope_hint = f"\nRe-pianifica SOLO i subtask {_ids} (gli altri sono già completati)." _logger.debug("P25-R1 scope ridotto: %d/%d subtask da replanare", len(_pending), len(plan["subtasks"])) failures_str = "\n".join(exec_warn[-3:]) if exec_warn else "nessun dettaglio" done_str = ", ".join(exec_done[-5:]) if exec_done else "nessuno" # P16-B6: estrai tool/approcci falliti — guida il replanner a evitarli # P18: rimosso import re lazy — usa re module-level (già importato riga 20) _tool_fails: list[str] = [] for _w in exec_warn[-5:]: _m = re.search( r"(web_search|run_python|write_file|read_file|web_fetch|" r"trigger_webhook|pip_install|shell_exec|delegate)\w*", _w, re.IGNORECASE, ) if _m: _tool_fails.append(_m.group(0)) _avoid_str = ", ".join(set(_tool_fails)) if _tool_fails else "" enriched_goal = ( f"{original_goal}\n\n" f"[CONTESTO RE-PLAN \u2014 tentativo precedente fallito]\n" f"Subtask gi\u00e0 completati (NON ripetere): {done_str}.\n" f"Problemi riscontrati:\n{failures_str}\n" ) if _avoid_str: enriched_goal += f"Tool che hanno fallito (usa ALTERNATIVE): {_avoid_str}.\n" if error_context: enriched_goal += f"Analisi errore: {error_context[:300]}\n" if _scope_hint: enriched_goal += _scope_hint _avoid_hint = f"Evita: {_avoid_str}. " if _avoid_str else "" enriched_goal += ( "[ISTRUZIONE] Genera un piano ALTERNATIVO che eviti gli stessi problemi. " f"{_avoid_hint}" "Usa approcci diversi per i subtask falliti. " "Se un tool ha fallito, usa un tool alternativo." ) try: new_plan = await asyncio.wait_for( planner.create_plan(enriched_goal), # type: ignore[attr-defined] timeout=20.0, ) if new_plan and new_plan.get("subtasks"): _logger.info( "COG-1 replan: %d subtask nel nuovo piano (da %d warn, %d done)", len(new_plan["subtasks"]), len(exec_warn), len(exec_done), ) new_plan["_replanned"] = True return new_plan except asyncio.TimeoutError: _logger.warning("COG-1 replan timeout 20s — mantengo piano originale") except Exception as exc: _logger.warning("COG-1 replan error: %s", exc) return None