history_store.py

"""
History Store for ComfyUI-Doctor.

Provides persistent storage for error analysis history using JSON files.
Supports cross-restart history retrieval and automatic cleanup.
"""

import json
import os
import threading
import shutil
import hashlib
from dataclasses import dataclass, field, asdict
from typing import Optional, List, Dict, Any
from datetime import datetime

try:
    from .services.time_utils import UTC_MIN, parse_utc_timestamp, utc_filename_timestamp
except ImportError as import_error:
    from import_compat import ensure_absolute_import_fallback_allowed
    ensure_absolute_import_fallback_allowed(import_error)
    from services.time_utils import UTC_MIN, parse_utc_timestamp, utc_filename_timestamp

try:
    from .terminal_output import emit_doctor_log
except ImportError as import_error:
    from import_compat import ensure_absolute_import_fallback_allowed
    ensure_absolute_import_fallback_allowed(import_error)
    from terminal_output import emit_doctor_log


@dataclass
class HistoryEntry:
    """
    Represents a single error analysis history entry.
    
    Attributes:
        timestamp: ISO format timestamp of when the error occurred
        error: The error message/traceback
        suggestion: Analysis suggestion dictionary with keys like 'pattern', 'message', 'actions'
        node_context: Optional node context dictionary
        workflow_snapshot: Optional workflow JSON snapshot (for F3)
        matched_pattern_id: Optional pattern ID that matched this error (for F4 statistics)
        pattern_category: Optional category of the matched pattern (for F4 statistics)
        pattern_priority: Optional priority of the matched pattern (for F4 statistics)
        resolution_status: Resolution status of the error (for F4 tracking)
        analysis_metadata: Optional analysis metadata (sanitization, sources, etc.)
    """
    timestamp: str
    error: str
    suggestion: Dict[str, Any]
    node_context: Optional[Dict[str, Any]] = None
    workflow_snapshot: Optional[str] = None
    # F4: Pattern metadata for statistics tracking
    matched_pattern_id: Optional[str] = None
    pattern_category: Optional[str] = None
    pattern_priority: Optional[int] = None
    resolution_status: str = "unresolved"  # "resolved"|"unresolved"|"ignored"
    analysis_metadata: Optional[Dict[str, Any]] = None
    # Aggregation fields (optional; backward compatible)
    repeat_count: int = 1
    first_seen: Optional[str] = None
    last_seen: Optional[str] = None
    error_signature: Optional[str] = None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert entry to dictionary for JSON serialization."""
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "HistoryEntry":
        """Create entry from dictionary (backward compatible with old format)."""
        timestamp = data.get("timestamp", "")
        return cls(
            timestamp=timestamp,
            error=data.get("error", ""),
            suggestion=data.get("suggestion", {}),
            node_context=data.get("node_context"),
            workflow_snapshot=data.get("workflow_snapshot"),
            # F4: Pattern metadata (optional for backward compatibility)
            matched_pattern_id=data.get("matched_pattern_id"),
            pattern_category=data.get("pattern_category"),
            pattern_priority=data.get("pattern_priority"),
            resolution_status=data.get("resolution_status", "unresolved"),
            analysis_metadata=data.get("analysis_metadata"),
            repeat_count=int(data.get("repeat_count", 1) or 1),
            first_seen=data.get("first_seen") or timestamp,
            last_seen=data.get("last_seen") or timestamp,
            error_signature=data.get("error_signature"),
        )


class HistoryStore:
    """
    Persistent storage for error analysis history.
    
    Features:
    - JSON file persistence
    - Thread-safe operations
    - Automatic size limiting (maxlen)
    - Cross-restart history retrieval
    
    Usage:
        store = HistoryStore("/path/to/history.json", maxlen=50)
        store.append(HistoryEntry(...))
        history = store.get_all()
    """
    
    def __init__(
        self,
        filepath: str,
        maxlen: int = 50,
        max_bytes: int = 0,
        aggregate_window_seconds: int = 60,
    ):
        """
        Initialize the history store.
        
        Args:
            filepath: Path to the JSON file for persistence
            maxlen: Maximum number of entries to keep (oldest are removed).
                    If maxlen <= 0, history is unbounded by count.
            max_bytes: Maximum size in bytes. If 0, unbounded by size.
        """
        self._filepath = filepath
        self._maxlen = maxlen  # 0 or negative means unbounded
        self._max_bytes = max_bytes
        self._lock = threading.Lock()
        self._history: List[HistoryEntry] = []
        self._loaded = False
        # Aggregation window: within this window, repeated identical errors are aggregated.
        try:
            window = int(aggregate_window_seconds)
        except Exception:
            window = 60
        self._aggregate_window_seconds = window if window > 0 else 60
    
    @property
    def filepath(self) -> str:
        """Get the history file path."""
        return self._filepath
    
    def _load(self) -> None:
        """Load history from JSON file."""
        if self._loaded:
            return
        
        try:
            if os.path.exists(self._filepath):
                with open(self._filepath, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    if isinstance(data, list):
                        self._history = [
                            HistoryEntry.from_dict(entry) 
                            for entry in data 
                            if isinstance(entry, dict)
                        ]
                        # Trim to maxlen (only if bounded)
                        if self._maxlen > 0 and len(self._history) > self._maxlen:
                            self._history = self._history[-self._maxlen:]
        except (json.JSONDecodeError, OSError, TypeError) as e:
            emit_doctor_log(f"Could not load history file: {e}", "WARNING")
            # If the history file is corrupted (common when the process is interrupted
            # while writing), move it aside so new errors can be recorded normally.
            try:
                if os.path.exists(self._filepath):
                    ts = utc_filename_timestamp()
                    backup_path = f"{self._filepath}.corrupt-{ts}"
                    shutil.move(self._filepath, backup_path)
                    emit_doctor_log(f"Corrupted history moved to: {backup_path}", "WARNING")
            except Exception:
                pass
            self._history = []
        
        self._loaded = True
    
    def _save(self) -> None:
        """Save history to JSON file."""
        try:
            # Ensure directory exists
            dir_path = os.path.dirname(self._filepath)
            if dir_path and not os.path.exists(dir_path):
                os.makedirs(dir_path, exist_ok=True)

            # Atomic write to avoid corrupting the JSON file on interruption.
            tmp_path = f"{self._filepath}.tmp"
            
            # Enforce size limit before writing if configured
            data_to_save = [entry.to_dict() for entry in self._history]
            
            if self._max_bytes > 0:
                # Iterative reduction if too large
                # We do this in-memory to avoid writing huge file then checking size
                retries = 10
                while retries > 0:
                    json_str = json.dumps(data_to_save, ensure_ascii=False, indent=None)
                    encoded_len = len(json_str.encode('utf-8'))
                    
                    if encoded_len <= self._max_bytes:
                        break
                        
                    # Needs trimming
                    if not data_to_save:
                        break # Can't reduce further
                        
                    # Calculate how much to trim. 
                    # Smart heuristic: trim 10% or at least 1 item.
                    current_count = len(data_to_save)
                    trim_count = max(1, int(current_count * 0.1))
                    data_to_save = data_to_save[trim_count:]
                    # Update local history to match (so we don't just crop the file but keep memory large)
                    # Note: this affects in-memory state too, which is desired.
                    self._history = self._history[-len(data_to_save):] if data_to_save else []
                    retries -= 1
            
            with open(tmp_path, "w", encoding="utf-8") as f:
                json.dump(
                    data_to_save,
                    f,
                    ensure_ascii=False,
                    indent=None
                )
                f.flush()
                try:
                    os.fsync(f.fileno())
                except Exception:
                    pass
            os.replace(tmp_path, self._filepath)
        except OSError as e:
            emit_doctor_log(f"Could not save history file: {e}", "WARNING")
            try:
                tmp_path = f"{self._filepath}.tmp"
                if os.path.exists(tmp_path):
                    os.remove(tmp_path)
            except Exception:
                pass

    def _parse_ts(self, ts: str) -> datetime:
        """Parse ISO timestamp; return UTC minimum on failure."""
        return parse_utc_timestamp(ts) or UTC_MIN

    def _compute_signature(self, error_text: str) -> str:
        """Compute a deterministic signature for an error text."""
        return hashlib.sha256((error_text or "").encode("utf-8", errors="ignore")).hexdigest()
    
    def append(self, entry: HistoryEntry) -> None:
        """
        Append a new entry to history.
        
        Thread-safe. Automatically persists to disk.
        Old entries are removed if maxlen is exceeded.
        
        Args:
            entry: The HistoryEntry to append
        """
        with self._lock:
            self._load()

            # Normalize aggregation fields
            if not entry.first_seen:
                entry.first_seen = entry.timestamp
            if not entry.last_seen:
                entry.last_seen = entry.timestamp
            if not entry.error_signature:
                entry.error_signature = self._compute_signature(entry.error)

            # Aggregate repeated identical errors within the time window.
            # This prevents unbounded growth when the same error repeats rapidly.
            if self._history:
                now_ts = self._parse_ts(entry.timestamp)
                # Search from newest to oldest for a matching signature within the window.
                for existing in reversed(self._history):
                    if not existing:
                        continue
                    sig = existing.error_signature or self._compute_signature(existing.error)
                    if sig != entry.error_signature:
                        continue
                    last_seen_ts = self._parse_ts(existing.last_seen or existing.timestamp)
                    if (now_ts - last_seen_ts).total_seconds() <= self._aggregate_window_seconds:
                        existing.repeat_count = int(getattr(existing, "repeat_count", 1) or 1) + 1
                        existing.last_seen = entry.timestamp
                        # Best-effort: keep richer metadata if the new entry has it.
                        if not existing.node_context and entry.node_context:
                            existing.node_context = entry.node_context
                        if (not existing.suggestion) and entry.suggestion:
                            existing.suggestion = entry.suggestion
                        if not existing.analysis_metadata and entry.analysis_metadata:
                            existing.analysis_metadata = entry.analysis_metadata
                        self._save()
                        return

            self._history.append(entry)
            
            # Trim to maxlen (only if bounded)
            if self._maxlen > 0 and len(self._history) > self._maxlen:
                self._history = self._history[-self._maxlen:]
            
            self._save()
    
    def get_all(self) -> List[Dict[str, Any]]:
        """
        Get all history entries as dictionaries.
        
        Returns entries in reverse chronological order (newest first).
        
        Returns:
            List of entry dictionaries
        """
        with self._lock:
            self._load()
            # Return in reverse order (newest first)
            return [entry.to_dict() for entry in reversed(self._history)]
    
    def get_latest(self) -> Optional[Dict[str, Any]]:
        """
        Get the most recent history entry.
        
        Returns:
            The latest entry dictionary, or None if history is empty
        """
        with self._lock:
            self._load()
            if self._history:
                return self._history[-1].to_dict()
            return None
    
    def clear(self) -> None:
        """
        Clear all history entries.
        
        Also clears the persisted file.
        """
        with self._lock:
            self._history = []
            self._save()
    
    def __len__(self) -> int:
        """Return the number of entries in history."""
        with self._lock:
            self._load()
            return len(self._history)
    
    def reload(self) -> None:
        """Force reload from disk."""
        with self._lock:
            self._loaded = False
            self._load()