Update app/agent/self_heal.py

app/agent/self_heal.py

@@ -1,10 +1,12 @@
 import json
 
 
-def check_and_fix(result, agent):
-
+# =====================================================
+# LEGACY FUNCTION (Backward Compatibility)
+# =====================================================
+def check_and_fix(result, agent=None):
     """
-    If error → ask AI to fix and retry
+    Legacy fallback healer (old system support)
     """
 
     if isinstance(result, dict) and "error" in result:
@@ -18,12 +20,114 @@ Error:
 Fix it and return corrected tool call JSON.
 """
 
-        fixed = agent.run(fix_prompt)
+        if agent:
+            fixed = agent.run(fix_prompt)
+
+            try:
+                tool_call = json.loads(fixed)
+                return tool_call
+            except:
+                return None
+
+    return None
+
+
+# =====================================================
+# MODERN SELF-HEALER CLASS (STEP 5.1 CORE)
+# =====================================================
+class SelfHealer:
+
+    def __init__(self, llm=None):
+        self.llm = llm
+        self.failure_history = []
+
+    # =========================
+    # DETECT FAILURE
+    # =========================
+    def is_failure(self, result: dict):
+        if not result:
+            return True
+
+        if isinstance(result, dict):
 
-        try:
-            tool_call = json.loads(fixed)
-            return tool_call
-        except:
+            if result.get("exit_code", 0) != 0:
+                return True
+
+            if "error" in result:
+                return True
+
+            if result.get("stderr"):
+                return True
+
+        return False
+
+    # =========================
+    # ANALYZE ERROR
+    # =========================
+    def analyze(self, result: dict, step: str):
+
+        error_text = (
+            result.get("stderr")
+            or result.get("error")
+            or "unknown error"
+        )
+
+        prompt = f"""
+You are a debugging and repair system.
+
+A tool execution failed.
+
+STEP:
+{step}
+
+ERROR:
+{error_text}
+
+Return ONLY a corrected tool call JSON if fix is possible:
+
+Example:
+{{
+  "tool": "run_shell",
+  "args": {{
+    "cmd": "fixed command"
+  }}
+}}
+
+If cannot fix → return:
+{{"fix": null}}
+"""
+
+        if self.llm:
+            try:
+                response = self.llm([
+                    {"role": "user", "content": prompt}
+                ])
+
+                content = response["choices"][0]["message"]["content"]
+
+                return json.loads(content)
+
+            except:
+                return {"fix": None}
+
+        return {"fix": None}
+
+    # =========================
+    # MAIN HEALING PIPELINE
+    # =========================
+    def heal(self, step: str, result: dict):
+
+        # no failure → skip
+        if not self.is_failure(result):
             return None
 
-    return None
+        # store failure
+        self.failure_history.append({
+            "step": step,
+            "result": result
+        })
+
+        # analyze + generate fix
+        analysis = self.analyze(result, step)
+
+        return analysis.get("fix")