From 447855ceec14c15ad382b15b8d0a45c81917f6da Mon Sep 17 00:00:00 2001 From: Mikkel Georgsen Date: Wed, 4 Feb 2026 17:32:16 +0000 Subject: [PATCH] feat(01-01): create SessionManager module - 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 --- telegram/__init__.py | 1 + telegram/session_manager.py | 336 ++++++++++++++++++++++++++++++++++++ 2 files changed, 337 insertions(+) create mode 100644 telegram/__init__.py create mode 100644 telegram/session_manager.py diff --git a/telegram/__init__.py b/telegram/__init__.py new file mode 100644 index 0000000..3e1613a --- /dev/null +++ b/telegram/__init__.py @@ -0,0 +1 @@ +"""Telegram bot for Claude Code homelab management.""" diff --git a/telegram/session_manager.py b/telegram/session_manager.py new file mode 100644 index 0000000..328fa3a --- /dev/null +++ b/telegram/session_manager.py @@ -0,0 +1,336 @@ +""" +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// + 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)