Debugging and Observability Patterns

Master the art of debugging and observing agents through comprehensive patterns for development troubleshooting, production monitoring, and performance optimization.

Overview

As AI-assisted development becomes central to modern workflows, understanding and debugging agent behavior is crucial. This guide covers patterns for effective debugging, observability, and monitoring in various environments.

Core Debugging Patterns

1. Hook-Based Debug Logging

Create comprehensive logging for all hook events:

#!/bin/bash
# Save as: .claude/hooks/debug_logger.sh
 
LOG_FILE="$HOME/.claude/debug.log"
JSON_INPUT=$(cat)
 
# Log everything with timestamp
{
    echo "=== [$(date -u +%Y-%m-%dT%H:%M:%SZ)] Hook Debug ==="
    echo "Event: $CLAUDE_HOOK_EVENT"
    echo "Tool: $CLAUDE_TOOL_NAME"
    echo "Files: $CLAUDE_FILE_PATHS"
    echo "JSON Input:"
    echo "$JSON_INPUT" | jq '.'
    echo "========================================"
} >> "$LOG_FILE"
 
# Pass through for Claude
echo "$JSON_INPUT"
exit 0

2. Interactive Debug Mode

Enable breakpoint-style debugging:

#!/usr/bin/env python3
import json
import sys
import os
 
def debug_breakpoint(data, context=""):
    """Pause execution for inspection"""
    if os.environ.get('CLAUDE_DEBUG') == 'true':
        print(f"🔍 Debug: {context}", file=sys.stderr)
        print(f"Data: {json.dumps(data, indent=2)}", file=sys.stderr)
        print("Press Enter to continue...", file=sys.stderr)
        input()
 
# Usage in your hook
data = json.load(sys.stdin)
debug_breakpoint(data, "Before processing")
# Your processing logic
debug_breakpoint(result, "After processing")

3. Error Capture and Recovery

Comprehensive error handling pattern:

#!/usr/bin/env python3
import json
import sys
import traceback
import logging
from datetime import datetime
 
