/** * @fileoverview Session Manager for coordinating multiple Claude sessions. * * Lifecycle management for Claude CLI sessions: * - Session creation with working directory, concurrent session limits (mutex-guarded) * - Event forwarding from individual sessions to subscribers * - State persistence via StateStore * - Graceful shutdown of all sessions * * Key exports: * - `SessionManager` class — coordinator, extends EventEmitter * - `SessionManagerEvents` interface — typed event map * - `getSessionManager()` — singleton accessor * * Key methods: `createSession(workingDir)`, `getSession(id)`, `getAllSessions()`, * `removeSession(id)`, `stopAll()` * * @dependencies session (Session class), state-store (persistence), types (SessionState) * @consumedby web/server, ralph-loop, respawn-controller * @emits sessionStarted, sessionStopped, sessionError, sessionOutput, sessionCompletion * * @module session-manager */ import { EventEmitter } from 'node:events'; import { Session } from './session.js'; import { getStore } from './state-store.js'; import { SessionState } from './types.js'; /** * Events emitted by SessionManager */ /** * Manages multiple Claude sessions with lifecycle coordination. * * @description * SessionManager acts as a coordinator for multiple Claude CLI sessions: * - Enforces concurrent session limits from config * - Forwards session events to subscribers * - Persists session state to disk * - Handles graceful shutdown * * @extends EventEmitter * @fires SessionManagerEvents.sessionStarted * @fires SessionManagerEvents.sessionStopped * @fires SessionManagerEvents.sessionError * @fires SessionManagerEvents.sessionOutput * @fires SessionManagerEvents.sessionCompletion */ /** Stored event handlers for a session, used for cleanup */ interface SessionHandlers { output: (data: string) => void; error: (data: string) => void; completion: (phrase: string) => void; exit: () => void; taskError: (taskId: string, error: string) => void; } export class SessionManager extends EventEmitter { private sessions: Map = new Map(); private sessionHandlers: Map = new Map(); private store = getStore(); // Mutex for session creation to prevent race conditions private _sessionCreationLock: Promise | null = null; /** * Creates a new SessionManager and loads previous session state. */ constructor() { super(); this.loadFromStore(); } private loadFromStore(): void { const storedSessions = this.store.getSessions(); // Note: We don't restore actual processes, just the state // Dead sessions are marked as stopped for (const [id, state] of Object.entries(storedSessions)) { if (state.status !== 'stopped') { state.status = 'stopped'; state.pid = null; this.store.setSession(id, state); } } } /** * Creates and starts a new Claude session. * Uses mutex to prevent race conditions when multiple requests arrive simultaneously. * * @param workingDir - Working directory for the session * @returns The newly created session * @throws Error if max concurrent sessions limit reached */ async createSession(workingDir: string): Promise { // Wait for any pending session creation to complete (mutex pattern) while (this._sessionCreationLock) { await this._sessionCreationLock; } // Create a new lock promise that others will wait on // Define unlock first to ensure it's always in scope before promise assignment let unlock!: () => void; const lockPromise = new Promise((resolve) => { unlock = resolve; }); this._sessionCreationLock = lockPromise; try { const config = this.store.getConfig(); // Check limit INSIDE the lock to prevent race conditions if (this.sessions.size >= config.maxConcurrentSessions) { throw new Error(`Maximum concurrent sessions (${config.maxConcurrentSessions}) reached`); } const session = new Session({ workingDir }); // Set up event forwarding with stored handlers for cleanup const handlers: SessionHandlers = { output: (data: string) => { this.emit('sessionOutput', session.id, data); this.updateSessionState(session); }, error: (data: string) => { this.emit('sessionError', session.id, data); this.updateSessionState(session); }, completion: (phrase: string) => { this.emit('sessionCompletion', session.id, phrase); }, exit: () => { this.emit('sessionStopped', session.id); this.updateSessionState(session); }, taskError: (taskId: string, error: string) => { this.emit('sessionTaskError', session.id, taskId, error); }, }; session.on('output', handlers.output); session.on('error', handlers.error); session.on('completion', handlers.completion); session.on('exit', handlers.exit); session.on('taskError', handlers.taskError); // Store handlers for later cleanup this.sessionHandlers.set(session.id, handlers); await session.start(); this.sessions.set(session.id, session); this.updateSessionState(session); this.emit('sessionStarted', session); return session; } finally { // Release the lock so other createSession calls can proceed this._sessionCreationLock = null; unlock(); } } /** * Stops a session by ID. * * @param id - Session ID to stop */ async stopSession(id: string): Promise { const session = this.sessions.get(id); if (!session) { // Update store to mark as stopped if it exists there const storedSession = this.store.getSession(id); if (storedSession) { storedSession.status = 'stopped'; storedSession.pid = null; this.store.setSession(id, storedSession); } return; } // Remove event listeners to prevent memory leaks const handlers = this.sessionHandlers.get(id); if (handlers) { session.off('output', handlers.output); session.off('error', handlers.error); session.off('completion', handlers.completion); session.off('exit', handlers.exit); session.off('taskError', handlers.taskError); this.sessionHandlers.delete(id); } await session.stop(); this.sessions.delete(id); this.updateSessionState(session); } /** * Stops all active sessions. * Uses Promise.allSettled to ensure all sessions are stopped even if some fail. */ async stopAllSessions(): Promise { const stopPromises = Array.from(this.sessions.keys()).map((id) => this.stopSession(id)); const results = await Promise.allSettled(stopPromises); // Log any failures but don't throw - best effort cleanup for (const result of results) { if (result.status === 'rejected') { console.error('[SessionManager] Failed to stop session:', result.reason); } } } /** * Gets a session by ID. * @param id - Session ID * @returns The session or undefined if not found */ getSession(id: string): Session | undefined { return this.sessions.get(id); } /** Gets all active sessions. */ getAllSessions(): Session[] { return Array.from(this.sessions.values()); } /** Gets all sessions currently idle (not processing). */ getIdleSessions(): Session[] { return this.getAllSessions().filter((s) => s.isIdle()); } /** Gets all sessions currently busy (processing). */ getBusySessions(): Session[] { return this.getAllSessions().filter((s) => s.isBusy()); } /** Gets the count of active sessions. */ getSessionCount(): number { return this.sessions.size; } /** Checks if a session exists by ID. */ hasSession(id: string): boolean { return this.sessions.has(id); } private updateSessionState(session: Session): void { // envOverrides is intentionally NOT on SessionState (API safety). For disk // persistence we augment the stored object with __envOverrides so reboot // recovery can restore them without leaking through any API serializer. // The key uses the reserved `__` prefix so it is visibly "internal" to any // future reader of state.json. const state = session.toState(); const envOverrides = session.getEnvOverridesForPersist(); const toStore = envOverrides ? { ...state, __envOverrides: envOverrides } : state; this.store.setSession(session.id, toStore as SessionState); } /** Gets all sessions from persistent storage (including stopped). */ getStoredSessions(): Record { return this.store.getSessions(); } /** * Sends input to a session. * * @param sessionId - Session ID to send to * @param input - Input string to send * @throws Error if session not found */ async sendToSession(sessionId: string, input: string): Promise { const session = this.sessions.get(sessionId); if (!session) { throw new Error(`Session ${sessionId} not found`); } await session.sendInput(input); } /** Gets the output buffer for a session. */ getSessionOutput(sessionId: string): string | null { const session = this.sessions.get(sessionId); return session?.getOutput() ?? null; } /** Gets the error buffer for a session. */ getSessionError(sessionId: string): string | null { const session = this.sessions.get(sessionId); return session?.getError() ?? null; } } // Singleton instance let managerInstance: SessionManager | null = null; /** * Gets or creates the singleton SessionManager instance. * @returns The global SessionManager */ export function getSessionManager(): SessionManager { if (!managerInstance) { managerInstance = new SessionManager(); } return managerInstance; }