""" Self-Healing System - Phase 1: Error Capture and Logging. Captures all errors with full context and logs them to MEMORY.md. No auto-fixing in this phase - observation only. """ import hashlib import json import traceback from dataclasses import dataclass from datetime import datetime from typing import Any, Dict, Optional @dataclass class ErrorContext: """Full context for a captured error.""" error_type: str # Exception class name message: str # Error message stack_trace: str # Full traceback component: str # Where it happened (e.g., "tools.py:read_file") intent: str # What was being attempted context: Dict[str, Any] # Additional context (tool inputs, user message, etc.) timestamp: str # ISO 8601 format class SelfHealingSystem: """ Phase 1: Error observation infrastructure. Captures errors with full context, deduplicates via error signatures, and logs them to MEMORY.md for future analysis. """ def __init__(self, memory_system: Any, agent: Any) -> None: self.memory = memory_system self.agent = agent self._error_counts: Dict[str, int] = {} def capture_error( self, error: Exception, component: str, intent: str, context: Optional[Dict[str, Any]] = None, ) -> None: """Capture an error with full context and log it. Args: error: The exception that occurred. component: Where the error happened (e.g., "tools.py:read_file"). intent: What was being attempted when the error occurred. context: Additional context such as tool inputs, user message, etc. """ error_ctx = ErrorContext( error_type=type(error).__name__, message=str(error), stack_trace=traceback.format_exc(), component=component, intent=intent, context=context or {}, timestamp=datetime.now().isoformat(), ) signature = self._generate_signature(error_ctx) # Track attempt count self._error_counts[signature] = self._error_counts.get(signature, 0) + 1 attempt = self._error_counts[signature] if attempt <= 3: self._log_error(error_ctx, attempt) print( f"[SelfHealing] Error captured: {error_ctx.error_type} " f"in {error_ctx.component} (attempt {attempt}/3)" ) def _generate_signature(self, error_ctx: ErrorContext) -> str: """Generate a deduplication signature for an error. Uses first 8 characters of SHA-256 hash of error type, component, and message combined. """ raw = f"{error_ctx.error_type}:{error_ctx.component}:{error_ctx.message}" return hashlib.sha256(raw.encode()).hexdigest()[:8] def _log_error(self, error_ctx: ErrorContext, attempt: int) -> None: """Log an error to MEMORY.md via the memory system. Formats the error as a markdown entry and appends it to the persistent MEMORY.md file (daily=False). """ # Serialize context to JSON for readability try: context_json = json.dumps(error_ctx.context, indent=2, default=str) except (TypeError, ValueError): context_json = str(error_ctx.context) # Format timestamp for the header try: dt = datetime.fromisoformat(error_ctx.timestamp) header_time = dt.strftime("%Y-%m-%d %H:%M:%S") except ValueError: header_time = error_ctx.timestamp log_entry = ( f"## Error Log - {header_time}\n" f"\n" f"**Type**: {error_ctx.error_type}\n" f"**Component**: {error_ctx.component}\n" f"**Intent**: {error_ctx.intent}\n" f"**Attempt**: {attempt}/3\n" f"**Message**: {error_ctx.message}\n" f"\n" f"**Context**:\n" f"```json\n" f"{context_json}\n" f"```\n" f"\n" f"**Stack Trace**:\n" f"```\n" f"{error_ctx.stack_trace}\n" f"```\n" f"---" ) try: self.memory.write_memory(log_entry, daily=False) except Exception as e: # Last resort: print to console if memory write fails print(f"[SelfHealing] Failed to write error log to MEMORY.md: {e}") print(f"[SelfHealing] Error was: {error_ctx.error_type}: {error_ctx.message}")