Building a Session Logging System — Central Command Architecture
When running multiple sessions in Claude Control, tracking "which session did what" becomes critical. This post documents the process of building a session logging system and a central command architecture.
Logging Architecture
All session inputs and outputs are stored in a structured format. Rather than plain text, each entry also captures metadata: timestamps, token usage, execution time, and so on.
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
class LogLevel(str, Enum):
INPUT = "input"
OUTPUT = "output"
ERROR = "error"
SYSTEM = "system"
@dataclass
class SessionLog:
session_id: str
level: LogLevel
content: str
timestamp: datetime = field(default_factory=datetime.now)
token_usage: int = 0
execution_time: float = 0.0
metadata: dict = field(default_factory=dict)
class SessionLogger:
def __init__(self, store: RedisSessionStore):
self.store = store
async def log(self, entry: SessionLog):
log_data = {
"level": entry.level.value,
"content": entry.content,
"timestamp": entry.timestamp.isoformat(),
"token_usage": entry.token_usage,
"execution_time": entry.execution_time,
"metadata": entry.metadata,
}
key = f"claude:session:{entry.session_id}:logs"
await self.store.redis.lpush(key, json.dumps(log_data))
await self.store.redis.ltrim(key, 0, 499) # keep at most 500 entries
Central Command System
I implemented a central command system that broadcasts commands to all sessions simultaneously — for example, running git status across every active session in one shot.
class CentralCommand:
def __init__(self, session_manager: SessionManager, logger: SessionLogger):
self.manager = session_manager
self.logger = logger
async def broadcast_command(self, command: str, target: str = "all") -> dict:
sessions = await self.manager.get_active_sessions()
if target != "all":
sessions = [s for s in sessions if s.id in target.split(",")]
results = {}
tasks = [
self._execute_on_session(session.id, command)
for session in sessions
]
responses = await asyncio.gather(*tasks, return_exceptions=True)
for session, response in zip(sessions, responses):
results[session.id] = {
"success": not isinstance(response, Exception),
"result": str(response),
}
return results
async def _execute_on_session(self, session_id: str, command: str):
await self.logger.log(SessionLog(
session_id=session_id,
level=LogLevel.INPUT,
content=command,
))
result = await self.manager.send_prompt(session_id, command)
await self.logger.log(SessionLog(
session_id=session_id,
level=LogLevel.OUTPUT,
content=result,
))
return result
Log Query API
@app.get("/api/sessions/{session_id}/logs")
async def get_session_logs(
session_id: str,
level: LogLevel | None = None,
limit: int = 50,
):
logs = await logger.get_logs(session_id, level=level, limit=limit)
return {"data": logs, "total": len(logs)}
Retrospective
Having a proper logging system made debugging dramatically easier. Whenever a session behaved unexpectedly, digging through the logs was usually enough to find the root cause. The central command system started out as a simple broadcast mechanism, but it later became the foundation for a manager-worker pattern.