core/auto_debug.py

"""
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<file>[^"]+)", line (?P<line>\d+).*?\n(?P<error_type>\w+(?:Error|Exception)?):\s*(?P<message>.*)',
        "node":    r'(?P<error_type>\w+(?:Error|Exception)?):\s*(?P<message>.*)\s+at\s+(?P<file>[^:]+):(?P<line>\d+)',
        "generic": r'(?P<error_type>\w+):\s*(?P<message>.*)',
    }

    # 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"]