Logger

Overview

The logger module provides structured JSON logging with automatic file rotation and session-based tracking. All logs are output in JSON format with consistent fields for easy parsing and analysis. File: common/logger.py

Features

JSON Format

Every log line is valid JSON for easy machine parsing

File Rotation

Automatic log rotation based on size with configurable backup count

Session Tracking

Associate logs with specific agent sessions

Dual Output

Logs written to both stdout and rotating files

Functions

get_logger()

Get or create a JSON logger for a component with optional session tracking.

get_logger(component: str, session_id: str = None) -> logging.Logger

component

str

required

Component name (e.g., ‘agent’, ‘server’, ‘transport’). Used as log filename and in structured output.

session_id

str

Optional session identifier to track logs for specific agent sessions

return

logging.Logger

Configured logger instance with JSON formatting and rotation

Logger Configuration:

Log level from config.LOG_LEVEL (default: INFO)
Dual handlers: stdout + rotating file
File location: {config.LOG_DIR}/{component}.log
Rotation: config.LOG_MAX_BYTES per file, config.LOG_BACKUP_COUNT backups
Caching: Loggers are cached by component:session_id key

Example:

from common.logger import get_logger

# Component-level logger
logger = get_logger('server')
logger.info('Server starting')

# Session-specific logger
logger = get_logger('agent', session_id='abc123')
logger.info('Task completed', task_id='task-456', duration_ms=142)

update_session()

Create a new logger for the same component with an updated session ID.

update_session(logger: logging.Logger, session_id: str) -> logging.Logger

logger

logging.Logger

required

Existing logger instance

session_id

str

required

New session identifier

return

logging.Logger

New logger instance for the same component with updated session ID

Use Case:

Agent receives session ID after check-in
Update logger to include session ID in all subsequent logs

Example:

from common.logger import get_logger, update_session

# Initial logger without session
logger = get_logger('agent')
logger.info('Checking in to server')

# Update after receiving session ID
session_id = checkin_response['session_id']
logger = update_session(logger, session_id)
logger.info('Session established')  # Now includes session_id

Log Format

Each log entry is a JSON object with the following structure:

{
  "timestamp": "2026-03-11T15:42:30.123456+07:00",
  "level": "INFO",
  "component": "agent",
  "session_id": "abc123",
  "message": "Task completed",
  "task_id": "task-456",
  "duration_ms": 142
}

Standard Fields

timestamp

str

ISO 8601 timestamp in UTC+7 timezone with microsecond precision

level

str

Log level: DEBUG, INFO, WARNING, ERROR, or CRITICAL

component

str

Component name specified in get_logger()

session_id

str

Session identifier if provided, otherwise null

message

str

Log message text

Extra Fields

Any additional fields passed via the extra kwarg are automatically included:

logger.info(
    'Task completed',
    extra={
        'task_id': 'task-456',
        'duration_ms': 142,
        'exit_code': 0
    }
)

Exception Logging

When logging exceptions, traceback is automatically included:

try:
    risky_operation()
except Exception as e:
    logger.error('Operation failed', exc_info=True)

Output:

{
  "timestamp": "2026-03-11T15:42:30.123456+07:00",
  "level": "ERROR",
  "component": "agent",
  "message": "Operation failed",
  "exception": "Traceback (most recent call last):\n  File ..."
}

Configuration

Logger behavior is controlled by common/config.py:

LOG_LEVEL        = 'INFO'              # Minimum log level
LOG_DIR          = 'logs'              # Log directory
LOG_MAX_BYTES    = 5 * 1024 * 1024     # 5 MB per file
LOG_BACKUP_COUNT = 3                    # Keep 3 rotated files

Log Levels

Level	Use Case
`DEBUG`	Detailed diagnostic information for development
`INFO`	General informational messages about normal operation
`WARNING`	Warning messages for potentially problematic situations
`ERROR`	Error messages for serious problems
`CRITICAL`	Critical messages for very serious errors

File Organization

Logs are organized by component:

logs/
├── agent.log         # Current agent logs
├── agent.log.1       # Previous rotation
├── agent.log.2       # Older rotation
├── server.log        # Current server logs
├── server.log.1
└── transport.log     # Current transport logs

Caching Behavior

Loggers are cached internally to prevent duplicate handlers:

Cache key: {component}:{session_id}
First call creates logger with handlers
Subsequent calls return cached instance
Each unique component:session_id pair gets its own logger

Example:

logger1 = get_logger('agent', 'session1')  # Creates new logger
logger2 = get_logger('agent', 'session1')  # Returns cached logger
assert logger1 is logger2  # True

logger3 = get_logger('agent', 'session2')  # Creates different logger
assert logger1 is logger3  # False

Usage Examples

Basic Logging

from common.logger import get_logger

