Usage Logging

# Log Formats

## JSONL Schema

### Standard Entry
```json
{
  "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
```json
{
  "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
```json
{
  "metadata": {
    "files_processed": 10,
    "total_bytes": 50000,
    "file_types": [".py", ".md"]
  }
}
```

### API Operations
```json
{
  "metadata": {
    "model": "gemini-2.5-pro",
    "input_tokens": 5000,
    "output_tokens": 1000,
    "temperature": 0.7
  }
}
```

## Query Patterns

### By Time Range
```bash
# Last 24 hours
jq 'select(.timestamp > "2025-12-04")' usage.jsonl

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

### By Success/Failure
```bash
# Failures only
jq 'select(.success == false)' usage.jsonl

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

### By Token Usage
```bash
# High token operations
jq 'select(.tokens > 10000)' usage.jsonl

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


---
name: session-patterns
description: Patterns for session management in usage logging
estimated_tokens: 350
---

# Session Patterns

## Session Lifecycle

### Creation
```python
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
```python
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
```python
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
```python
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:

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

### Session Correlation
```python
def correlate_sessions(loggers: list[UsageLogger]) -> dict:
    """Correlate operations across services in same session."""
    all_ops = []
    for logger in loggers: