Mikkel Georgsen
447855ceec
- SessionManager class with session lifecycle management - Session CRUD: create, list, switch, get, update - Session validation: alphanumeric, hyphens, underscores only - Persona inheritance from library on session creation - Session status tracking: idle, active, suspended - Metadata persistence with JSON on disk - Active session tracking and switching logic
336 lines
10 KiB
Python
336 lines
10 KiB
Python
"""
|
|
Session management for Claude Code Telegram bot.
|
|
|
|
This module provides the SessionManager class that handles lifecycle management
|
|
for isolated Claude Code conversation sessions. Each session is a directory
|
|
containing metadata, persona configuration, and Claude Code's .claude/ data.
|
|
|
|
Sessions enable:
|
|
- Multiple independent Claude Code conversations
|
|
- Persona-based behavior customization
|
|
- Context isolation per conversation thread
|
|
- Session switching without losing state
|
|
|
|
Directory structure:
|
|
~/telegram/sessions/<name>/
|
|
metadata.json # Session state (status, timestamps, PID)
|
|
persona.json # Persona configuration (copied from library)
|
|
.claude/ # Auto-created by Claude Code CLI
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SessionManager:
|
|
"""Manages Claude Code session lifecycle and persona library."""
|
|
|
|
SESSION_NAME_PATTERN = re.compile(r'^[a-zA-Z0-9_-]+$')
|
|
MAX_NAME_LENGTH = 50
|
|
|
|
def __init__(self, base_dir: Optional[Path] = None):
|
|
"""
|
|
Initialize SessionManager.
|
|
|
|
Args:
|
|
base_dir: Base directory for sessions. Defaults to ~/telegram/sessions/
|
|
"""
|
|
if base_dir is None:
|
|
base_dir = Path.home() / "telegram" / "sessions"
|
|
|
|
self.base_dir = Path(base_dir)
|
|
self.personas_dir = Path.home() / "telegram" / "personas"
|
|
self.active_session: Optional[str] = None
|
|
|
|
# Create directories if they don't exist
|
|
self.base_dir.mkdir(parents=True, exist_ok=True)
|
|
self.personas_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Load active session from disk if any
|
|
self._load_active_session()
|
|
|
|
logger.info(f"SessionManager initialized: base_dir={self.base_dir}")
|
|
|
|
def _load_active_session(self) -> None:
|
|
"""Load active session name from existing sessions."""
|
|
try:
|
|
sessions = self.list_sessions()
|
|
for session in sessions:
|
|
if session.get('status') == 'active':
|
|
self.active_session = session['name']
|
|
logger.debug(f"Loaded active session: {self.active_session}")
|
|
return
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load active session: {e}")
|
|
|
|
def _validate_session_name(self, name: str) -> None:
|
|
"""
|
|
Validate session name.
|
|
|
|
Args:
|
|
name: Session name to validate
|
|
|
|
Raises:
|
|
ValueError: If name is invalid
|
|
"""
|
|
if not name:
|
|
raise ValueError("Session name cannot be empty")
|
|
|
|
if len(name) > self.MAX_NAME_LENGTH:
|
|
raise ValueError(
|
|
f"Session name too long (max {self.MAX_NAME_LENGTH} chars): {name}"
|
|
)
|
|
|
|
if not self.SESSION_NAME_PATTERN.match(name):
|
|
raise ValueError(
|
|
f"Invalid session name '{name}': only alphanumeric, hyphens, "
|
|
"and underscores allowed"
|
|
)
|
|
|
|
def _read_metadata(self, name: str) -> dict:
|
|
"""Read session metadata from disk."""
|
|
metadata_path = self.base_dir / name / "metadata.json"
|
|
if not metadata_path.exists():
|
|
raise ValueError(f"Session '{name}' does not exist")
|
|
|
|
with metadata_path.open('r') as f:
|
|
return json.load(f)
|
|
|
|
def _write_metadata(self, name: str, metadata: dict) -> None:
|
|
"""Write session metadata to disk."""
|
|
metadata_path = self.base_dir / name / "metadata.json"
|
|
with metadata_path.open('w') as f:
|
|
json.dump(metadata, f, indent=2)
|
|
logger.debug(f"Wrote metadata for session '{name}'")
|
|
|
|
def create_session(self, name: str, persona: Optional[str] = None) -> Path:
|
|
"""
|
|
Create a new session.
|
|
|
|
Args:
|
|
name: Session name (alphanumeric, hyphens, underscores only)
|
|
persona: Persona name from library (defaults to 'default')
|
|
|
|
Returns:
|
|
Path to created session directory
|
|
|
|
Raises:
|
|
ValueError: If session already exists or name is invalid
|
|
"""
|
|
self._validate_session_name(name)
|
|
|
|
session_dir = self.base_dir / name
|
|
if session_dir.exists():
|
|
raise ValueError(f"Session '{name}' already exists")
|
|
|
|
# Use default persona if not specified
|
|
if persona is None:
|
|
persona = 'default'
|
|
|
|
# Load persona from library
|
|
persona_source = self.personas_dir / f"{persona}.json"
|
|
if not persona_source.exists():
|
|
raise FileNotFoundError(
|
|
f"Persona '{persona}' not found in library: {persona_source}"
|
|
)
|
|
|
|
# Create session directory
|
|
session_dir.mkdir(parents=True, exist_ok=False)
|
|
logger.info(f"Created session directory: {session_dir}")
|
|
|
|
# Copy persona to session
|
|
persona_dest = session_dir / "persona.json"
|
|
persona_dest.write_text(persona_source.read_text())
|
|
logger.debug(f"Copied persona '{persona}' to session")
|
|
|
|
# Create metadata
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
metadata = {
|
|
"name": name,
|
|
"created": now,
|
|
"last_active": now,
|
|
"persona": persona,
|
|
"pid": None,
|
|
"status": "idle"
|
|
}
|
|
self._write_metadata(name, metadata)
|
|
|
|
logger.info(f"Created session '{name}' with persona '{persona}'")
|
|
return session_dir
|
|
|
|
def switch_session(self, name: str) -> Optional[str]:
|
|
"""
|
|
Switch to a different session.
|
|
|
|
Args:
|
|
name: Session name to switch to
|
|
|
|
Returns:
|
|
Previous active session name, or None if no previous session
|
|
|
|
Raises:
|
|
ValueError: If session does not exist
|
|
"""
|
|
self._validate_session_name(name)
|
|
|
|
if not self.session_exists(name):
|
|
raise ValueError(f"Session '{name}' does not exist")
|
|
|
|
# No-op if already active
|
|
if self.active_session == name:
|
|
logger.debug(f"Session '{name}' already active")
|
|
return None
|
|
|
|
previous_session = self.active_session
|
|
|
|
# Mark previous session as suspended
|
|
if previous_session:
|
|
try:
|
|
prev_meta = self._read_metadata(previous_session)
|
|
if prev_meta['status'] == 'active':
|
|
prev_meta['status'] = 'suspended'
|
|
self._write_metadata(previous_session, prev_meta)
|
|
logger.debug(f"Suspended previous session: {previous_session}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to suspend previous session: {e}")
|
|
|
|
# Mark new session as active
|
|
new_meta = self._read_metadata(name)
|
|
new_meta['status'] = 'active'
|
|
new_meta['last_active'] = datetime.now(timezone.utc).isoformat()
|
|
self._write_metadata(name, new_meta)
|
|
|
|
self.active_session = name
|
|
logger.info(f"Switched to session '{name}'")
|
|
|
|
return previous_session
|
|
|
|
def get_session(self, name: str) -> dict:
|
|
"""
|
|
Get session metadata.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
Session metadata dict
|
|
|
|
Raises:
|
|
ValueError: If session does not exist
|
|
"""
|
|
return self._read_metadata(name)
|
|
|
|
def list_sessions(self) -> list[dict]:
|
|
"""
|
|
List all sessions.
|
|
|
|
Returns:
|
|
List of session metadata dicts, sorted by last_active (most recent first)
|
|
"""
|
|
sessions = []
|
|
|
|
if not self.base_dir.exists():
|
|
return sessions
|
|
|
|
for session_dir in self.base_dir.iterdir():
|
|
if not session_dir.is_dir():
|
|
continue
|
|
|
|
metadata_path = session_dir / "metadata.json"
|
|
if not metadata_path.exists():
|
|
logger.warning(f"Session directory missing metadata: {session_dir}")
|
|
continue
|
|
|
|
try:
|
|
with metadata_path.open('r') as f:
|
|
metadata = json.load(f)
|
|
sessions.append(metadata)
|
|
except Exception as e:
|
|
logger.error(f"Failed to read metadata for {session_dir}: {e}")
|
|
continue
|
|
|
|
# Sort by last_active, most recent first
|
|
sessions.sort(
|
|
key=lambda s: s.get('last_active', ''),
|
|
reverse=True
|
|
)
|
|
|
|
return sessions
|
|
|
|
def get_active_session(self) -> Optional[str]:
|
|
"""
|
|
Get active session name.
|
|
|
|
Returns:
|
|
Active session name, or None if no active session
|
|
"""
|
|
return self.active_session
|
|
|
|
def update_session(self, name: str, **kwargs) -> None:
|
|
"""
|
|
Update session metadata fields.
|
|
|
|
Args:
|
|
name: Session name
|
|
**kwargs: Fields to update
|
|
|
|
Raises:
|
|
ValueError: If session does not exist
|
|
"""
|
|
metadata = self._read_metadata(name)
|
|
metadata.update(kwargs)
|
|
self._write_metadata(name, metadata)
|
|
logger.debug(f"Updated session '{name}': {kwargs}")
|
|
|
|
def session_exists(self, name: str) -> bool:
|
|
"""
|
|
Check if session exists.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
True if session exists, False otherwise
|
|
"""
|
|
session_dir = self.base_dir / name
|
|
return session_dir.exists() and (session_dir / "metadata.json").exists()
|
|
|
|
def get_session_dir(self, name: str) -> Path:
|
|
"""
|
|
Get session directory path.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
Path to session directory
|
|
"""
|
|
return self.base_dir / name
|
|
|
|
def load_persona(self, name: str) -> dict:
|
|
"""
|
|
Load persona from library.
|
|
|
|
Args:
|
|
name: Persona name
|
|
|
|
Returns:
|
|
Persona configuration dict
|
|
|
|
Raises:
|
|
FileNotFoundError: If persona does not exist
|
|
"""
|
|
persona_path = self.personas_dir / f"{name}.json"
|
|
if not persona_path.exists():
|
|
raise FileNotFoundError(
|
|
f"Persona '{name}' not found in library: {persona_path}"
|
|
)
|
|
|
|
with persona_path.open('r') as f:
|
|
return json.load(f)
|