ajarbot/self_healing.py

"""
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}")