"""Tmux session lifecycle management using libtmux. This module provides the core functionality for creating, listing, and destroying tmux sessions. It is the primary interface to tmux that all other components use. """ from __future__ import annotations import re import uuid from dataclasses import dataclass from datetime import datetime from typing import List, Optional import libtmux import libtmux.exc from gsd_openclaw.state_manager import StateManager, SessionState from gsd_openclaw.workspace import WorkspaceManager class TmuxSessionError(Exception): """Errors related to tmux session management.""" pass @dataclass class TmuxSession: """Represents a managed tmux session. Attributes: session_id: Unique identifier for this session. project_name: Name of the project this session belongs to. tmux_session_name: Name of the tmux session (gsd-{project} format). tmux_session: Runtime reference to the actual tmux session (optional). """ session_id: str project_name: str tmux_session_name: str tmux_session: Optional[libtmux.Session] = None @property def exists(self) -> bool: """Check if the tmux session still exists. Returns: True if the session exists in tmux, False otherwise. """ if self.tmux_session is None: return False try: # Check if session still exists in tmux by accessing its ID _ = self.tmux_session.session_id return True except libtmux.exc.LibTmuxException: return False @dataclass class SessionInfo: """Session info for display. Attributes: tmux_name: Name of the tmux session. project_name: Name of the project. session_id: Unique session identifier. created_at: ISO timestamp when session was created. last_active_at: ISO timestamp of last activity. command_count: Number of commands executed. display_name: User-friendly display name. is_orphan: True if tmux session exists but no state file. """ tmux_name: str project_name: str session_id: str created_at: str last_active_at: str command_count: int display_name: Optional[str] is_orphan: bool = False @dataclass class CleanupResult: """Result of orphan cleanup. Attributes: cleaned_sessions: List of tmux session names that were cleaned up. errors: List of error messages for sessions that failed to clean. """ cleaned_sessions: List[str] errors: List[str] class TmuxSessionManager: """Manages tmux session lifecycle using libtmux. This class provides methods for creating, listing, and destroying tmux sessions. It handles session naming with collision detection and integrates with StateManager for persistence and WorkspaceManager for project isolation. Attributes: SESSION_PREFIX: Prefix for all gsd-managed sessions ("gsd-"). server: libtmux.Server instance for tmux operations. """ SESSION_PREFIX = "gsd-" def __init__(self, server: Optional[libtmux.Server] = None): """Initialize TmuxSessionManager. Args: server: Optional libtmux.Server instance. If not provided, a new server connection will be created. """ self.server = server or libtmux.Server() def _generate_session_name(self, project_name: str) -> str: """Generate tmux session name from project name. Args: project_name: Name of the project. Returns: Session name in format "gsd-{project_name}". """ return f"{self.SESSION_PREFIX}{project_name}" def _resolve_collision(self, base_name: str) -> str: """Add timestamp suffix if session name already exists. Args: base_name: The base session name to check. Returns: The original name if no collision, or name with timestamp suffix in format "gsd-{project}-YYYYMMDD-HHMMSS". """ existing = self.server.sessions.filter(session_name=base_name) if not existing: return base_name suffix = datetime.now().strftime("-%Y%m%d-%H%M%S") return f"{base_name}{suffix}" def _extract_project_name(self, tmux_name: str) -> str: """Extract project name from tmux session name. Handles both standard format (gsd-{project}) and collision format (gsd-{project}-YYYYMMDD-HHMMSS). Args: tmux_name: The tmux session name. Returns: Extracted project name. """ # Remove prefix name = tmux_name[len(self.SESSION_PREFIX):] # Remove timestamp suffix if present (format: -YYYYMMDD-HHMMSS) match = re.match(r'^(.+?)(-\d{8}-\d{6})?$', name) return match.group(1) if match else name def list_sessions(self) -> List[libtmux.Session]: """List all gsd-managed sessions. Returns: List of libtmux.Session objects with names starting with "gsd-". """ return list(self.server.sessions.filter( session_name__startswith=self.SESSION_PREFIX )) def create_session( self, project_name: str, workspace_manager: Optional[WorkspaceManager] = None, display_name: Optional[str] = None ) -> TmuxSession: """Create a new tmux session for a project. This method: 1. Ensures the workspace directory exists 2. Generates a unique session name (with collision handling) 3. Creates the tmux session (detached) 4. Persists session state 5. Returns a TmuxSession object Args: project_name: Name of the project. workspace_manager: Optional WorkspaceManager instance. If not provided, a new one will be created for the project. display_name: Optional user-friendly display name for the session. Returns: TmuxSession object with all metadata populated. """ # Ensure workspace exists ws_manager = workspace_manager or WorkspaceManager(project_name) ws_manager.create() # Generate unique session name base_name = self._generate_session_name(project_name) tmux_name = self._resolve_collision(base_name) # Create tmux session (detached) tmux_session = self.server.new_session( session_name=tmux_name, attach=False, window_name='main' ) # Create session state session_id = str(uuid.uuid4())[:8] now = datetime.now().isoformat() state = SessionState( session_id=session_id, project_name=project_name, tmux_session_name=tmux_name, tmux_pane_id=tmux_session.active_pane.pane_id, created_at=now, last_active_at=now, display_name=display_name ) # Persist state state_manager = StateManager(project_name) state_manager.save(state) return TmuxSession( session_id=session_id, project_name=project_name, tmux_session_name=tmux_name, tmux_session=tmux_session ) def destroy_session( self, project_name: str, cleanup_workspace: bool = False ) -> bool: """Destroy a tmux session and optionally cleanup workspace. This method: 1. Loads session state to find tmux session name 2. Kills the tmux session 3. Removes state files 4. Optionally removes the entire workspace Args: project_name: Name of the project. cleanup_workspace: If True, also remove the workspace directory. Returns: True if session was destroyed, False if no session existed. """ state_manager = StateManager(project_name) state = state_manager.load() if state is None: return False # No session to destroy # Kill tmux session try: tmux_session = self.server.sessions.get( session_name=state.tmux_session_name ) if tmux_session: tmux_session.kill() except libtmux.exc.LibTmuxException: pass # Session already gone # Remove state file if state_manager.state_path.exists(): state_manager.state_path.unlink() backup_path = state_manager.state_path.with_suffix('.json.bak') if backup_path.exists(): backup_path.unlink() # Optionally cleanup workspace if cleanup_workspace: workspace_manager = WorkspaceManager(project_name) workspace_manager.destroy() return True def list_gsd_sessions(self) -> List[SessionInfo]: """List all gsd sessions with project associations. This method combines tmux session info with persisted state to provide a complete view of all gsd sessions. Returns: List of SessionInfo objects for all gsd-* sessions. """ sessions_info = [] # Get all gsd-* tmux sessions tmux_sessions = self.list_sessions() for tmux_session in tmux_sessions: tmux_name = tmux_session.name # Extract project name from tmux session name project_name = self._extract_project_name(tmux_name) # Try to load state state_manager = StateManager(project_name) state = state_manager.load() if state and state.tmux_session_name == tmux_name: # Matched session sessions_info.append(SessionInfo( tmux_name=tmux_name, project_name=project_name, session_id=state.session_id, created_at=state.created_at, last_active_at=state.last_active_at, command_count=state.command_count, display_name=state.display_name, is_orphan=False )) else: # Orphan session (no matching state) sessions_info.append(SessionInfo( tmux_name=tmux_name, project_name=project_name, session_id="?", created_at="?", last_active_at="?", command_count=0, display_name=None, is_orphan=True )) return sessions_info def get_session(self, project_name: str) -> Optional[TmuxSession]: """Get existing session without creating new one. Args: project_name: Name of the project. Returns: TmuxSession if state exists, None otherwise. """ state_manager = StateManager(project_name) state = state_manager.load() if state is None: return None # Find tmux session tmux_session = None try: tmux_session = self.server.sessions.get( session_name=state.tmux_session_name ) except libtmux.exc.LibTmuxException: pass return TmuxSession( session_id=state.session_id, project_name=project_name, tmux_session_name=state.tmux_session_name, tmux_session=tmux_session ) def resume_session(self, project_name: str) -> TmuxSession: """Resume an existing session, recreating tmux if needed. Args: project_name: Name of the project. Returns: TmuxSession with all metadata populated. Raises: TmuxSessionError: If no session found, unsupported state version, or workspace missing. """ state_manager = StateManager(project_name) state = state_manager.load() if state is None: raise TmuxSessionError(f"No session found for project: {project_name}") # Validate state if state.schema_version != "1.0": raise TmuxSessionError( f"Unsupported state version: {state.schema_version}" ) # Check if tmux session exists tmux_session = None try: tmux_session = self.server.sessions.get( session_name=state.tmux_session_name ) except libtmux.exc.LibTmuxException: pass # Recreate tmux session if missing if tmux_session is None: workspace_manager = WorkspaceManager(project_name) if not workspace_manager.exists(): raise TmuxSessionError( f"Workspace missing for project: {project_name}" ) # Create new tmux session with same name tmux_session = self.server.new_session( session_name=state.tmux_session_name, attach=False, window_name='main', start_directory=str(workspace_manager.get_workspace().root) ) # Update pane ID in state state.tmux_pane_id = tmux_session.active_pane.pane_id state.last_active_at = datetime.now().isoformat() state_manager.save(state) return TmuxSession( session_id=state.session_id, project_name=project_name, tmux_session_name=state.tmux_session_name, tmux_session=tmux_session ) def cleanup_orphans(self) -> CleanupResult: """Remove tmux sessions without corresponding state files. This method identifies orphan tmux sessions (gsd-* sessions without valid state files) and kills them. Returns: CleanupResult with lists of cleaned sessions and errors. """ cleaned = [] errors = [] # Get all gsd-* tmux sessions tmux_sessions = self.list_sessions() for tmux_session in tmux_sessions: tmux_name = tmux_session.name project_name = self._extract_project_name(tmux_name) # Check for state file state_manager = StateManager(project_name) state = state_manager.load() should_clean = False if state is None: # No state file - orphan should_clean = True elif state.tmux_session_name != tmux_name: # State exists but references different tmux session # This could be a collision case, treat as orphan should_clean = True if should_clean: try: tmux_session.kill() cleaned.append(tmux_name) except libtmux.exc.LibTmuxException as e: errors.append(f"Failed to kill {tmux_name}: {e}") return CleanupResult(cleaned_sessions=cleaned, errors=errors)