""" core/auto_debug.py — parse → focused fix → re-test loop (max 3 attempts). v4.6 additions -------------- * Queries :mod:`core.patch_memory` for previously-failed patches that match the current error signature; results are injected into the fix prompt as "known-bad" guidance so the coder is nudged away from repeating them. * On loop exhaustion, records the final failed patch back into patch memory so future runs benefit from the same signal. """ from __future__ import annotations import logging import re from datetime import datetime from typing import Any, Dict, List, Optional try: from core.patch_memory import get_default_memory except Exception: # pragma: no cover — memory is optional get_default_memory = None # type: ignore[assignment] class AutoDebugLoop: MAX_AUTO_DEBUG_ATTEMPTS = 3 ERROR_PATTERNS = { "python": r'File "(?P[^"]+)", line (?P\d+).*?\n(?P\w+(?:Error|Exception)?):\s*(?P.*)', "node": r'(?P\w+(?:Error|Exception)?):\s*(?P.*)\s+at\s+(?P[^:]+):(?P\d+)', "generic": r'(?P\w+):\s*(?P.*)', } # v4.6 — a lightweight fingerprint for lint-only failures so patch memory # can distinguish "test failed" from "lint rejected the patch". _LINT_MARKER = "[static-analysis] lint gate failed" def __init__(self, memory=None) -> None: # Injected for tests; falls back to the module-level singleton so # existing call sites (``AutoDebugLoop()``) keep working unchanged. self._memory = memory # ------------------------------------------------------------------ # Memory access (lazy so tests can monkey-patch the module) # ------------------------------------------------------------------ def _get_memory(self): if self._memory is not None: return self._memory if get_default_memory is None: return None try: return get_default_memory() except Exception as exc: # pragma: no cover — never fatal logging.warning("patch_memory unavailable: %s", exc) return None # ------------------------------------------------------------------ # Error parsing # ------------------------------------------------------------------ def parse_error(self, log: str, language: str = "python") -> Dict[str, Any]: pat = self.ERROR_PATTERNS.get(language, self.ERROR_PATTERNS["generic"]) m = re.search(pat, log or "", re.DOTALL) if m: g = m.groupdict() line_s = g.get("line", "0") diag = { "error_type": g.get("error_type", "Unknown"), "file_path": g.get("file", "unknown"), "line_number": int(line_s) if str(line_s).isdigit() else 0, "message": (g.get("message") or "").strip(), "raw_log": log, } else: diag = { "error_type": "Unknown", "file_path": "unknown", "line_number": 0, "message": (log or "")[:200], "raw_log": log, } if log and self._LINT_MARKER in log: # Tag lint-only failures so patch memory can filter them later. diag["error_type"] = diag.get("error_type") or "LintError" diag["tags"] = ["lint"] return diag # ------------------------------------------------------------------ # Fix request (with patch-memory advisory injection) # ------------------------------------------------------------------ def request_fix(self, diag: Dict[str, Any], coder, state: Dict[str, Any] ) -> Optional[str]: advisory = "" memory = self._get_memory() if memory is not None: try: matches = memory.query_similar(diag) if matches: advisory = "\n\n" + memory.format_advisory(matches) + "\n\n" state.setdefault("patch_memory_matches", []).extend(matches) except Exception as exc: # pragma: no cover — memory is advisory logging.warning("patch_memory query failed: %s", exc) instruction = ( f"URGENT FIX: file `{diag['file_path']}` line {diag['line_number']}\n" f"Error: {diag['error_type']}: {diag['message']}\n" f"{advisory}" f"Provide ONLY a minimal Unified Diff patch (or replacement code " f"in a fenced block) fixing this exact error. " f"Do NOT rewrite the entire file." ) try: new_state = coder.run(state.copy(), instruction=instruction) return new_state.get("generated_code") except Exception as exc: logging.error("AutoDebug coder call failed: %s", exc) return None # ------------------------------------------------------------------ # Main loop # ------------------------------------------------------------------ def run(self, error_log: str, coder, tester, state: Dict[str, Any], language: str = "python") -> Dict[str, Any]: timeline: List[Dict[str, Any]] = [] current = error_log last_patch: Optional[str] = None last_diag: Dict[str, Any] = {} for attempt in range(1, self.MAX_AUTO_DEBUG_ATTEMPTS + 1): diag = self.parse_error(current, language) last_diag = diag timeline.append({ "timestamp": datetime.now().strftime("%H:%M:%S"), "agent": "AutoDebug", "status": "running", "message": f"attempt {attempt}/{self.MAX_AUTO_DEBUG_ATTEMPTS}: " f"fixing {diag['error_type']}", }) patch = self.request_fix(diag, coder, state) if not patch: timeline.append({ "timestamp": datetime.now().strftime("%H:%M:%S"), "agent": "AutoDebug", "status": "failed", "message": f"attempt {attempt}: coder returned empty", }) continue last_patch = patch state["generated_code"] = patch state["file_path"] = diag["file_path"] prev_flag = state.get("auto_debug_enabled", True) state["auto_debug_enabled"] = False try: test_state = tester.run(state) finally: state["auto_debug_enabled"] = prev_flag if test_state.get("test_passed"): timeline.append({ "timestamp": datetime.now().strftime("%H:%M:%S"), "agent": "AutoDebug", "status": "success", "message": f"attempt {attempt}: fixed", }) return { "success": True, "attempts": attempt, "final_logs": test_state.get("test_logs", ""), "timeline_entries": timeline, } current = test_state.get("test_logs", current) # ------------------------------------------------------------------ # Exhausted → record final failed patch into memory for next time. # ------------------------------------------------------------------ memory = self._get_memory() if memory is not None and last_patch: try: stored = memory.record_failure( diagnostic=last_diag, patch_text=last_patch, attempts=self.MAX_AUTO_DEBUG_ATTEMPTS, tags=list(last_diag.get("tags", []) or []) + ["autodebug-exhausted"], extra={ "language": language, "command": state.get("command"), }, ) state.setdefault("patch_memory_recorded", []).append(stored) timeline.append({ "timestamp": datetime.now().strftime("%H:%M:%S"), "agent": "PatchMemory", "status": "OK", "message": f"recorded failed patch {stored.get('id')}", }) except Exception as exc: # pragma: no cover — never fatal timeline.append({ "timestamp": datetime.now().strftime("%H:%M:%S"), "agent": "PatchMemory", "status": "FAIL", "message": f"record failed: {exc}", }) timeline.append({ "timestamp": datetime.now().strftime("%H:%M:%S"), "agent": "AutoDebug", "status": "failed", "message": f"all {self.MAX_AUTO_DEBUG_ATTEMPTS} attempts failed", }) return { "success": False, "attempts": self.MAX_AUTO_DEBUG_ATTEMPTS, "final_logs": current, "timeline_entries": timeline, } __all__ = ["AutoDebugLoop"]