Real-Time Session Management For Voice Agents That Don't Crash

// Text chat: stateless, simple
app.post('/chat', async (req, res) => {
  const { message, userId } = req.body;
  const context = await loadContext(userId);
  const response = await agent.respond(message, context);
  await saveContext(userId, response.newContext);
  res.json({ reply: response.text });
});

interface VoiceSession {
  // Identity
  sessionId: string;
  userId: string;
  agentId: string;
  
  // Connection
  connectionId: string;
  transport: 'websocket' | 'webrtc';
  clientInfo: {
    userAgent: string;
    ip: string;
    network: 'wifi' | 'cellular' | 'ethernet';
  };
  
  // State
  status: 'connecting' | 'active' | 'paused' | 'disconnected' | 'ended';
  createdAt: Date;
  lastActivityAt: Date;
  disconnectedAt?: Date;
  
  // Conversation
  conversationState: {
    messages: Message[];
    audioBuffers: AudioBuffer[];
    currentTurn: 'user' | 'agent';
    agentIsSpeaking: boolean;
    userIsInterrupting: boolean;
  };
  
  // Context
  context: {
    userProfile: any;
    sessionGoal: string;
    toolCallHistory: ToolCall[];
    resolvedEntities: Record<string, any>;
  };
  
  // Recovery
  checkpoint: {
    snapshotId: string;
    timestamp: Date;
    recoverable: boolean;
  };
  
  // Metrics
  metrics: {
    totalDuration: number;
    audioBytesSent: number;
    audioBytesReceived: number;
    toolCallCount: number;
    errorCount: number;
    disconnectCount: number;
  };
}

import { EventEmitter } from 'events';
import { Redis } from 'ioredis';
import { RealtimeClient } from '@openai/realtime-api';

class VoiceSessionManager extends EventEmitter {
  private sessions: Map<string, VoiceSession> = new Map();
  private redis: Redis;
  private snapshotInterval: number = 5000; // Snapshot every 5s
  
  constructor() {
    super();
    this.redis = new Redis(process.env.REDIS_URL);
    this.startHealthMonitor();
  }
  
  async createSession(userId: string, agentConfig: any): Promise<VoiceSession> {
    const sessionId = generateSessionId();
    
    const session: VoiceSession = {
      sessionId,
      userId,
      agentId: agentConfig.agentId,
      connectionId: null,
      transport: 'websocket',
      clientInfo: null,
      status: 'connecting',
      createdAt: new Date(),
      lastActivityAt: new Date(),
      conversationState: {
        messages: [],
        audioBuffers: [],
        currentTurn: 'user',
        agentIsSpeaking: false,
        userIsInterrupting: false
      },
      context: {
        userProfile: await this.loadUserProfile(userId),
        sessionGoal: agentConfig.goal,
        toolCallHistory: [],
        resolvedEntities: {}
      },
      checkpoint: {
        snapshotId: null,
        timestamp: new Date(),
        recoverable: false
      },
      metrics: {
        totalDuration: 0,
        audioBytesSent: 0,
        audioBytesReceived: 0,
        toolCallCount: 0,
        errorCount: 0,
        disconnectCount: 0
      }
    };
    
    // Store in memory and Redis
    this.sessions.set(sessionId, session);
    await this.persistSession(session);
    
    // Start auto-snapshotting
    this.startSnapshotting(sessionId);
    
    this.emit('session:created', session);
    return session;
  }
  
  async connectClient(sessionId: string, connectionId: string, clientInfo: any) {
    const session = this.sessions.get(sessionId);
    if (!session) {
      throw new Error('Session not found');
    }
    
    session.connectionId = connectionId;
    session.clientInfo = clientInfo;
    session.status = 'active';
    session.lastActivityAt = new Date();
    
    await this.persistSession(session);
    this.emit('session:connected', session);
  }
  
  async handleDisconnect(sessionId: string, reason: string) {
    const session = this.sessions.get(sessionId);
    if (!session) return;
    
    session.status = 'disconnected';
    session.disconnectedAt = new Date();
    session.metrics.disconnectCount++;
    
    // Create recovery checkpoint
    await this.createCheckpoint(session);
    
    // Keep session alive for reconnection window
    setTimeout(() => {
      this.checkSessionRecovery(sessionId);
    }, 30000); // 30s reconnection window
    
    this.emit('session:disconnected', { session, reason });
  }
  
