""" Simple debug logger for AI prompts and responses. Writes files chronologically to the configured log directory with sequential numbering. """ import os from datetime import datetime, UTC from typing import List, Optional, Any from modules.shared.configuration import APP_CONFIG def _resolveLogDir() -> str: """Resolve the absolute log directory from configuration.""" logDir = APP_CONFIG.get("APP_LOGGING_LOG_DIR", "./") if not os.path.isabs(logDir): # If relative path, make it relative to the gateway directory gatewayDir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) logDir = os.path.join(gatewayDir, logDir) return logDir def _ensureDir(path: str) -> None: """Create directory if it does not exist.""" os.makedirs(path, exist_ok=True) def _isDebugEnabled() -> bool: """Check if debug workflow logging is enabled.""" return APP_CONFIG.get("APP_DEBUG_CHAT_WORKFLOW_ENABLED", False) def _getBaseDebugDir() -> str: """Get the base debug directory path from configuration.""" # Check if custom debug directory is configured customDebugDir = APP_CONFIG.get("APP_DEBUG_CHAT_WORKFLOW_DIR", None) if customDebugDir: # Use custom debug directory if configured if not os.path.isabs(customDebugDir): # If relative path, make it relative to the gateway directory gatewayDir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) customDebugDir = os.path.join(gatewayDir, customDebugDir) return customDebugDir # Default: Get log directory from config (same as used by main logging system) logDir = _resolveLogDir() # Create debug subdirectory within the log directory return os.path.join(logDir, 'debug') def _getDebugDir() -> str: """Get the debug prompts directory path from configuration.""" baseDebugDir = _getBaseDebugDir() return os.path.join(baseDebugDir, 'prompts') def _getNextSequenceNumber() -> int: """Get the next sequence number by counting existing files.""" debugDir = _getDebugDir() if not os.path.exists(debugDir): return 1 # Count existing numbered files files = [f for f in os.listdir(debugDir) if f.startswith(('0', '1', '2', '3', '4', '5', '6', '7', '8', '9'))] return len(files) + 1 def writeDebugFile(content: str, fileType: str, documents: Optional[List] = None) -> None: """ Write debug content to a file with sequential numbering. Writes the content as-is since it's already the final integrated prompt. Includes document list labels for tracing enhancement. Only writes if debug logging is enabled via _isDebugEnabled() function. Args: content: The main content to write (already integrated) fileType: Type of file (e.g., 'prompt_final', 'response') documents: Optional list of documents for tracing """ try: # Check if debug logging is enabled if not _isDebugEnabled(): return debugDir = _getDebugDir() _ensureDir(debugDir) seqNum = _getNextSequenceNumber() ts = datetime.now(UTC).strftime('%Y%m%d-%H%M%S') # Add 3-digit sequence number for uniqueness tsWithSeq = f"{ts}-{seqNum:03d}" # Allow callers to pass an extension; if none, default to .txt if "." in (fileType or ""): filename = f"{tsWithSeq}-{fileType}" else: filename = f"{tsWithSeq}-{fileType}.txt" filepath = os.path.join(debugDir, filename) # Build content with document tracing debug_content = content # Add document list labels for tracing enhancement if documents: debug_content += "\n\n=== DOCUMENT LIST FOR TRACING ===\n" for i, doc in enumerate(documents): if hasattr(doc, 'fileName'): debug_content += f"Document {i+1}: {doc.fileName} ({doc.mimeType})\n" elif hasattr(doc, 'fileId'): debug_content += f"Document {i+1}: {doc.fileId} ({getattr(doc, 'mimeType', 'unknown')})\n" else: debug_content += f"Document {i+1}: {str(doc)[:100]}...\n" # Write the content with document tracing with open(filepath, 'w', encoding='utf-8') as f: f.write(debug_content) except Exception as e: # Don't log debug errors to avoid recursion pass def debugLogToFile(message: str, context: str = "DEBUG") -> None: """ Log debug message to file if debug logging is enabled. Args: message: Debug message to log context: Context identifier for the debug message """ try: # Check if debug logging is enabled if not _isDebugEnabled(): return # Get debug directory (use base debug dir, not prompts subdirectory) debug_dir = _getBaseDebugDir() _ensureDir(debug_dir) # Create debug file path debug_file = os.path.join(debug_dir, "debug_workflow.log") # Format the debug entry from modules.shared.timeUtils import getUtcTimestamp timestamp = getUtcTimestamp() debug_entry = f"[{timestamp}] [{context}] {message}\n" # Write to debug file with open(debug_file, "a", encoding="utf-8") as f: f.write(debug_entry) except Exception as e: # Don't log debug errors to avoid recursion pass