cleveragents-core/.opencode/agents/shared/logging.md

Standardized Logging Format

This document defines the mandatory logging format and utilities that all agents MUST use to ensure consistent, searchable, and debuggable logs across the system.

Log Format Specification

All agents MUST use this exact format for all log entries:

[AGENT:{name}][SESSION:{id}][ACTION:{action}][LEVEL:{level}] {message}

Components

• AGENT: The agent's name (e.g., implementation-worker, pr-self-reviewer)
• SESSION: Unique session identifier (use first 8 chars if long)
• ACTION: The current operation (e.g., clone, test, review, merge)
• LEVEL: Log level - INFO, WARNING, ERROR, DEBUG
• Message: Human-readable description of what happened

Examples

[AGENT:implementation-worker][SESSION:abc123de][ACTION:clone][LEVEL:INFO] Cloning repository to /tmp/impl-123
[AGENT:pr-self-reviewer][SESSION:def456gh][ACTION:review][LEVEL:WARNING] Found 3 CONTRIBUTING.md violations
[AGENT:behave-tester][SESSION:ghi789jk][ACTION:test][LEVEL:ERROR] Test execution failed: 5 scenarios failed

Python Implementation

import json
import time
from datetime import datetime
from typing import Optional, Dict, Any

class AgentLogger:
    """Standardized logger for all CleverAgents agents."""
    
    def __init__(self, agent_name: str, session_id: str, 
                 session_state_issue: Optional[int] = None,
                 owner: Optional[str] = None, 
                 repo: Optional[str] = None):
        """
        Initialize the logger.
        
        Args:
            agent_name: Name of the agent
            session_id: Unique session identifier
            session_state_issue: Forgejo issue number for persistence
            owner: Repository owner (for Forgejo persistence)
            repo: Repository name (for Forgejo persistence)
        """
        self.agent_name = agent_name
        self.session_id = session_id[:8] if len(session_id) > 8 else session_id
        self.session_state_issue = session_state_issue
        self.owner = owner
        self.repo = repo
        self.current_action = "init"
        
    def set_action(self, action: str):
        """Set the current action context."""
        self.current_action = action
        
    def _format_log(self, level: str, message: str, action: Optional[str] = None) -> str:
        """Format a log entry according to standards."""
        action_str = action or self.current_action
        return f"[AGENT:{self.agent_name}][SESSION:{self.session_id}][ACTION:{action_str}][LEVEL:{level}] {message}"
    
    def _persist_critical(self, log_line: str):
        """Persist critical logs to Forgejo."""
        if self.session_state_issue and self.owner and self.repo:
            try:
                comment = f"[LOG] Critical Event\n```\n{log_line}\n```\n\n---\n**Automated by CleverAgents Bot**"
                forgejo_create_issue_comment(self.owner, self.repo, 
                                           self.session_state_issue, comment)
            except:
                pass  # Best effort
    
    def info(self, message: str, action: Optional[str] = None):
        """Log an informational message."""
        log_line = self._format_log("INFO", message, action)
        print(log_line)
        return log_line
    
    def warning(self, message: str, action: Optional[str] = None):
        """Log a warning message."""
        log_line = self._format_log("WARNING", message, action)
        print(log_line)
        return log_line
    
    def error(self, message: str, action: Optional[str] = None, persist: bool = True):
        """Log an error message."""
        log_line = self._format_log("ERROR", message, action)
        print(log_line)
        
        if persist:
            self._persist_critical(log_line)
        
        return log_line
    
    def debug(self, message: str, action: Optional[str] = None):
        """Log a debug message."""
        log_line = self._format_log("DEBUG", message, action)
        print(log_line)
        return log_line
        
    def structured(self, action: str, data: Dict[str, Any], level: str = "INFO"):
        """Log structured data for automated analysis."""
        entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "agent": self.agent_name,
            "session": self.session_id,
            "action": action,
            "level": level,
            "data": data
        }
        
        # Standard format for visibility
        message = f"Structured data: {action}"
        self._format_log(level, message, action)
        
        # JSON format for parsing
        print(f"[STRUCTURED] {json.dumps(entry)}")
        
        # Persist if critical
        if level in ["ERROR", "CRITICAL"]:
            self._persist_critical(f"[STRUCTURED] {json.dumps(entry, indent=2)}")
        
        return entry

Usage Examples

Basic Usage

# Initialize at agent start
logger = AgentLogger(
    agent_name="implementation-worker",
    session_id=session_id,
    session_state_issue=SESSION_STATE_ISSUE,
    owner=owner,
    repo=repo
)

# Log with current action context
logger.set_action("setup")
logger.info("Validating credentials")
logger.info("Loading reference materials")

# Log with specific action
logger.info("Repository cloned successfully", action="clone")

# Log warnings
logger.warning("Retry attempt 2 of 3", action="push")

# Log errors (automatically persisted)
logger.error("Failed to merge PR: CI checks failing", action="merge")

# Log debug info
logger.debug(f"Queue depth: {len(queue)}", action="dispatch")