  async reconnectSession(sessionId: string, newConnectionId: string): Promise<VoiceSession> {
    const session = this.sessions.get(sessionId);
    
    if (!session) {
      // Try to recover from Redis
      const recovered = await this.recoverSession(sessionId);
      if (!recovered) {
        throw new Error('Session expired or not found');
      }
      return recovered;
    }
    
    if (session.status !== 'disconnected') {
      throw new Error('Session is not in disconnected state');
    }
    
    // Restore from last checkpoint
    await this.restoreFromCheckpoint(session);
    
    session.connectionId = newConnectionId;
    session.status = 'active';
    session.disconnectedAt = null;
    session.lastActivityAt = new Date();
    
    await this.persistSession(session);
    this.emit('session:reconnected', session);
    
    return session;
  }
  
  private async createCheckpoint(session: VoiceSession) {
    const snapshotId = `snapshot:${session.sessionId}:${Date.now()}`;
    
    const checkpoint = {
      snapshotId,
      timestamp: new Date(),
      conversationState: JSON.parse(JSON.stringify(session.conversationState)),
      context: JSON.parse(JSON.stringify(session.context)),
      metrics: { ...session.metrics }
    };
    
    // Store in Redis with 5 min TTL
    await this.redis.setex(
      snapshotId,
      300,
      JSON.stringify(checkpoint)
    );
    
    session.checkpoint = {
      snapshotId,
      timestamp: new Date(),
      recoverable: true
    };
  }
  
  private async restoreFromCheckpoint(session: VoiceSession) {
    if (!session.checkpoint.recoverable) {
      throw new Error('Session not recoverable');
    }
    
    const checkpointData = await this.redis.get(session.checkpoint.snapshotId);
    if (!checkpointData) {
      throw new Error('Checkpoint expired');
    }
    
    const checkpoint = JSON.parse(checkpointData);
    session.conversationState = checkpoint.conversationState;
    session.context = checkpoint.context;
    session.metrics = checkpoint.metrics;
  }
  
  private startSnapshotting(sessionId: string) {
    const intervalId = setInterval(async () => {
      const session = this.sessions.get(sessionId);
      if (!session || session.status === 'ended') {
        clearInterval(intervalId);
        return;
      }
      
      if (session.status === 'active') {
        await this.createCheckpoint(session);
      }
    }, this.snapshotInterval);
  }
  
  private async persistSession(session: VoiceSession) {
    await this.redis.setex(
      `session:${session.sessionId}`,
      3600, // 1 hour TTL
      JSON.stringify(session)
    );
  }
  
  async updateActivity(sessionId: string) {
    const session = this.sessions.get(sessionId);
    if (session) {
      session.lastActivityAt = new Date();
      await this.persistSession(session);
    }
  }
  
  private startHealthMonitor() {
    setInterval(() => {
      const now = Date.now();
      for (const [sessionId, session] of this.sessions) {
        const inactiveMs = now - session.lastActivityAt.getTime();
        
        // Mark stale after 60s of inactivity
        if (inactiveMs > 60000 && session.status === 'active') {
          this.handleDisconnect(sessionId, 'inactivity_timeout');
        }
        
        // Clean up old disconnected sessions
        if (session.status === 'disconnected' && inactiveMs > 300000) {
          this.endSession(sessionId);
        }
      }
    }, 10000); // Check every 10s
  }
  
  async endSession(sessionId: string) {
    const session = this.sessions.get(sessionId);
    if (!session) return;
    
    session.status = 'ended';
    session.metrics.totalDuration = Date.now() - session.createdAt.getTime();
    
    // Final persist
    await this.persistSession(session);
    
    // Archive to long-term storage
    await this.archiveSession(session);
    
    // Clean up
    this.sessions.delete(sessionId);
    this.emit('session:ended', session);
  }
  
  private async archiveSession(session: VoiceSession) {
    // Store in database for analytics
    await this.redis.lpush('archived_sessions', JSON.stringify(session));
  }
}

class ResilientVoiceClient {
  constructor(config) {
    this.config = config;
    this.sessionId = null;
    this.connectionId = null;
    this.ws = null;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
    this.localBuffer = {
      pendingAudio: [],
      pendingMessages: []
    };
  }
  
