platform-core/modules/shared/debugLogger.py

# Copyright (c) 2026 PowerOn AG
# All rights reserved.
"""
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 resolve_app_log_dir() -> str:
    """Absolute filesystem path for ``APP_LOGGING_LOG_DIR``."""
    return _resolveLogDir()


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