Structured Logging

# Log complex data for analysis
logger.structured("worker_dispatched", {
    "issue_number": 123,
    "worker_session": worker_session_id,
    "priority": "high",
    "attempt": 1
})

# Log performance metrics
logger.structured("operation_complete", {
    "operation": "subtask_implementation",
    "duration_seconds": 145.3,
    "lines_changed": 230,
    "files_modified": 5
}, level="INFO")

# Log errors with context
logger.structured("merge_failed", {
    "pr_number": 456,
    "reason": "insufficient_approvals",
    "current_approvals": 1,
    "required_approvals": 2
}, level="ERROR")

Integration with Error Handling

from shared.error_handling import safe_operation_with_recovery
from shared.logging import AgentLogger

logger = AgentLogger("implementation-worker", session_id)

def clone_with_logging(repo_url, target_dir):
    logger.set_action("clone")
    
    def do_clone():
        logger.info(f"Starting clone of {repo_url}")
        result = git_clone(repo_url, target_dir)
        logger.info("Clone completed successfully")
        return result
    
    def recovery(error_type, error):
        logger.warning(f"Recovery attempt for {error_type}: {error}")
        # Recovery logic...
    
    try:
        return safe_operation_with_recovery(
            "clone repository",
            do_clone,
            recovery
        )
    except OperationFailedError as e:
        logger.error(f"Clone failed after all attempts: {e}")
        raise

Context Manager for Actions

class LoggedAction:
    """Context manager for action-scoped logging."""
    
    def __init__(self, logger: AgentLogger, action: str):
        self.logger = logger
        self.action = action
        self.previous_action = None
        
    def __enter__(self):
        self.previous_action = self.logger.current_action
        self.logger.set_action(self.action)
        self.logger.info(f"Starting {self.action}")
        return self
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.logger.error(f"{self.action} failed: {exc_val}")
        else:
            self.logger.info(f"{self.action} completed")
        
        # Restore previous action
        self.logger.set_action(self.previous_action)
        
        # Don't suppress exceptions
        return False

# Usage
with LoggedAction(logger, "test_execution"):
    run_tests()  # Logs automatically scoped to this action

Log Aggregation Patterns

Health Signals

def log_health_signal(logger: AgentLogger, metrics: Dict[str, Any]):
    """Log standardized health signal."""
    
    logger.structured("health_signal", {
        "cycle": metrics.get("cycle", 0),
        "active_workers": metrics.get("active_workers", 0),
        "queue_depth": metrics.get("queue_depth", 0),
        "failure_rate": metrics.get("failure_rate", 0),
        "status": "active"
    }, level="INFO")

Operation Summaries

def log_operation_summary(logger: AgentLogger, operation: str, 
                         start_time: float, success: bool, 
                         details: Optional[Dict] = None):
    """Log operation completion with timing."""
    
    duration = time.time() - start_time
    
    data = {
        "operation": operation,
        "duration_seconds": round(duration, 2),
        "success": success
    }
    
    if details:
        data.update(details)
    
    level = "INFO" if success else "ERROR"
    logger.structured(f"{operation}_complete", data, level=level)

Best Practices

1. Always initialize logger at agent start - Don't create multiple instances
2. Set action context - Use set_action() for logical operation groups
3. Be consistent with action names - Use standard names across agents
4. Log state changes - Entry/exit of major operations
5. Include relevant context - But avoid logging sensitive data
6. Use appropriate levels - INFO for normal, WARNING for recoverable, ERROR for failures
7. Structure complex data - Use structured() for analysis-friendly logs

Standard Action Names

Use these standardized action names for consistency:

• init - Agent initialization
• setup - Environment setup
• clone - Repository cloning
• implement - Code implementation
• test - Test execution
• review - Code review
• merge - PR merging
• dispatch - Worker dispatch
• monitor - Health monitoring
• cleanup - Resource cleanup

Anti-Patterns to Avoid

# ❌ WRONG - Inconsistent format
print(f"Starting work on issue {issue_number}")

# ✓ CORRECT - Use logger
logger.info(f"Starting work on issue {issue_number}")

# ❌ WRONG - No action context
logger.info("Failed")  # Failed what?

# ✓ CORRECT - Clear action context
logger.error("Repository clone failed", action="clone")

# ❌ WRONG - Logging credentials
logger.info(f"Using PAT: {pat}")

# ✓ CORRECT - Log presence, not value
logger.info("Forgejo PAT validated")

# ❌ WRONG - Unstructured complex data
logger.info(f"Stats: workers={w}, queue={q}, failed={f}")

# ✓ CORRECT - Structured for parsing
logger.structured("pool_stats", {
    "workers": w,
    "queue_depth": q, 
    "failures": f
})

Integration Requirements

All agents MUST:

1. Import and initialize AgentLogger at startup
2. Use ONLY the logger for output (no raw print statements)
3. Set appropriate action context
4. Log all state transitions
5. Persist ERROR level logs to Forgejo
6. Use structured logging for metrics and complex data