270 lines
9.4 KiB
Python
270 lines
9.4 KiB
Python
"""
|
|
R14 Service: Context Extractor.
|
|
Extracts structured error context from tracebacks and logs.
|
|
|
|
Features:
|
|
- extract_error_summary(): Extract exception type + message from traceback
|
|
- collapse_stack_frames(): Keep first N + last M frames for token efficiency
|
|
- detect_fatal_pattern(): Identify non-traceback fatal errors
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
from typing import Optional, Tuple, List, Dict, Any
|
|
from dataclasses import dataclass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# CONSTANTS
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
# Pattern to extract the final exception line from a traceback
|
|
EXCEPTION_LINE_PATTERN = re.compile(
|
|
r'^([A-Za-z_][A-Za-z0-9_.]*(?:Error|Exception|Warning|Interrupt)?)\s*:\s*(.+)$',
|
|
re.MULTILINE
|
|
)
|
|
|
|
# Pattern to identify stack frame lines
|
|
FRAME_PATTERN = re.compile(r'^\s+File\s+"[^"]+",\s+line\s+\d+', re.MULTILINE)
|
|
|
|
# Non-traceback fatal error markers (Phase 3)
|
|
FATAL_MARKERS = [
|
|
re.compile(r'^ERROR:\s*Exception\s+in\s+', re.IGNORECASE),
|
|
re.compile(r'^CRITICAL:\s*', re.IGNORECASE),
|
|
re.compile(r'^RuntimeError:\s*', re.IGNORECASE),
|
|
re.compile(r'^\[ComfyUI\]\s+ERROR:\s*', re.IGNORECASE),
|
|
re.compile(r'^OOM\s*:', re.IGNORECASE), # Out of memory
|
|
re.compile(r'^CUDA\s+out\s+of\s+memory', re.IGNORECASE),
|
|
]
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# DATA CLASSES
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
@dataclass
|
|
class ErrorSummary:
|
|
"""Structured error summary."""
|
|
exception_type: str
|
|
message: str
|
|
category: Optional[str] = None # From PatternMatcher if available
|
|
|
|
def to_string(self) -> str:
|
|
"""Format as single-line summary."""
|
|
if self.category:
|
|
return f"[{self.category}] {self.exception_type}: {self.message}"
|
|
return f"{self.exception_type}: {self.message}"
|
|
|
|
@dataclass
|
|
class ContextManifest:
|
|
"""Manifest of what context was included (for observability)."""
|
|
traceback_chars: int = 0
|
|
traceback_frames: int = 0
|
|
logs_lines: int = 0
|
|
workflow_nodes: int = 0
|
|
env_packages_included: int = 0
|
|
summary_present: bool = False
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"traceback_chars": self.traceback_chars,
|
|
"traceback_frames": self.traceback_frames,
|
|
"logs_lines": self.logs_lines,
|
|
"workflow_nodes": self.workflow_nodes,
|
|
"env_packages_included": self.env_packages_included,
|
|
"summary_present": self.summary_present,
|
|
}
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# CORE FUNCTIONS
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
def extract_error_summary(
|
|
traceback_text: str,
|
|
pattern_category: Optional[str] = None
|
|
) -> Optional[ErrorSummary]:
|
|
"""
|
|
Extract a short error summary from traceback text.
|
|
|
|
Args:
|
|
traceback_text: Full traceback string
|
|
pattern_category: Optional category from PatternMatcher
|
|
|
|
Returns:
|
|
ErrorSummary or None if extraction fails
|
|
"""
|
|
if not traceback_text:
|
|
return None
|
|
|
|
# Find the last exception line in the traceback
|
|
lines = traceback_text.strip().split('\n')
|
|
|
|
# Search from the end for the exception line
|
|
for line in reversed(lines):
|
|
line = line.strip()
|
|
match = EXCEPTION_LINE_PATTERN.match(line)
|
|
if match:
|
|
exc_type = match.group(1)
|
|
exc_message = match.group(2).strip()
|
|
# Truncate message if too long
|
|
if len(exc_message) > 200:
|
|
exc_message = exc_message[:197] + "..."
|
|
return ErrorSummary(
|
|
exception_type=exc_type,
|
|
message=exc_message,
|
|
category=pattern_category
|
|
)
|
|
|
|
# Fallback: Try to find any line that looks like an exception
|
|
for line in reversed(lines):
|
|
line = line.strip()
|
|
if line and not line.startswith(('File ', 'Traceback ', ' ', 'During handling')):
|
|
# This might be the exception line
|
|
if ':' in line:
|
|
parts = line.split(':', 1)
|
|
return ErrorSummary(
|
|
exception_type=parts[0].strip(),
|
|
message=parts[1].strip()[:200] if len(parts) > 1 else "",
|
|
category=pattern_category
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def collapse_stack_frames(
|
|
traceback_text: str,
|
|
head_frames: int = 3,
|
|
tail_frames: int = 5,
|
|
preserve_exception: bool = True
|
|
) -> str:
|
|
"""
|
|
Collapse stack frames to reduce token usage while preserving context.
|
|
|
|
Keeps:
|
|
- First N frames (head_frames) - shows entry point
|
|
- Last M frames (tail_frames) - shows where error occurred
|
|
- Exception line (always preserved)
|
|
|
|
Args:
|
|
traceback_text: Full traceback string
|
|
head_frames: Number of frames to keep from the start
|
|
tail_frames: Number of frames to keep from the end
|
|
preserve_exception: Always keep the final exception line
|
|
|
|
Returns:
|
|
Collapsed traceback string
|
|
"""
|
|
if not traceback_text:
|
|
return ""
|
|
|
|
lines = traceback_text.split('\n')
|
|
result = []
|
|
frame_indices = []
|
|
|
|
# Find all frame start indices (lines starting with " File ")
|
|
for i, line in enumerate(lines):
|
|
if FRAME_PATTERN.match(line):
|
|
frame_indices.append(i)
|
|
|
|
if len(frame_indices) <= head_frames + tail_frames:
|
|
# No need to collapse
|
|
return traceback_text
|
|
|
|
# Determine which frames to keep
|
|
keep_frames = set(frame_indices[:head_frames] + frame_indices[-tail_frames:])
|
|
omitted_count = len(frame_indices) - len(keep_frames)
|
|
|
|
# Build result
|
|
i = 0
|
|
omit_section_added = False
|
|
|
|
while i < len(lines):
|
|
line = lines[i]
|
|
|
|
if FRAME_PATTERN.match(line):
|
|
frame_idx = frame_indices.index(i) if i in frame_indices else -1
|
|
|
|
if i in keep_frames:
|
|
result.append(line)
|
|
# Include the next line (code snippet) if it exists
|
|
if i + 1 < len(lines) and lines[i + 1].strip().startswith('^') is False:
|
|
if not FRAME_PATTERN.match(lines[i + 1]) and not lines[i + 1].strip().startswith('Traceback'):
|
|
i += 1
|
|
result.append(lines[i])
|
|
omit_section_added = False
|
|
else:
|
|
if not omit_section_added:
|
|
result.append(f" ... ({omitted_count} frames omitted) ...")
|
|
omit_section_added = True
|
|
else:
|
|
result.append(line)
|
|
omit_section_added = False
|
|
|
|
i += 1
|
|
|
|
return '\n'.join(result)
|
|
|
|
|
|
def detect_fatal_pattern(line: str) -> Optional[str]:
|
|
"""
|
|
Detect if a log line matches a non-traceback fatal error pattern.
|
|
|
|
Args:
|
|
line: Single log line
|
|
|
|
Returns:
|
|
Pattern name if matched, None otherwise
|
|
"""
|
|
if not line:
|
|
return None
|
|
|
|
line = line.strip()
|
|
|
|
for i, pattern in enumerate(FATAL_MARKERS):
|
|
if pattern.match(line):
|
|
return f"fatal_marker_{i}"
|
|
|
|
return None
|
|
|
|
|
|
def build_context_manifest(
|
|
traceback_text: Optional[str] = None,
|
|
execution_logs: Optional[List[str]] = None,
|
|
workflow_json: Optional[Dict[str, Any]] = None,
|
|
system_info: Optional[Dict[str, Any]] = None,
|
|
error_summary: Optional[ErrorSummary] = None
|
|
) -> ContextManifest:
|
|
"""
|
|
Build a manifest of included context for observability.
|
|
|
|
Args:
|
|
traceback_text: The traceback string
|
|
execution_logs: List of log lines
|
|
workflow_json: Workflow dict
|
|
system_info: System info dict (may contain packages)
|
|
error_summary: Extracted error summary
|
|
|
|
Returns:
|
|
ContextManifest with counts
|
|
"""
|
|
manifest = ContextManifest()
|
|
|
|
if traceback_text:
|
|
manifest.traceback_chars = len(traceback_text)
|
|
manifest.traceback_frames = len(FRAME_PATTERN.findall(traceback_text))
|
|
|
|
if execution_logs:
|
|
manifest.logs_lines = len(execution_logs)
|
|
|
|
if workflow_json:
|
|
manifest.workflow_nodes = len(workflow_json)
|
|
|
|
if system_info and "packages" in system_info:
|
|
packages = system_info.get("packages", [])
|
|
if isinstance(packages, list):
|
|
manifest.env_packages_included = len(packages)
|
|
|
|
manifest.summary_present = error_summary is not None
|
|
|
|
return manifest
|