temp/docs/architecture/decisions/ADR-010-logging-observability.md

ADR-010: Logging and Observability Strategy

Status

Accepted

Context

CleverAgents needs comprehensive logging and observability for:

• Debugging during development
• Production issue diagnosis
• Performance monitoring
• Usage analytics (opt-in)
• Audit trails for sensitive operations
• Distributed tracing in server mode

Requirements from discovery:

• Structured logging for 62 workflows
• Metrics for 33 retry patterns
• Tracing for async operations
• Privacy-preserving (no PII in logs)
• Optional telemetry integration

Decision

We will use structlog for structured logging with optional OpenTelemetry integration for distributed tracing and metrics.

Logging Architecture

# cleveragents.core.logging
import structlog
from structlog.processors import (
    add_log_level,
    TimeStamper,
    StackInfoRenderer,
    format_exc_info,
    CallsiteParameterAdder,
    dict_tracebacks
)

def configure_logging(
    level: str = "INFO",
    format: str = "json",  # json or console
    enable_tracing: bool = False
):
    """Configure structured logging for CleverAgents"""
    
    processors = [
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        TimeStamper(fmt="iso"),
        StackInfoRenderer(),
        format_exc_info,
        structlog.processors.UnicodeDecoder(),
    ]
    
    if enable_tracing:
        processors.append(inject_trace_context)
    
    if format == "json":
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())
    
    structlog.configure(
        processors=processors,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

Logging Patterns

# cleveragents.core.logging.patterns
import structlog
from functools import wraps
from typing import Any, Dict

logger = structlog.get_logger()

def log_operation(operation_type: str):
    """Decorator for logging operations with timing"""
    def decorator(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            log = logger.bind(
                operation=operation_type,
                function=func.__name__
            )
            log.info("operation_started")
            start = time.time()
            
            try:
                result = await func(*args, **kwargs)
                duration = time.time() - start
                log.info(
                    "operation_completed",
                    duration_ms=duration * 1000,
                    success=True
                )
                return result
            except Exception as e:
                duration = time.time() - start
                log.error(
                    "operation_failed",
                    duration_ms=duration * 1000,
                    error_type=type(e).__name__,
                    error_message=str(e)
                )
                raise
        
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            # Similar for sync functions
            pass
        
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    return decorator

Privacy and Scrubbing

# cleveragents.core.logging.privacy
import re
from typing import Any, Dict

class SensitiveDataScrubber:
    """Scrub sensitive data from logs"""
    
    PATTERNS = {
        'api_key': re.compile(r'(api[_-]?key|token|secret)["\']?\s*[:=]\s*["\']?([^"\'\s]+)'),
        'email': re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
        'credit_card': re.compile(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'),
    }
    
    def scrub(self, data: Any) -> Any:
        if isinstance(data, str):
            for pattern_name, pattern in self.PATTERNS.items():
                data = pattern.sub(f'[REDACTED_{pattern_name.upper()}]', data)
        elif isinstance(data, dict):
            return {k: self.scrub(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [self.scrub(item) for item in data]
        return data

# Add to logging configuration
processors.append(SensitiveDataScrubber().scrub)

Metrics Collection

# cleveragents.core.metrics
from opentelemetry import metrics
from opentelemetry.exporter.prometheus import PrometheusMetricReader

def setup_metrics():
    """Setup OpenTelemetry metrics"""
    meter = metrics.get_meter("cleveragents")
    
    # Counter for operations
    operation_counter = meter.create_counter(
        "cleveragents.operations.total",
        description="Total number of operations",
        unit="1"
    )
    
    # Histogram for operation duration
    duration_histogram = meter.create_histogram(
        "cleveragents.operation.duration",
        description="Operation duration",
        unit="ms"
    )
    
    # Gauge for active plans
    active_plans = meter.create_up_down_counter(
        "cleveragents.plans.active",
        description="Number of active plans",
        unit="1"
    )
    
    return {
        'operations': operation_counter,
        'duration': duration_histogram,
        'active_plans': active_plans
    }

Distributed Tracing

# cleveragents.core.tracing
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
from contextlib import contextmanager

tracer = trace.get_tracer("cleveragents")

@contextmanager
def trace_operation(name: str, attributes: Dict[str, Any] = None):
    """Context manager for tracing operations"""
    with tracer.start_as_current_span(name) as span:
        if attributes:
            span.set_attributes(attributes)
        
        try:
            yield span
        except Exception as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

Log Aggregation

# cleveragents.core.logging.aggregation
class LogContext:
    """Aggregate logs for an operation"""
    
    def __init__(self, operation_id: str):
        self.operation_id = operation_id
        self.logger = logger.bind(operation_id=operation_id)
    
    def __enter__(self):
        self.logger.info("context_started")
        return self.logger
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.logger.error(
                "context_failed",
                error_type=exc_type.__name__,
                error_message=str(exc_val)
            )
        else:
            self.logger.info("context_completed")

Usage Examples

# Application service with logging
class PlanService:
    @log_operation("plan.create")
    @trace_operation("create_plan")
    async def create_plan(self, name: str, **kwargs):
        log = logger.bind(plan_name=name)
        log.info("creating_plan")
        
        # Validate
        log.debug("validating_input")
        validated = self.validate(name, **kwargs)
        
        # Create
        with LogContext(f"plan_create_{name}") as ctx_log:
            ctx_log.info("saving_to_database")
            plan = await self.repository.create(validated)
            
            ctx_log.info("plan_created", plan_id=plan.id)
            
            # Record metric
            metrics['operations'].add(1, {"operation": "plan.create"})
            
        return plan

Consequences

Positive

• Structured logs are queryable and parseable
• Automatic context propagation
• Privacy protection built-in
• Optional telemetry for self-hosted deployments
• Rich debugging information
• Performance metrics included

Negative

• Additional dependencies (structlog, OpenTelemetry)
• Learning curve for structured logging
• Potential performance impact if over-logged
• Need to maintain scrubbing patterns

Neutral

• Requires discipline about what to log
• Need to document log schemas
• Configuration complexity for different environments

Log Levels and Guidelines

Level	Usage	Example
DEBUG	Detailed diagnostic info	Input validation steps
INFO	Normal operations	Plan created, Build started
WARNING	Recoverable issues	Retry attempted, Fallback used
ERROR	Failures requiring attention	API call failed, Database error
CRITICAL	System-wide issues	Out of memory, Corruption detected

Privacy Guidelines

1. Never log:

   • Passwords or API keys
   • Full file contents
   • Personal information (emails, names)
   • Raw AI model responses with user data

2. Always scrub:

   • Environment variables
   • Configuration dumps
   • Stack traces with sensitive data

3. Consider redacting:

   • File paths (may reveal usernames)
   • Plan/project names (if sensitive)
   • Model selections (if proprietary)

References

• structlog Documentation
• OpenTelemetry Python
• 12-Factor App Logging
• OWASP Logging Cheat Sheet