  async connect() {
    try {
      // Create or resume session
      const response = await fetch('/api/voice/session', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          userId: this.config.userId,
          sessionId: this.sessionId, // Null for new, set for resume
          agentConfig: this.config.agent
        })
      });
      
      const { sessionId, wsUrl, checkpointData } = await response.json();
      
      this.sessionId = sessionId;
      this.connectionId = generateConnectionId();
      
      // Restore from checkpoint if reconnecting
      if (checkpointData) {
        this.restoreFromCheckpoint(checkpointData);
      }
      
      // Establish WebSocket
      this.ws = new WebSocket(wsUrl);
      this.setupWebSocket();
      
      this.reconnectAttempts = 0;
      
    } catch (error) {
      console.error('Connection failed:', error);
      this.handleConnectionFailure();
    }
  }
  
  setupWebSocket() {
    this.ws.onopen = () => {
      console.log('WebSocket connected');
      
      // Send connection handshake
      this.ws.send(JSON.stringify({
        type: 'handshake',
        sessionId: this.sessionId,
        connectionId: this.connectionId,
        clientInfo: {
          userAgent: navigator.userAgent,
          network: this.detectNetworkType()
        }
      }));
      
      // Flush buffered data
      this.flushLocalBuffer();
      
      this.emit('connected');
    };
    
    this.ws.onmessage = (event) => {
      this.handleMessage(event.data);
    };
    
    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
      this.emit('error', error);
    };
    
    this.ws.onclose = (event) => {
      console.log('WebSocket closed:', event.code, event.reason);
      this.handleDisconnect(event);
    };
    
    // Heartbeat
    this.startHeartbeat();
  }
  
  handleDisconnect(event) {
    this.emit('disconnected', { code: event.code, reason: event.reason });
    
    // Attempt reconnection if not intentional close
    if (event.code !== 1000 && this.reconnectAttempts < this.maxReconnectAttempts) {
      this.attemptReconnect();
    } else {
      this.emit('connection_failed');
    }
  }
  
  async attemptReconnect() {
    this.reconnectAttempts++;
    
    // Exponential backoff
    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 10000);
    
    console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})`);
    
    this.emit('reconnecting', { attempt: this.reconnectAttempts, delay });
    
    await new Promise(resolve => setTimeout(resolve, delay));
    
    await this.connect();
  }
  
  sendAudio(audioData) {
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(audioData);
    } else {
      // Buffer for later
      this.localBuffer.pendingAudio.push({
        data: audioData,
        timestamp: Date.now()
      });
    }
  }
  
  flushLocalBuffer() {
    // Send buffered audio
    for (const item of this.localBuffer.pendingAudio) {
      // Only send recent audio (< 5s old)
      if (Date.now() - item.timestamp < 5000) {
        this.ws.send(item.data);
      }
    }
    this.localBuffer.pendingAudio = [];
    
    // Send buffered messages
    for (const msg of this.localBuffer.pendingMessages) {
      this.ws.send(JSON.stringify(msg));
    }
    this.localBuffer.pendingMessages = [];
  }
  
  startHeartbeat() {
    this.heartbeatInterval = setInterval(() => {
      if (this.ws && this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify({ type: 'ping' }));
      }
    }, 15000); // Ping every 15s
  }
  
  detectNetworkType() {
    if (navigator.connection) {
      return navigator.connection.effectiveType;
    }
    return 'unknown';
  }
  
  restoreFromCheckpoint(checkpoint) {
    // Restore UI state
    this.emit('restore', checkpoint);
  }
  
  disconnect() {
    if (this.heartbeatInterval) {
      clearInterval(this.heartbeatInterval);
    }
    if (this.ws) {
      this.ws.close(1000, 'Client disconnect');
    }
  }
}

// User's WiFi drops for 2 seconds during conversation

// What happens:
// 1. Client detects disconnect
client.on('disconnected', () => {
  showUI('Reconnecting...');
  // Don't stop audio recording - buffer locally
});

// 2. Client reconnects automatically
client.on('reconnected', async () => {
  showUI('Connected');
  
  // 3. Server restores from checkpoint (5s ago)
  const restored = await sessionManager.reconnectSession(sessionId, newConnectionId);
  
  // 4. Client sends buffered audio
  await client.flushLocalBuffer();
  
  // Result: Seamless recovery, user barely notices
});

// User's phone goes into tunnel for 45 seconds

// What happens:
// 1. Session marked 'disconnected', checkpoint created
await sessionManager.handleDisconnect(sessionId, 'extended_timeout');

// 2. Session kept alive for 30s reconnection window
// 3. Client reconnects after 45s
try {
  const recovered = await sessionManager.reconnectSession(sessionId, newConnectionId);
  // Session expired, checkpoint gone
} catch (error) {
  // 4. Start fresh session with context summary
  const newSession = await sessionManager.createSession(userId, {
    ...agentConfig,
    resumeContext: summarizeOldSession(sessionId)
  });
  
  // Agent says: "Welcome back. We were discussing your order #12345. Let's continue."
}

// Your server deploys new code mid-conversation

// Session state lives in Redis, not server memory
// Gateway routes to new server instance
// New server rehydrates session from Redis

async function recoverSession(sessionId: string): Promise<VoiceSession> {
  // Load from Redis
  const sessionData = await redis.get(`session:${sessionId}`);
  if (!sessionData) {
    throw new Error('Session not found');
  }
  
  const session = JSON.parse(sessionData);
  
  // Restore checkpoint
  if (session.checkpoint.recoverable) {
    await this.restoreFromCheckpoint(session);
  }
  
  // Rebuild agent state
  const agent = await this.initializeAgent(session.agentId);
  agent.restoreContext(session.context);
  
  // Resume
  this.sessions.set(sessionId, session);
  return session;
}

class SmartCheckpointer {
  private dirtyFlags: Map<string, boolean> = new Map();
  
  markDirty(sessionId: string) {
    this.dirtyFlags.set(sessionId, true);
  }
  
  async flushIfDirty(sessionId: string) {
    if (this.dirtyFlags.get(sessionId)) {
      await this.createCheckpoint(sessionId);
      this.dirtyFlags.set(sessionId, false);
    }
  }
  
  startBatchFlusher() {
    setInterval(async () => {
      for (const [sessionId, isDirty] of this.dirtyFlags) {
        if (isDirty) {
          await this.flushIfDirty(sessionId);
        }
      }
    }, 5000); // Batch flush every 5s
  }
}

async createDifferentialCheckpoint(session: VoiceSession) {
  const lastCheckpoint = await this.getLastCheckpoint(session.sessionId);
  
  const diff = {
    newMessages: session.conversationState.messages.slice(lastCheckpoint.messageCount),
    contextUpdates: deepDiff(lastCheckpoint.context, session.context),
    metricDeltas: {
      toolCalls: session.metrics.toolCallCount - lastCheckpoint.metrics.toolCallCount
    }
  };
  
  // Store diff + pointer to base
  await redis.setex(
    `checkpoint:${session.sessionId}:${Date.now()}`,
    300,
    JSON.stringify({ base: lastCheckpoint.id, diff })
  );
}

class SessionHealthMonitor {
  async getMetrics(sessionId: string) {
    const session = await sessionManager.getSession(sessionId);
    
    return {
      // Connection health
      uptime: Date.now() - session.createdAt.getTime(),
      disconnectRate: session.metrics.disconnectCount / (session.metrics.totalDuration / 60000),
      currentLatency: await this.measureLatency(sessionId),
      
      // Conversation health  
      turnCount: session.conversationState.messages.length,
      avgTurnDuration: this.calculateAvgTurnDuration(session),
      interruptionRate: this.calculateInterruptionRate(session),
      
      // Resource usage
      memoryFootprint: this.estimateMemoryUsage(session),
      audioBufferSize: session.conversationState.audioBuffers.reduce((sum, buf) => sum + buf.length, 0),
      
      // Recovery readiness
      checkpointAge: Date.now() - session.checkpoint.timestamp.getTime(),
      recoverable: session.checkpoint.recoverable
    };
  }
  
  async alertOnUnhealthy(sessionId: string) {
    const metrics = await this.getMetrics(sessionId);
    
    if (metrics.disconnectRate > 0.5) {
      this.alert('High disconnect rate', { sessionId, rate: metrics.disconnectRate });
    }
    
    if (metrics.currentLatency > 2000) {
      this.alert('High latency', { sessionId, latency: metrics.currentLatency });
    }
    
    if (metrics.checkpointAge > 30000) {
      this.alert('Stale checkpoint', { sessionId, age: metrics.checkpointAge });
    }
  }
}