# Configure logging
logging.basicConfig(
    filename='/tmp/claude_hooks_errors.log',
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
 
def safe_hook_execution(func):
    """Decorator for safe hook execution"""
    def wrapper():
        try:
            data = json.load(sys.stdin)
            result = func(data)
            
            # Success
            logging.info(f"Hook completed: {func.__name__}")
            print(json.dumps(result))
            return 0
            
        except json.JSONDecodeError as e:
            logging.error(f"JSON error: {e}")
            error_response = {
                "continue": False,
                "stopReason": f"Invalid JSON: {str(e)}"
            }
            print(json.dumps(error_response))
            return 2
            
        except Exception as e:
            logging.error(f"Unexpected error: {e}")
            logging.error(traceback.format_exc())
            
            # Graceful failure
            error_response = {
                "continue": False,
                "stopReason": f"Hook error: {type(e).__name__}"
            }
            print(json.dumps(error_response))
            return 2
    
    return wrapper
 
@safe_hook_execution
def process_hook(data):
    # Your hook logic here
    return {"continue": True}
 
if __name__ == '__main__':
    sys.exit(process_hook())

OpenTelemetry Integration Patterns

1. Basic OpenTelemetry Setup

Configure the tool for observability:

# Environment configuration
export CLAUDE_CODE_ENABLE_TELEMETRY=1
export OTEL_METRICS_EXPORTER=otlp
export OTEL_LOGS_EXPORTER=otlp
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
 
# Custom attributes for team identification
export OTEL_RESOURCE_ATTRIBUTES="department=engineering,team.id=platform,project=webapp"
 
# Debug configuration (short intervals)
export OTEL_METRIC_EXPORT_INTERVAL=10000  # 10 seconds
export OTEL_LOGS_EXPORT_INTERVAL=5000     # 5 seconds

2. Console Debugging Pattern

For immediate feedback during development:

# Quick console debugging setup
export CLAUDE_CODE_ENABLE_TELEMETRY=1
export OTEL_METRICS_EXPORTER=console
export OTEL_LOGS_EXPORTER=console
export OTEL_METRIC_EXPORT_INTERVAL=1000  # 1 second for debugging
 
# Run with verbose output
claude --verbose "Help me debug this function"

3. OpenTelemetry Collector Integration

Deploy a collector for production debugging:

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
 
processors:
  batch:
    timeout: 10s
  
  # Add custom attributes
  attributes:
    actions:
      - key: environment
        value: production
        action: insert
 
exporters:
  prometheus:
    endpoint: "0.0.0.0:8889"
  
  loki:
    endpoint: "http://loki:3100/loki/api/v1/push"
  
  debug:
    verbosity: detailed
 
service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [batch, attributes]
      exporters: [prometheus, debug]
    
    logs:
      receivers: [otlp]
      processors: [batch, attributes]
      exporters: [loki, debug]

Production Observability Patterns

1. Comprehensive Metrics Stack

Deploy the full observability stack:

# Clone the observability solution
git clone https://github.com/ColeMurray/claude-code-otel
cd claude-code-otel
 
# Start the stack
docker-compose up -d
 
# Configure Claude Code
export CLAUDE_CODE_ENABLE_TELEMETRY=1
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317

Key metrics tracked:

• Active sessions and concurrent agents
• Token usage and costs by model
• Tool performance and success rates
• Latency and error rates
• Lines of code changed
• Commit and PR metrics

2. Real-time Event Monitoring

Monitor events in real-time:

#!/usr/bin/env python3
"""Real-time Claude Code event monitor"""
 
import json
import asyncio
from datetime import datetime
from collections import deque
 
class EventMonitor:
    def __init__(self, max_events=100):
        self.events = deque(maxlen=max_events)
        self.event_counts = {}
    
    async def process_event(self, event):
        """Process incoming event"""
        event_type = event.get('event_type', 'unknown')
        
        # Update counts
        self.event_counts[event_type] = self.event_counts.get(event_type, 0) + 1
        
        # Store event
        self.events.append({
            'timestamp': datetime.utcnow().isoformat(),
            'type': event_type,
            'data': event
        })
        
        # Alert on specific patterns
        if event_type == 'tool_error':
            await self.alert_on_error(event)
    
    async def alert_on_error(self, event):
        """Send alerts for critical errors"""
        error_rate = self.calculate_error_rate()
        if error_rate > 0.1:  # 10% error rate
            print(f"⚠️ High error rate detected: {error_rate:.1%}")
    
    def calculate_error_rate(self):
        """Calculate recent error rate"""
        recent_events = list(self.events)[-20:]  # Last 20 events
        if not recent_events:
            return 0
        
        errors = sum(1 for e in recent_events if 'error' in e['type'])
        return errors / len(recent_events)
    
    def get_dashboard_data(self):
        """Get data for dashboard display"""
        return {
            'event_counts': self.event_counts,
            'recent_events': list(self.events)[-10:],
            'error_rate': self.calculate_error_rate(),
            'total_events': sum(self.event_counts.values())
        }

3. Cost and Performance Tracking

Track and optimize costs:

#!/usr/bin/env python3
"""Cost and performance tracker"""
 
import json
from datetime import datetime, timedelta
from collections import defaultdict
 
class CostTracker:
    # Approximate token costs (update as needed)
    TOKEN_COSTS = {
        'claude-3-opus': {'input': 0.015, 'output': 0.075},
        'claude-3-sonnet': {'input': 0.003, 'output': 0.015},
        'claude-3-haiku': {'input': 0.00025, 'output': 0.00125}
    }
    
    def __init__(self):
        self.usage_by_user = defaultdict(lambda: defaultdict(int))
        self.usage_by_project = defaultdict(lambda: defaultdict(int))
    
    def track_usage(self, event):
        """Track token usage from event"""
        if event.get('event_type') != 'api_request_complete':
            return
        
        user = event.get('user_id', 'unknown')
        project = event.get('project', 'default')
        model = event.get('model', 'claude-3-sonnet')
        
        input_tokens = event.get('input_tokens', 0)
        output_tokens = event.get('output_tokens', 0)
        
        # Update usage
        self.usage_by_user[user]['input_tokens'] += input_tokens
        self.usage_by_user[user]['output_tokens'] += output_tokens
        
        self.usage_by_project[project]['input_tokens'] += input_tokens
        self.usage_by_project[project]['output_tokens'] += output_tokens
        
        # Calculate cost
        cost = self.calculate_cost(model, input_tokens, output_tokens)
        self.usage_by_user[user]['cost'] += cost
        self.usage_by_project[project]['cost'] += cost
    
    def calculate_cost(self, model, input_tokens, output_tokens):
        """Calculate cost for token usage"""
        rates = self.TOKEN_COSTS.get(model, self.TOKEN_COSTS['claude-3-sonnet'])
        
        input_cost = (input_tokens / 1000) * rates['input']
        output_cost = (output_tokens / 1000) * rates['output']
        
        return input_cost + output_cost
    
    def get_cost_report(self, period_days=7):
        """Generate cost report for the period"""
        return {
            'by_user': dict(self.usage_by_user),
            'by_project': dict(self.usage_by_project),
            'total_cost': sum(u['cost'] for u in self.usage_by_user.values()),
            'period_days': period_days
        }

Advanced Debugging Techniques

1. Session Replay Pattern

Capture and replay sessions:

#!/usr/bin/env python3
"""Session recorder and replayer"""
 
import json
import time
from pathlib import Path
 
class SessionRecorder:
    def __init__(self, session_dir="./claude_sessions"):
        self.session_dir = Path(session_dir)
        self.session_dir.mkdir(exist_ok=True)
        self.current_session = None
    
    def start_session(self, session_id):
        """Start recording a new session"""
        self.current_session = {
            'id': session_id,
            'start_time': time.time(),
            'events': []
        }
    
    def record_event(self, event_type, data):
        """Record an event in the current session"""
        if not self.current_session:
            return
        
        event = {
            'timestamp': time.time(),
            'type': event_type,
            'data': data
        }
        
        self.current_session['events'].append(event)
    
    def save_session(self):
        """Save the current session to disk"""
        if not self.current_session:
            return
        
        session_file = self.session_dir / f"{self.current_session['id']}.json"
        
        with open(session_file, 'w') as f:
            json.dump(self.current_session, f, indent=2)
    
    def replay_session(self, session_id, speed=1.0):
        """Replay a recorded session"""
        session_file = self.session_dir / f"{session_id}.json"
        
        with open(session_file, 'r') as f:
            session = json.load(f)
        
        start_time = session['events'][0]['timestamp']
        
        for event in session['events']:
            # Calculate delay
            delay = (event['timestamp'] - start_time) / speed
            time.sleep(delay)
            
            # Process event
            print(f"[{delay:.2f}s] {event['type']}: {event['data']}")
            
            # Update start_time for next iteration
            start_time = event['timestamp']

2. Performance Profiling Pattern

Profile performance:

#!/bin/bash
# Performance profiling wrapper
 
PROFILE_DIR="$HOME/.claude/profiles"
mkdir -p "$PROFILE_DIR"
 
# Start profiling
START_TIME=$(date +%s.%N)
START_MEMORY=$(ps -o rss= -p $$)
 
# Run with profiling
CLAUDE_CODE_ENABLE_TELEMETRY=1 \
OTEL_METRICS_EXPORTER=console \
time -v claude "$@" 2>&1 | tee "$PROFILE_DIR/profile_$(date +%Y%m%d_%H%M%S).log"
 
# Calculate metrics
END_TIME=$(date +%s.%N)
END_MEMORY=$(ps -o rss= -p $$)
 
DURATION=$(echo "$END_TIME - $START_TIME" | bc)
MEMORY_DELTA=$((END_MEMORY - START_MEMORY))
 
echo "Performance Summary:"
echo "  Duration: ${DURATION}s"
echo "  Memory Delta: ${MEMORY_DELTA}KB"

3. Distributed Tracing Pattern

Implement distributed tracing for multi-agent scenarios:

#!/usr/bin/env python3
"""Distributed tracing for agents"""
 
import uuid
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
 
tracer = trace.get_tracer(__name__)
 
class AgentTracer:
    def __init__(self):
        self.active_spans = {}
    
    def start_agent_span(self, agent_id, operation):
        """Start a span for an agent operation"""
        span = tracer.start_span(
            f"claude_agent.{operation}",
            attributes={
                "agent.id": agent_id,
                "agent.operation": operation
            }
        )
        
        self.active_spans[agent_id] = span
        return span
    
    def add_event(self, agent_id, event_name, attributes=None):
        """Add an event to the agent's span"""
        span = self.active_spans.get(agent_id)
        if span:
            span.add_event(event_name, attributes=attributes or {})
    
    def end_agent_span(self, agent_id, success=True):
        """End the agent's span"""
        span = self.active_spans.pop(agent_id, None)
        if span:
            if success:
                span.set_status(Status(StatusCode.OK))
            else:
                span.set_status(Status(StatusCode.ERROR))
            span.end()
 
# Usage example
tracer = AgentTracer()
 
# Start tracing an agent
span = tracer.start_agent_span("agent_123", "code_generation")
tracer.add_event("agent_123", "started_analysis", {"files": 10})
tracer.add_event("agent_123", "completed_generation", {"lines": 150})
tracer.end_agent_span("agent_123", success=True)

Best Practices

1. Debugging Workflow

1. Start Simple: Use console exporters for immediate feedback
2. Gradual Complexity: Add more sophisticated debugging as needed
3. Production Ready: Deploy full observability stack for production
4. Continuous Monitoring: Set up alerts for anomalies

2. Performance Guidelines

• Keep hook execution under 1 second
• Use background processing for heavy operations
• Implement caching for repeated operations
• Monitor token usage to control costs

3. Security Considerations

• Redact sensitive information in logs
• Use secure endpoints for telemetry
• Implement access controls for debugging tools
• Regularly audit debug logs

4. Team Collaboration

• Use custom attributes to segment by team
• Share debugging dashboards
• Document debugging procedures
• Create runbooks for common issues

Common Issues and Solutions

Issue: No Telemetry Data

# Verify configuration
echo $CLAUDE_CODE_ENABLE_TELEMETRY  # Should be 1
 
# Test with console exporter
export OTEL_METRICS_EXPORTER=console
export OTEL_METRIC_EXPORT_INTERVAL=1000
 
# Check for errors
claude --verbose "test command"

Issue: High Error Rates

# Implement circuit breaker pattern
class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.is_open = False

Issue: Performance Degradation

# Profile specific operations
CLAUDE_PROFILE=1 claude "complex task"
 
# Analyze flame graphs
flamegraph.pl perf.data > flamegraph.svg

Integration Examples

Datadog Integration

export OTEL_EXPORTER_OTLP_ENDPOINT=http://datadog-agent:4317
export OTEL_EXPORTER_OTLP_HEADERS="api-key=YOUR_DATADOG_API_KEY"

Prometheus + Grafana

# prometheus.yml
global:
  scrape_interval: 15s
 
scrape_configs:
  - job_name: 'claude-code'
    static_configs:
      - targets: ['localhost:8889']

Resources

• OpenTelemetry GitHub
• Multi-Agent Observability
• Official Monitoring Docs
• Datadog Integration Guide

Related Patterns

• Error Recovery Patterns
• Performance Optimization
• Multi-Agent Patterns
• Hooks System Guide