Usage Logging

Table of Contents

• Overview
• When to Use
• Core Concepts
• Session Management
• Log Entry Structure
• Quick Start
• Initialize Logger
• Log Operations
• Query Usage
• Integration Pattern
• Log Storage
• Detailed Resources
• Exit Criteria

Usage Logging

Overview

Session-aware logging infrastructure for tracking operations across plugins. Provides structured JSONL logging with automatic session management for audit trails and analytics.

When To Use

• Need audit trails for operations
• Tracking costs across sessions
• Building usage analytics
• Debugging with operation history

When NOT To Use

• Simple operations without logging needs

Core Concepts

Session Management

Sessions group related operations:

• Auto-created on first operation
• Timeout after 1 hour of inactivity
• Unique session IDs for tracking

Log Entry Structure

{
  "timestamp": "2025-12-05T10:30:00Z",
  "session_id": "session_1733394600",
  "service": "my-service",
  "operation": "analyze_files",
  "tokens": 5000,
  "success": true,
  "duration_seconds": 2.5,
  "metadata": {}
}

Verification: Run the command with --help flag to verify availability.

Quick Start

Initialize Logger

from leyline.usage_logger import UsageLogger

logger = UsageLogger(service="my-service")

Verification: Run the command with --help flag to verify availability.

Log Operations

logger.log_usage(
    operation="analyze_files",
    tokens=5000,
    success=True,
    duration=2.5,
    metadata={"files": 10}
)

Verification: Run the command with --help flag to verify availability.

Query Usage

# Recent operations
recent = logger.get_recent_operations(hours=24)

# Usage summary
summary = logger.get_usage_summary(days=7)
print(f"Total tokens: {summary['total_tokens']}")
print(f"Total cost: ${summary['estimated_cost']:.2f}")

# Recent errors
errors = logger.get_recent_errors(count=10)

Verification: Run the command with --help flag to verify availability.

Integration Pattern

# In your skill's frontmatter
dependencies: [leyline:usage-logging]

Verification: Run the command with --help flag to verify availability.

Standard integration flow: 1. Initialize logger for your service 2. Log operations after completion 3. Query for analytics and debugging

Log Storage

Default location: ~/.claude/leyline/usage/{service}.jsonl

# View recent logs
tail -20 ~/.claude/leyline/usage/my-service.jsonl | jq .

# Query by date
grep "2025-12-05" ~/.claude/leyline/usage/my-service.jsonl

Verification: Run the command with --help flag to verify availability.

Detailed Resources

• Session Patterns: See modules/session-patterns.md for session management
• Log Formats: See modules/log-formats.md for structured formats

Exit Criteria

• Operation logged with all required fields
• Session tracked for grouping
• Logs queryable for analytics

Log Formats

JSONL Schema

Standard Entry

{
  "timestamp": "ISO-8601 datetime",
  "session_id": "session_UNIX_TIMESTAMP",
  "service": "service-name",
  "operation": "operation-name",
  "tokens": 0,
  "success": true,
  "duration_seconds": 0.0,
  "metadata": {}
}

Error Entry

{
  "timestamp": "ISO-8601 datetime",
  "session_id": "session_xxx",
  "service": "service-name",
  "operation": "operation-name",
  "tokens": 0,
  "success": false,
  "error_type": "RateLimitError",
  "error_message": "Rate limit exceeded",
  "duration_seconds": 0.5
}

Metadata Patterns

File Operations

{
  "metadata": {
    "files_processed": 10,
    "total_bytes": 50000,
    "file_types": [".py", ".md"]
  }
}

API Operations

{
  "metadata": {
    "model": "gemini-2.5-pro",
    "input_tokens": 5000,
    "output_tokens": 1000,
    "temperature": 0.7
  }
}

Query Patterns

By Time Range

# Last 24 hours
jq 'select(.timestamp > "2025-12-04")' usage.jsonl

# Specific date
grep "2025-12-05" usage.jsonl | jq .

By Success/Failure

# Failures only
jq 'select(.success == false)' usage.jsonl

# Success rate
jq -s '[.[] | .success] | add / length' usage.jsonl

By Token Usage

# High token operations
jq 'select(.tokens > 10000)' usage.jsonl

# Total tokens
jq -s '[.[] | .tokens] | add' usage.jsonl

Session Patterns

Session Lifecycle

Creation

def _get_or_create_session(self) -> str:
    """Get existing session or create new one."""
    session_file = self.log_dir / "session.json"

    if session_file.exists():
        session = json.loads(session_file.read_text())
        # Check if session is still active (1 hour timeout)
        if time.time() - session["last_activity"] < 3600:
            return session["session_id"]

    # Create new session
    session_id = f"session_{int(time.time())}"
    session_file.write_text(json.dumps({
        "session_id": session_id,
        "created_at": time.time(),
        "last_activity": time.time()
    }))
    return session_id

Activity Tracking

def _update_session_activity(self):
    """Update session last activity timestamp."""
    session_file = self.log_dir / "session.json"
    if session_file.exists():
        session = json.loads(session_file.read_text())
        session["last_activity"] = time.time()
        session_file.write_text(json.dumps(session))

Session Grouping

Query by Session

def get_session_operations(self, session_id: str) -> list[dict]:
    """Get all operations for a session."""
    operations = []
    for line in self.log_file.read_text().splitlines():
        entry = json.loads(line)
        if entry.get("session_id") == session_id:
            operations.append(entry)
    return operations

Session Statistics

def get_session_stats(self, session_id: str) -> dict:
    """Get statistics for a session."""
    ops = self.get_session_operations(session_id)
    return {
        "operation_count": len(ops),
        "total_tokens": sum(o.get("tokens", 0) for o in ops),
        "success_rate": sum(1 for o in ops if o["success"]) / len(ops),
        "total_duration": sum(o.get("duration_seconds", 0) for o in ops)
    }

Cross-Service Sessions

Shared Session IDs

When operations span multiple services:

# Pass session_id between services
logger1 = UsageLogger(service="gemini", session_id=shared_session)
logger2 = UsageLogger(service="qwen", session_id=shared_session)

Session Correlation

def correlate_sessions(loggers: list[UsageLogger]) -> dict:
    """Correlate operations across services in same session."""
    all_ops = []
    for logger in loggers:
        all_ops.extend(logger.get_recent_operations(hours=1))

    # Sort by timestamp
    all_ops.sort(key=lambda x: x["timestamp"])
    return all_ops