""" Smart Logger for ComfyUI Runtime Diagnostics. Intercepts stdout/stderr to capture logs and provide real-time error analysis. ARCHITECTURE (v1.7.5): - SafeStreamWrapper: Wraps sys.stdout/stderr (after ComfyUI's LogInterceptor if present) - DoctorLogProcessor: Background thread that processes queued messages - Zero deadlock risk: write() holds no locks, only enqueues messages Previous issues (v1.2.x): - Used SmartLogger with AsyncFileWriter - Attempted to use ComfyUI's on_flush callbacks (failed due to callback ordering) - Had potential deadlock risks with nested stream wrapping See: .planning/STAGE1_LOGGER_FIX_PLAN.md for full design details """ import sys import datetime import threading import queue import time import os import logging import hashlib from collections import deque from typing import Optional, Dict, Any, List try: from .services.time_utils import UTC_MIN, parse_utc_timestamp, utc_filename_timestamp, utc_isoformat 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, utc_isoformat try: from .terminal_output import DoctorLogFormatter, emit_doctor_log, stream_supports_color 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 DoctorLogFormatter, emit_doctor_log, stream_supports_color # ============================================================================== # R22: Asyncio transport GC exclusion patterns # ============================================================================== # Windows ProactorEventLoop transport GC raises ValueError during __del__. # These are non-actionable Python-internal warnings that should never trigger # Doctor error analysis. See: roadmap.md R22 _ASYNCIO_GC_EXCLUSION_PATTERNS = ( "_ProactorBasePipeTransport.__del__", "I/O operation on closed pipe", "unclosed transport", ) def _is_asyncio_gc_noise(message: str) -> bool: """Check if a message is a known-benign asyncio transport GC warning. These warnings originate from Python's asyncio internals on Windows when a ProactorEventLoop transport is garbage-collected after its pipe has already been closed. They are non-actionable and would cause recursive log pollution if Doctor tries to analyse them. """ if not message: return False return any(pat in message for pat in _ASYNCIO_GC_EXCLUSION_PATTERNS) # IMPORTANT: keep legacy emoji markers here for backward compatibility # with old logs and pre-R26 console output. New Doctor-owned output must stay # ASCII, but the parser still needs to recognize historical messages. _LEGACY_DOCTOR_OUTPUT_MARKERS = ( "🎯 ERROR LOCATION:", "💡 SUGGESTION:", ) _ASCII_DOCTOR_OUTPUT_MARKERS = ( "ERROR LOCATION:", "SUGGESTION:", ) _LEGACY_TENSOR_ALERT_MARKERS = ( "❌ CRITICAL: Tensor contains", "⚠️ Meta Tensor", ) _ASCII_TENSOR_ALERT_MARKERS = ( "CRITICAL: Tensor contains", "WARNING: Meta Tensor", ) def _contains_tensor_alert(message: str) -> bool: if not message: return False return any(marker in message for marker in (_LEGACY_TENSOR_ALERT_MARKERS + _ASCII_TENSOR_ALERT_MARKERS)) def _normalize_backend_suggestion(text: str) -> str: if not isinstance(text, str): return text normalized = text for prefix in ("💡 SUGGESTION: ", "💡 Suggestion: "): if normalized.startswith(prefix): return "SUGGESTION: " + normalized[len(prefix):] return normalized.replace("💡 SUGGESTION:", "SUGGESTION:") # ============================================================================== # R22: Dedicated internal logger (bypasses SafeStreamWrapper) # ============================================================================== _doctor_internal_logger = logging.getLogger("doctor._internal") _doctor_internal_logger.propagate = False # CRITICAL: do not propagate to root # R22: Thread-local reentrance guard for SafeStreamWrapper _stream_reentrance = threading.local() try: # CRITICAL: do not replace these with bare internal imports; ComfyUI loads # Doctor as a package and the extension root is not guaranteed on sys.path. from .analyzer import ErrorAnalyzer from .config import CONFIG from .history_store import HistoryStore, HistoryEntry from .services.log_ring_buffer import get_ring_buffer from .services.context_extractor import detect_fatal_pattern from .services import doctor_paths from .services.dynamic_vram_advisory import ( clear_dynamic_vram_advisory, record_dynamic_vram_warning, replay_host_dynamic_vram_advisory, ) except ImportError as import_error: # Fallback for direct execution (tests) from import_compat import ensure_absolute_import_fallback_allowed ensure_absolute_import_fallback_allowed(import_error) from analyzer import ErrorAnalyzer from config import CONFIG from history_store import HistoryStore, HistoryEntry try: from services.log_ring_buffer import get_ring_buffer except ImportError: get_ring_buffer = None try: from services.context_extractor import detect_fatal_pattern except ImportError: detect_fatal_pattern = None try: from services import doctor_paths except ImportError: doctor_paths = None from services.dynamic_vram_advisory import ( clear_dynamic_vram_advisory, record_dynamic_vram_warning, replay_host_dynamic_vram_advisory, ) # ============================================================================== # Global State for API Access # ============================================================================== _last_analysis: Dict[str, Any] = { "error": None, "suggestion": None, "timestamp": None, "node_context": None, # NodeContext.to_dict() result "analysis_metadata": None, "matched_pattern_id": None, "pattern_category": None, "pattern_priority": None, "resolution_status": None, } # P1: Error history buffer (ring buffer for last N errors) # When history_size=0, use unbounded deque _analysis_history: deque = deque() if CONFIG.history_size == 0 else deque(maxlen=CONFIG.history_size) # F1: Persistent history store # R18: Use canonical data directory via doctor_paths _current_dir = os.path.dirname(os.path.abspath(__file__)) def _resolve_history_path() -> str: """ Choose a writable, stable location for history persistence. Why: - ComfyUI Desktop / Manager may treat writes inside the extension folder as "conflicts" and may also place extensions under locations with stricter permissions. - Using the canonical Doctor data dir is robust across install modes (portable/git clone/desktop) and respects ComfyUI's user directory when available. """ if doctor_paths: try: data_dir = doctor_paths.get_doctor_data_dir() return os.path.join(data_dir, "error_history.json") except Exception: pass return os.path.join(_current_dir, "logs", "error_history.json") _history_file = _resolve_history_path() _history_store: Optional[HistoryStore] = None def _migrate_legacy_data(): """R18: One-time migration of legacy history to new location.""" if not doctor_paths: return legacy_path = os.path.join(_current_dir, "logs", "error_history.json") target_path = _history_file # If legacy exists and target doesn't, migrate. # Best-effort: prefer moving the file out of the extension folder to avoid "conflict" warnings. if os.path.exists(legacy_path) and not os.path.exists(target_path): try: os.makedirs(os.path.dirname(target_path), exist_ok=True) import shutil # Copy into canonical location first, then rename legacy to a backup marker. # This preserves a forensic trail while preventing repeated migrations. shutil.copy2(legacy_path, target_path) try: ts = utc_filename_timestamp() os.rename(legacy_path, f"{legacy_path}.migrated-{ts}") except Exception: # If rename fails (permissions/locks), leave legacy in place; migration is best-effort. pass except Exception as e: emit_doctor_log(f"Migration failed: {e}", "ERROR", stream=sys.stderr) def _get_history_store() -> HistoryStore: """Lazy initialization of history store.""" global _history_store if _history_store is None: _migrate_legacy_data() aggregate_window_seconds = 60 try: aggregate_window_seconds = int( getattr(CONFIG.guardrails, "AGGREGATION_WINDOW_SECONDS", 60) ) except Exception: aggregate_window_seconds = 60 _history_store = HistoryStore( _history_file, maxlen=CONFIG.history_size, max_bytes=getattr(CONFIG, 'history_size_bytes', 5*1024*1024), aggregate_window_seconds=aggregate_window_seconds, ) return _history_store # R2: Thread safety locks for shared mutable state _history_lock = threading.Lock() _TRACEBACK_ACTIVE = threading.Event() class DroppingQueue: """ Bounded queue with drop policy and counters. - Non-priority messages are dropped when full. - Priority messages evict the oldest non-priority item if possible. - Drop counters are tracked for health/observability. """ def __init__(self, maxsize: int = 1000): self._maxsize = maxsize self._queue = deque() self._cv = threading.Condition() self._stats = { "queue_dropped_total": 0, "queue_dropped_priority": 0, "queue_dropped_non_priority": 0, "queue_dropped_oldest": 0, } def _record_drop(self, item): if not item: return priority, _ = item self._stats["queue_dropped_total"] += 1 if priority: self._stats["queue_dropped_priority"] += 1 else: self._stats["queue_dropped_non_priority"] += 1 def _remove_at(self, index: int): self._queue.rotate(-index) item = self._queue.popleft() self._queue.rotate(index) return item def _drop_oldest(self, prefer_non_priority: bool): if not self._queue: return None if prefer_non_priority: for idx, item in enumerate(self._queue): if not item[0]: return self._remove_at(idx) return self._queue.popleft() def put_nowait(self, item, priority: bool = False) -> bool: with self._cv: if self._maxsize and len(self._queue) >= self._maxsize: if priority: dropped = self._drop_oldest(prefer_non_priority=True) self._record_drop(dropped) self._stats["queue_dropped_oldest"] += 1 else: self._stats["queue_dropped_total"] += 1 self._stats["queue_dropped_non_priority"] += 1 return False self._queue.append((priority, item)) self._cv.notify() return True def get(self, timeout: float = None): with self._cv: if not self._queue: self._cv.wait(timeout) if not self._queue: raise queue.Empty return self._queue.popleft() def qsize(self) -> int: with self._cv: return len(self._queue) def get_stats(self) -> Dict[str, int]: with self._cv: return dict(self._stats) def clear(self) -> None: with self._cv: self._queue.clear() def _is_priority_message(message: str) -> bool: if _TRACEBACK_ACTIVE.is_set(): return True if not message: return False if "Traceback (most recent call last):" in message: return True if "Failed to validate prompt for output" in message: return True return False # ============================================================================== # API Functions (preserved from original implementation) # ============================================================================== def get_last_analysis() -> Dict[str, Any]: """Get the last error analysis result for API access.""" with _history_lock: return _last_analysis.copy() def update_resolution_status(timestamp: str, status: str) -> bool: """Update resolution status in history store and in-memory analysis data.""" updated = False try: store = _get_history_store() with store._lock: store._load() for entry in store._history: if entry.timestamp == timestamp: entry.resolution_status = status updated = True break store._save() except Exception: pass with _history_lock: global _last_analysis if _last_analysis.get("timestamp") == timestamp: _last_analysis["resolution_status"] = status updated = True for entry in _analysis_history: if entry.get("timestamp") == timestamp: entry["resolution_status"] = status updated = True break return updated def get_analysis_history() -> List[Dict[str, Any]]: """Get the error analysis history (most recent first).""" # Prefer persistent store, fallback to in-memory try: store = _get_history_store() history = store.get_all() if history: return history except Exception: pass # Fallback to in-memory with _history_lock: return list(reversed(_analysis_history)) def clear_analysis_history() -> bool: """Clear all history (both persistent and in-memory).""" success = True try: store = _get_history_store() store.clear() except Exception: success = False with _history_lock: _analysis_history.clear() global _last_analysis _last_analysis = { "error": None, "suggestion": None, "timestamp": None, "node_context": None, "analysis_metadata": None, "matched_pattern_id": None, "pattern_category": None, "pattern_priority": None, "resolution_status": None, } # Best-effort reset of in-flight logger state so old queued lines don't repopulate # _last_analysis right after clearing (important for deterministic tests). try: if _message_queue: _message_queue.clear() except Exception: success = False try: if _log_processor: # IMPORTANT: keep traceback state reset inside DoctorLogProcessor; direct field writes race capture. _log_processor.reset_traceback_state() except Exception: success = False return success # ============================================================================== # New Architecture: SafeStreamWrapper + DoctorLogProcessor # ============================================================================== class SafeStreamWrapper: """ Safe stream wrapper that avoids deadlock with ComfyUI's LogInterceptor. Design principles: 1. write() immediately pass-through (holds NO locks) 2. Message is enqueued for background processing 3. All other attributes are proxied to original stream This avoids conflicts with ComfyUI's LogInterceptor because: - We don't hold any locks during write() - We call original_stream.write() FIRST, then enqueue - Background thread is completely decoupled """ def __init__(self, original_stream, message_queue): """ Initialize wrapper. Args: original_stream: Original stream (may be LogInterceptor or raw stdout/stderr) message_queue: Queue for background processing """ self._original_stream = original_stream self._queue = message_queue def write(self, data): """ Write data: immediate pass-through + enqueue for analysis. CRITICAL: This method holds NO locks to avoid deadlock. R22: Thread-local reentrance guard prevents recursive capture. """ # R22: Reentrance guard — if we are already inside a write() on this # thread (e.g. because _record_analysis triggered a logging call that # routes back here), skip the enqueue step to break the recursion. if getattr(_stream_reentrance, "active", False): try: self._original_stream.write(data) except (OSError, AttributeError, UnicodeError, ValueError): pass return _stream_reentrance.active = True try: # 1. Immediately write to original stream (may be LogInterceptor) try: self._original_stream.write(data) except (OSError, AttributeError, UnicodeError, ValueError): pass # Stream may be closed during shutdown # 2. Enqueue for background processing (non-blocking) try: priority = _is_priority_message(data) self._queue.put_nowait(data, priority=priority) except Exception: pass finally: _stream_reentrance.active = False # R14: Add to ring buffer for reliable log context capture try: if get_ring_buffer: ring_buffer = get_ring_buffer() ring_buffer.add_line(data) except Exception: pass # Never fail on ring buffer operations def flush(self): """Flush original stream.""" try: self._original_stream.flush() except (OSError, AttributeError, UnicodeError, ValueError): pass def __getattr__(self, name): """Proxy all other attributes to original stream (encoding, isatty, fileno, etc).""" return getattr(self._original_stream, name) class FlushSafeProxy: """ Stream proxy that never raises from write()/flush(). Purpose: - Some ComfyUI Desktop builds use a LogInterceptor whose flush() can raise `OSError: [Errno 22] Invalid argument` under Windows packaging. - Python's logging.StreamHandler calls flush() after every emit. - If flush raises, logging prints an exception traceback, causing a log storm. This proxy keeps the original stream behavior but makes flush safe. It intentionally does NOT enqueue messages for Doctor analysis. """ def __init__(self, original_stream): self._original_stream = original_stream def write(self, data): try: return self._original_stream.write(data) except (OSError, AttributeError, UnicodeError, ValueError): return 0 def flush(self): try: return self._original_stream.flush() except (OSError, AttributeError, UnicodeError, ValueError): return None def __getattr__(self, name): return getattr(self._original_stream, name) # IMPORTANT: keep the internal logger on a flush-safe stream. On some # Windows/Anaconda console states, raw sys.__stderr__ writes can raise # WinError 6 / invalid handle from background threads; if that escapes a # StreamHandler emit, Python logging prints a secondary traceback that Doctor # then captures as a false error. if not _doctor_internal_logger.handlers: _internal_handler = logging.StreamHandler(FlushSafeProxy(sys.__stderr__)) _internal_handler.setFormatter( DoctorLogFormatter( include_timestamp=False, color_enabled=stream_supports_color(sys.__stderr__), ) ) _doctor_internal_logger.addHandler(_internal_handler) _doctor_internal_logger.setLevel(logging.DEBUG) class DoctorLogProcessor(threading.Thread): """ Background thread for log analysis. Functionality: 1. Read messages from queue 2. Assemble traceback buffer (same logic as old SmartLogger._analyze_stream) 3. Call ErrorAnalyzer.analyze() 4. Update history store This runs completely independently from the main thread. """ def __init__(self, message_queue): super().__init__(daemon=True, name="DoctorLogProcessor") self._queue = message_queue self._running = True self._metrics = { "buffer_dropped": 0, "traceback_resets": 0, "queue_timeouts": 0, } # Traceback buffer state (migrated from SmartLogger) self._buffer_lock = threading.RLock() self.buffer = [] self.in_traceback = False self.last_buffer_time = 0 try: self._aggregate_window_seconds = int( getattr(CONFIG.guardrails, "AGGREGATION_WINDOW_SECONDS", 60) ) except Exception: self._aggregate_window_seconds = 60 if self._aggregate_window_seconds <= 0: self._aggregate_window_seconds = 60 def _parse_ts(self, ts: str) -> datetime.datetime: return parse_utc_timestamp(ts) or UTC_MIN def run(self): """Main loop: process queued messages.""" while self._running: try: # Use timeout to allow periodic buffer timeout checks priority, message = self._queue.get(timeout=0.5) self._process_message(message) except queue.Empty: # Timeout: check if buffer needs flushing self._metrics["queue_timeouts"] += 1 self._check_buffer_timeout() except Exception as e: # Log error but don't crash the thread _doctor_internal_logger.error("LogProcessor error: %s", e, exc_info=True) def _process_message(self, message): with self._buffer_lock: return self._process_message_locked(message) def _process_message_locked(self, message): """ Process a single message (migrated from SmartLogger._analyze_stream). """ current_time = time.time() # Always check buffer timeout, even when logs keep coming. # In some environments (e.g., ComfyUI Desktop), periodic Doctor-API logs # can prevent queue.Empty from firing, which previously blocked buffer flush # for non-traceback errors like "Failed to validate prompt for output". self._check_buffer_timeout() # ═══════════════════════════════════════════════════════════════ # CRITICAL FIX (2026-01-06): Prevent recursive log capture # ═══════════════════════════════════════════════════════════════ # BUG: Doctor's own log messages (e.g., "[Doctor] _record_analysis called") # were being captured by SafeStreamWrapper, and since they contain # the original error text, they triggered new analysis cycles. # This caused duplicate suggestions and infinite recursion. # # SOLUTION: Skip any message that contains Doctor's internal log markers. # # DO NOT REMOVE THIS CHECK! It prevents infinite recursion. # See: .planning/260106-BUGFIX_DUPLICATE_LOG_CAPTURE.md # ═══════════════════════════════════════════════════════════════ doctor_markers = [ "[ComfyUI-Doctor ", # Canonical severity-aware output prefix "[Doctor]", # Internal logging prefix "[Doctor-API]", # API logging prefix "----------------------------------------", # Divider line ] doctor_markers.extend(_ASCII_DOCTOR_OUTPUT_MARKERS) doctor_markers.extend(_LEGACY_DOCTOR_OUTPUT_MARKERS) for marker in doctor_markers: if marker in message: return # Skip Doctor's own output to prevent recursion # R22: Exclude asyncio transport GC warnings (Windows ProactorEventLoop) # These are non-actionable Python-internal warnings that cause recursive # log pollution when Doctor captures and re-analyses them. if _is_asyncio_gc_noise(message): return # CRITICAL: DynamicVRAM startup fallback is a nonfatal health advisory. # Never send it through ErrorAnalyzer or runtime error/history state. if record_dynamic_vram_warning(message): return # P3: Urgent single-line warnings (immediate analysis) # R14: Enhanced with detect_fatal_pattern for non-traceback errors is_urgent = _contains_tensor_alert(message) # R14: Check for fatal patterns (CUDA OOM, CRITICAL, etc.) if not is_urgent and detect_fatal_pattern: fatal_marker = detect_fatal_pattern(message) if fatal_marker: is_urgent = True _doctor_internal_logger.debug("R14 fatal pattern detected: %s", fatal_marker) if is_urgent and not self.in_traceback: result = ErrorAnalyzer.analyze(message) suggestion, metadata = result if result else (None, None) if suggestion: self._record_analysis(message, suggestion, metadata) return # Detect traceback start if "Traceback (most recent call last):" in message: self._set_traceback_state(True) self.buffer = [message] self.last_buffer_time = current_time return # Handle validation errors if "Failed to validate prompt for output" in message: if not self.in_traceback: self._set_traceback_state(True) self.buffer = [message] self.last_buffer_time = current_time else: self.buffer.append(message) self.last_buffer_time = current_time return if self.in_traceback: # Check timeout if current_time - self.last_buffer_time > CONFIG.traceback_timeout_seconds: full_traceback = "".join(self.buffer) result = ErrorAnalyzer.analyze(full_traceback) suggestion, metadata = result if result else (None, None) if suggestion or "Failed to validate" in full_traceback: self._record_analysis(full_traceback, suggestion, metadata) self._set_traceback_state(False) self.buffer = [] else: # ═══════════════════════════════════════════════════════════════ # CRITICAL FIX (2026-01-06): Check completion marker BEFORE buffer.append() # ═══════════════════════════════════════════════════════════════ # BUG: "Prompt executed in X seconds" was being added to buffer BEFORE # is_complete_traceback() detected completion, causing it to be # recorded as part of the error message. # # SOLUTION: Intercept "Prompt executed" BEFORE appending to buffer. # Use current buffer (without this line) for recording. # # DO NOT MOVE THIS CHECK AFTER buffer.append()! # ═══════════════════════════════════════════════════════════════ if "Prompt executed" in message: full_traceback = "".join(self.buffer) if full_traceback.strip(): # Only record if buffer has content result = ErrorAnalyzer.analyze(full_traceback) suggestion, metadata = result if result else (None, None) self._record_analysis(full_traceback, suggestion, metadata) self._set_traceback_state(False) self.buffer = [] return self.buffer.append(message) self.last_buffer_time = current_time full_traceback = "".join(self.buffer) # Normal traceback completion if ErrorAnalyzer.is_complete_traceback(full_traceback): result = ErrorAnalyzer.analyze(full_traceback) suggestion, metadata = result if result else (None, None) self._record_analysis(full_traceback, suggestion, metadata) self._set_traceback_state(False) self.buffer = [] # Buffer limit safety elif len(self.buffer) > CONFIG.buffer_limit: self._metrics["buffer_dropped"] += 1 self._metrics["traceback_resets"] += 1 self._set_traceback_state(False) self.buffer = [] def _check_buffer_timeout(self): """Check if buffer has timed out (called on queue.Empty).""" with self._buffer_lock: if self.in_traceback and self.buffer: current_time = time.time() if current_time - self.last_buffer_time > CONFIG.traceback_timeout_seconds: full_traceback = "".join(self.buffer) result = ErrorAnalyzer.analyze(full_traceback) suggestion, metadata = result if result else (None, None) if suggestion or "Failed to validate" in full_traceback: self._record_analysis(full_traceback, suggestion, metadata) self._metrics["traceback_resets"] += 1 self._set_traceback_state(False) self.buffer = [] def reset_traceback_state(self) -> None: """Clear in-flight traceback assembly under the processor-owned lock.""" with self._buffer_lock: self.buffer = [] self._set_traceback_state(False) self.last_buffer_time = 0 def _set_traceback_state(self, active: bool) -> None: self.in_traceback = active if active: _TRACEBACK_ACTIVE.set() else: _TRACEBACK_ACTIVE.clear() def _record_analysis(self, full_traceback, suggestion, metadata=None): """ Record analysis result (migrated from SmartLogger._record_analysis). Args: full_traceback: The full traceback string suggestion: Suggestion text (or None if no match) metadata: Optional metadata dict with pattern info (from F4) """ # R22: Use dedicated internal logger that bypasses SafeStreamWrapper # to prevent cross-contamination between Doctor's diagnostic output # and the captured exception text. _doctor_internal_logger.debug( "_record_analysis called with traceback preview: %s...", full_traceback[:100] if full_traceback else "None", ) # ═══════════════════════════════════════════════════════════════ # CRITICAL FIX (2026-01-06): Prevent non-error messages from # overwriting legitimate error records # ═══════════════════════════════════════════════════════════════ # BUG: Normal messages like "Prompt executed in X seconds" were # being recorded as errors, overwriting real error logs. # # SOLUTION: Validate that full_traceback contains actual error # indicators before recording. # # DO NOT REMOVE THIS VALIDATION! # ═══════════════════════════════════════════════════════════════ # Skip recording if no actual error content # This prevents normal messages like "Prompt executed in X seconds" from # overwriting legitimate error records if not full_traceback: return # Desktop-specific: ignore ComfyUI logging flush exceptions. # These can create high-frequency "error storms" even when inference isn't running, # rapidly filling error_history.json with non-actionable noise. if ( "logging\\__init__.py" in full_traceback and "ComfyUI\\app\\logger.py" in full_traceback and "OSError: [Errno 22] Invalid argument" in full_traceback ): return # R22: Exclude asyncio transport GC warnings # These originate from Windows ProactorEventLoop transport __del__ # and are non-actionable noise that would fill history with duplicates. if _is_asyncio_gc_noise(full_traceback): return # Explicit exclusion for normal execution messages # These messages should NEVER be recorded as errors normal_messages = [ "Prompt executed in", "got prompt", "Executing node", "To see the GUI go to:", "Starting server", ] for normal_msg in normal_messages: if normal_msg in full_traceback and "Error" not in full_traceback and "Exception" not in full_traceback: _doctor_internal_logger.debug("Skipping normal message: %s", full_traceback[:100]) return # Require at least one error indicator error_indicators = [ "Traceback (most recent call last):", "Error:", "Error ", # Catch RuntimeError, ValueError etc without colon "Exception:", "Exception ", "Failed to validate", "❌ CRITICAL", "⚠️ Meta Tensor", "CRITICAL: Tensor contains", "WARNING: Meta Tensor", ] has_error_indicator = any(indicator in full_traceback for indicator in error_indicators) # Also accept if we have a valid suggestion (pattern matched) if not has_error_indicator and not suggestion: _doctor_internal_logger.debug( "Skipping non-error message (no indicators): %s", full_traceback[:100], ) return analysis_metadata = metadata if isinstance(metadata, dict) else {} matched_pattern_id = analysis_metadata.get("matched_pattern_id") pattern_category = analysis_metadata.get("pattern_category") or analysis_metadata.get("category") pattern_priority = analysis_metadata.get("pattern_priority") or analysis_metadata.get("priority") node_context = ErrorAnalyzer.extract_node_context(full_traceback) timestamp = utc_isoformat() error_signature = hashlib.sha256(full_traceback.encode("utf-8", errors="ignore")).hexdigest() new_analysis = { "error": full_traceback, "suggestion": suggestion, "timestamp": timestamp, "first_seen": timestamp, "last_seen": timestamp, "repeat_count": 1, "error_signature": error_signature, "node_context": node_context.to_dict() if node_context else None, "analysis_metadata": analysis_metadata, "matched_pattern_id": matched_pattern_id, "pattern_category": pattern_category, "pattern_priority": pattern_priority, "resolution_status": "unresolved", } # R2: Thread-safe update of shared state with _history_lock: global _last_analysis _last_analysis = new_analysis # Aggregate repeated identical errors within 60 seconds to avoid unbounded history growth. # NOTE: We still update _last_analysis every time so the UI reflects new occurrences. aggregated = False try: now_ts = self._parse_ts(timestamp) # Look back across recent history to find a matching signature in-window. # Limit scan to keep cost bounded even when history is unbounded. scan_limit = 50 hist_len = len(_analysis_history) start_index = max(0, hist_len - scan_limit) for i in range(hist_len - 1, start_index - 1, -1): existing = _analysis_history[i] if not isinstance(existing, dict): continue if existing.get("error_signature") != error_signature: continue last_seen_str = existing.get("last_seen") or existing.get("timestamp") or "" last_seen_ts = self._parse_ts(last_seen_str) if (now_ts - last_seen_ts).total_seconds() <= self._aggregate_window_seconds: existing["repeat_count"] = int(existing.get("repeat_count", 1) or 1) + 1 existing["last_seen"] = timestamp # Keep first_seen stable if not existing.get("first_seen"): existing["first_seen"] = existing.get("timestamp") or timestamp aggregated = True break except Exception: pass if not aggregated: _analysis_history.append(new_analysis.copy()) # F1: Persist to history store try: store = _get_history_store() entry = HistoryEntry( timestamp=timestamp, error=full_traceback, suggestion=suggestion if suggestion else {}, node_context=node_context.to_dict() if node_context else None, matched_pattern_id=matched_pattern_id, pattern_category=pattern_category, pattern_priority=pattern_priority, resolution_status="unresolved", analysis_metadata=analysis_metadata, repeat_count=1, first_seen=timestamp, last_seen=timestamp, error_signature=error_signature, ) store.append(entry) except Exception: pass # Persistence failure should not break error analysis # Only print if we have something useful (context OR suggestion) if not suggestion and (not node_context or not node_context.is_valid()): return # Build formatted output output_parts = [] output_parts.append(f"\n{'-'*40}") # Add node context if available if node_context and node_context.is_valid(): node_info = [] if node_context.node_id: node_info.append(f"Node ID: #{node_context.node_id}") if node_context.node_name: node_info.append(f"Name: {node_context.node_name}") if node_context.node_class: node_info.append(f"Class: {node_context.node_class}") if node_context.custom_node_path: node_info.append(f"Source: {node_context.custom_node_path}") output_parts.append("ERROR LOCATION: " + " | ".join(node_info)) # Add suggestion (defensive: ensure suggestion is a string) if suggestion: if isinstance(suggestion, str): output_parts.append(_normalize_backend_suggestion(suggestion)) elif isinstance(suggestion, (tuple, list)) and len(suggestion) > 0: # Handle case where suggestion is accidentally a tuple (e.g., from analyzer) normalized = str(suggestion[0]) if suggestion[0] else "" output_parts.append(_normalize_backend_suggestion(normalized)) output_parts.append(f"{'-'*40}\n") formatted_output = "\n".join(output_parts) # Print to stdout (will be captured by our wrapper and ComfyUI's LogInterceptor) try: print(formatted_output, file=sys.__stdout__) except (OSError, AttributeError, UnicodeError, ValueError): pass # Silently fail if stdout is unavailable def stop(self): """Gracefully stop the processor thread.""" self._running = False # ============================================================================== # Installation API (preserved interface, new implementation) # ============================================================================== # Global state for installation _message_queue = None _log_processor = None _original_stdout = None _original_stderr = None def install(log_path: str): """ Install Doctor logger (new architecture). Args: log_path: Log file path (currently unused, kept for API compatibility) Architecture: - Creates message queue and background processor thread - Wraps sys.stdout/stderr with SafeStreamWrapper - Background thread handles all error analysis Note: This completely avoids the on_flush callback issues from v1.2.x """ global _message_queue, _log_processor, _original_stdout, _original_stderr # IMPORTANT: do a clean reinstall instead of skipping outright. Tests, # hot-reload, or interrupted teardown can leave the global logger state # installed across calls; a hard skip would make subsequent captures flaky. if _message_queue is not None or isinstance(sys.stdout, SafeStreamWrapper) or isinstance(sys.stderr, SafeStreamWrapper): try: _doctor_internal_logger.debug("install() detected existing logger state; forcing clean reinstall") except Exception: pass uninstall() clear_dynamic_vram_advisory() # Create the existing queue/processor, replay the bounded startup snapshot # directly into advisory state, then start the sole worker. _message_queue = DroppingQueue(maxsize=CONFIG.log_queue_maxsize) _log_processor = DoctorLogProcessor(_message_queue) replay_host_dynamic_vram_advisory() _log_processor.start() # Save original streams (may already be ComfyUI's LogInterceptor) _original_stdout = sys.stdout _original_stderr = sys.stderr # Wrap stdout/stderr sys.stdout = SafeStreamWrapper(_original_stdout, _message_queue) sys.stderr = SafeStreamWrapper(_original_stderr, _message_queue) # ComfyUI Desktop (and some ComfyUI builds) install StreamHandler(s) before Doctor, # capturing the original sys.stdout/sys.stderr object (often LogInterceptor). # If that stream's flush() raises (e.g., OSError 22), Python logging will emit a traceback # for EVERY log record, causing massive log spam and filling Doctor history with noise. # # Patch existing StreamHandler streams to wrap them in a flush-safe proxy. try: root_logger = logging.getLogger() for handler in list(getattr(root_logger, "handlers", []) or []): stream = getattr(handler, "stream", None) if stream in (_original_stdout, _original_stderr) or getattr(type(stream), "__name__", "") == "LogInterceptor": handler.stream = FlushSafeProxy(stream) except Exception: pass emit_doctor_log("Logger installed (SafeStreamWrapper mode)", "INFO", stream=sys.stdout) emit_doctor_log(f"Original stdout type: {type(_original_stdout).__name__}", "INFO", stream=sys.stdout) emit_doctor_log(f"Original stderr type: {type(_original_stderr).__name__}", "INFO", stream=sys.stdout) def uninstall(): """Uninstall logger and restore original streams.""" global _log_processor, _original_stdout, _original_stderr, _message_queue # Restore streams first so shutdown logging cannot recurse through wrappers. if _original_stdout is not None: sys.stdout = _original_stdout if _original_stderr is not None: sys.stderr = _original_stderr # Stop background thread if _log_processor: _log_processor.stop() _log_processor.join(timeout=2.0) _log_processor = None # Clear queue if _message_queue: _message_queue.clear() _message_queue = None clear_dynamic_vram_advisory() _original_stdout = None _original_stderr = None emit_doctor_log("Logger uninstalled", "INFO", stream=sys.stdout) def get_logger_metrics() -> Dict[str, Any]: """Return logger health metrics for diagnostics and tests.""" metrics = { "queue_size": 0, "queue_dropped_total": 0, "queue_dropped_priority": 0, "queue_dropped_non_priority": 0, "queue_dropped_oldest": 0, "buffer_dropped": 0, "traceback_resets": 0, "queue_timeouts": 0, } if _message_queue: metrics.update(_message_queue.get_stats()) metrics["queue_size"] = _message_queue.qsize() if _log_processor: metrics.update(_log_processor._metrics) return metrics # ============================================================================== # Backward Compatibility # ============================================================================== class SmartLogger: """ Backward compatibility class. The old SmartLogger interface is preserved for existing code that may reference it, but internally it now uses the new architecture. """ @classmethod def install(cls, log_path: str): """Install logger (delegates to new install()).""" install(log_path) @classmethod def uninstall(cls): """Uninstall logger (delegates to new uninstall()).""" uninstall()