150 lines
4.8 KiB
Python
150 lines
4.8 KiB
Python
"""
|
|
MARK XL — Error Handler
|
|
Replaces google.generativeai with local Ollama via core.llm_client.
|
|
"""
|
|
import json
|
|
import re
|
|
import sys
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
|
|
from core.llm_client import call_llm_text
|
|
|
|
|
|
def get_base_dir() -> Path:
|
|
if getattr(sys, "frozen", False):
|
|
return Path(sys.executable).parent
|
|
return Path(__file__).resolve().parent.parent
|
|
|
|
|
|
BASE_DIR = get_base_dir()
|
|
|
|
|
|
class ErrorDecision(Enum):
|
|
RETRY = "retry"
|
|
SKIP = "skip"
|
|
REPLAN = "replan"
|
|
ABORT = "abort"
|
|
|
|
|
|
ERROR_ANALYST_PROMPT = """You are the error recovery module of MARK XL AI assistant.
|
|
|
|
A task step has failed. Analyze the error and decide what to do.
|
|
|
|
DECISIONS:
|
|
- retry : Transient error (network timeout, temporary file lock, race condition).
|
|
- skip : This step is not critical and the task can succeed without it.
|
|
- replan : The approach was wrong. A different tool or method should be tried.
|
|
- abort : The task is fundamentally impossible or unsafe to continue.
|
|
|
|
Return ONLY valid JSON:
|
|
{
|
|
"decision": "retry|skip|replan|abort",
|
|
"reason": "why it failed",
|
|
"fix_suggestion": "what to try instead (for replan)",
|
|
"max_retries": 1,
|
|
"user_message": "Short message to tell the user (max 15 words)"
|
|
}
|
|
"""
|
|
|
|
|
|
def analyze_error(
|
|
step: dict,
|
|
error: str,
|
|
attempt: int = 1,
|
|
max_attempts: int = 2,
|
|
) -> dict:
|
|
if attempt >= max_attempts:
|
|
print(f"[ErrorHandler] ⚠️ Max attempts for step {step.get('step')} — forcing replan")
|
|
return {
|
|
"decision": ErrorDecision.REPLAN,
|
|
"reason": f"Failed {attempt} times: {error[:100]}",
|
|
"fix_suggestion": "Try a completely different approach or tool",
|
|
"max_retries": 0,
|
|
"user_message": "Trying a different approach, sir.",
|
|
}
|
|
|
|
prompt = f"""Failed step:
|
|
Tool: {step.get('tool')}
|
|
Description: {step.get('description')}
|
|
Parameters: {json.dumps(step.get('parameters', {}), indent=2)}
|
|
Critical: {step.get('critical', False)}
|
|
|
|
Error:
|
|
{error[:500]}
|
|
|
|
Attempt number: {attempt}"""
|
|
|
|
try:
|
|
text = call_llm_text(prompt, system=ERROR_ANALYST_PROMPT)
|
|
text = re.sub(r"```(?:json)?", "", text).strip().rstrip("`").strip()
|
|
result = json.loads(text)
|
|
|
|
decision_str = result.get("decision", "replan").lower()
|
|
decision_map = {
|
|
"retry": ErrorDecision.RETRY,
|
|
"skip": ErrorDecision.SKIP,
|
|
"replan": ErrorDecision.REPLAN,
|
|
"abort": ErrorDecision.ABORT,
|
|
}
|
|
result["decision"] = decision_map.get(decision_str, ErrorDecision.REPLAN)
|
|
|
|
if step.get("critical") and result["decision"] == ErrorDecision.SKIP:
|
|
result["decision"] = ErrorDecision.REPLAN
|
|
result["user_message"] = "This step is critical — finding alternative approach, sir."
|
|
|
|
print(f"[ErrorHandler] Decision: {result['decision'].value} — {result.get('reason', '')}")
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"[ErrorHandler] ⚠️ Analysis failed: {e} — defaulting to replan")
|
|
return {
|
|
"decision": ErrorDecision.REPLAN,
|
|
"reason": str(e),
|
|
"fix_suggestion": "Try alternative approach",
|
|
"max_retries": 1,
|
|
"user_message": "Encountered an issue, adjusting approach, sir.",
|
|
}
|
|
|
|
|
|
def generate_fix(step: dict, error: str, fix_suggestion: str) -> dict:
|
|
prompt = f"""A task step failed. Generate a replacement step.
|
|
|
|
Original step:
|
|
Tool: {step.get('tool')}
|
|
Description: {step.get('description')}
|
|
Parameters: {json.dumps(step.get('parameters', {}), indent=2)}
|
|
|
|
Error: {error[:300]}
|
|
Fix suggestion: {fix_suggestion}
|
|
|
|
Write a Python script that accomplishes the same goal differently.
|
|
Return ONLY the Python code, no explanation."""
|
|
|
|
try:
|
|
code = call_llm_text(prompt)
|
|
code = re.sub(r"```(?:python)?", "", code).strip().rstrip("`").strip()
|
|
return {
|
|
"step": step.get("step"),
|
|
"tool": "code_helper",
|
|
"description": f"Auto-fix for: {step.get('description')}",
|
|
"parameters": {
|
|
"action": "run",
|
|
"description": fix_suggestion,
|
|
"code": code,
|
|
"language": "python",
|
|
},
|
|
"depends_on": step.get("depends_on", []),
|
|
"critical": step.get("critical", False),
|
|
}
|
|
except Exception as e:
|
|
print(f"[ErrorHandler] ⚠️ Fix generation failed: {e}")
|
|
return {
|
|
"step": step.get("step"),
|
|
"tool": "generated_code",
|
|
"description": f"Fallback for: {step.get('description')}",
|
|
"parameters": {"description": step.get("description", "")},
|
|
"depends_on": step.get("depends_on", []),
|
|
"critical": step.get("critical", False),
|
|
}
|