logger = get_logger('server')

logger.debug('Debug message')
logger.info('Server started on port 8443')
logger.warning('High memory usage detected')
logger.error('Failed to connect to database')
logger.critical('System shutdown imminent')

Session-Based Logging

from common.logger import get_logger, update_session

# Start without session
logger = get_logger('agent')
logger.info('Agent starting')

# Update after check-in
session_id = perform_checkin()
logger = update_session(logger, session_id)
logger.info('Session established', session_id=session_id)

# All subsequent logs include session_id
logger.info('Polling for tasks')

Structured Logging with Extra Fields

logger = get_logger('agent', session_id='abc123')

logger.info(
    'Task executed',
    extra={
        'task_id': 'task-789',
        'command': 'whoami',
        'exit_code': 0,
        'duration_ms': 142,
        'stdout_len': 15,
        'stderr_len': 0
    }
)

Error Handling

logger = get_logger('transport')

try:
    send_message(envelope)
except ConnectionError as e:
    logger.error(
        'Failed to send message',
        exc_info=True,
        extra={'host': config.SERVER_HOST, 'port': config.SERVER_PORT}
    )

Implementation Details

Timezone

Logs use UTC+7 timezone:

_TZ_UTC7 = timezone(timedelta(hours=7))
timestamp = datetime.now(_TZ_UTC7).isoformat()

Reserved Keys

The following keys are reserved and excluded from extra fields:

_RESERVED_KEYS = {
    'name', 'msg', 'args', 'levelname', 'levelno', 'pathname',
    'filename', 'module', 'exc_info', 'exc_text', 'stack_info',
    'lineno', 'funcName', 'created', 'msecs', 'relativeCreated',
    'thread', 'threadName', 'processName', 'process', 'message',
    'taskName',
}

Formatter Implementation

class _JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            'timestamp':  datetime.now(_TZ_UTC7).isoformat(),
            'level':      record.levelname,
            'component':  self._component,
            'session_id': self._session_id,
            'message':    record.getMessage(),
        }
        
        # Add extra fields
        for key, value in record.__dict__.items():
            if key not in _RESERVED_KEYS:
                entry[key] = value
        
        # Add exception info
        if record.exc_info:
            entry['exception'] = self.formatException(record.exc_info)
        
        return json.dumps(entry)

Best Practices

Use Structured Logging

Always pass structured data as extra fields instead of string formatting:

# Good
logger.info('Task completed', extra={'task_id': task_id, 'duration_ms': duration})

# Bad
logger.info(f'Task {task_id} completed in {duration}ms')

Include Session IDs

Always use session-aware loggers for agent operations:

logger = get_logger('agent', session_id=session_id)

Log Exceptions Properly

Use exc_info=True to include tracebacks:

try:
    operation()
except Exception as e:
    logger.error('Operation failed', exc_info=True)

Choose Appropriate Log Levels

DEBUG: Detailed info for troubleshooting
INFO: Normal operations and state changes
WARNING: Degraded operation or potential issues
ERROR: Operation failures that need attention
CRITICAL: System-level failures

Dependencies

import json
import logging
import logging.handlers
import os
from datetime import datetime, timezone, timedelta

from common import config

Server

Agent

Common

Transport

Evasion

Telemetry

Overview

Features

JSON Format

File Rotation

Session Tracking

Dual Output

Functions

get_logger()

update_session()

Log Format

Standard Fields

Extra Fields

Exception Logging

Configuration

Log Levels

File Organization

Caching Behavior

Usage Examples

Basic Logging

Session-Based Logging

Structured Logging with Extra Fields

Error Handling

Implementation Details

Timezone

Reserved Keys

Formatter Implementation

Best Practices

Dependencies

See Also

Build docs developers (and LLMs) love

Server

Agent

Common

Transport

Evasion

Telemetry

​Overview

​Features

JSON Format

File Rotation

Session Tracking

Dual Output

​Functions

​get_logger()

​update_session()

​Log Format

​Standard Fields

​Extra Fields

​Exception Logging

​Configuration

​Log Levels

​File Organization

​Caching Behavior

​Usage Examples

​Basic Logging

​Session-Based Logging

​Structured Logging with Extra Fields

​Error Handling

​Implementation Details

​Timezone

​Reserved Keys

​Formatter Implementation

​Best Practices

​Dependencies

​See Also

Build docs developers (and LLMs) love

Overview

Features

Functions

get_logger()

update_session()

Log Format

Standard Fields

Extra Fields

Exception Logging

Configuration

Log Levels

File Organization

Caching Behavior

Usage Examples

Basic Logging

Session-Based Logging

Structured Logging with Extra Fields

Error Handling

Implementation Details

Timezone

Reserved Keys

Formatter Implementation

Best Practices

Dependencies

See Also