---
title: Connection Manager Algorithm
summary: WebSocket multiplexing, heartbeat/keepalive mechanics, and graceful reconnection strategies for reliable client connectivity
---

# Connection Manager Algorithm

## Overview

SophiaClaw gateway manages persistent WebSocket connections to clients, providing:

1. **Connection multiplexing** - Multiple logical sessions over single WebSocket
2. **Heartbeat/keepalive** - Detect stale connections and maintain NAT mappings
3. **Graceful reconnection** - Exponential backoff with connection state recovery

**Locations:**

- `src/gateway/server/ws-connection.ts` - Server-side connection handling
- `src/web/reconnect.ts` - Client-side reconnection policy
- `src/web/session.ts` - WebSocket session management

---

## WebSocket Multiplexing

### Architecture

```
┌─────────────────────────────────────────────────────────┐
│  SophiaClaw Gateway                                     │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │  WebSocket Server (ws://:37521)                 │   │
│  │                                                  │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐      │   │
│  │  │ Client 1 │  │ Client 2 │  │ Client 3 │      │   │
│  │  │ (mobile) │  │ (desktop)│  │ (browser)│      │   │
│  │  └────┬─────┘  └────┬─────┘  └────┬─────┘      │   │
│  │       │             │             │             │   │
│  │       └─────────────┴─────────────┘             │   │
│  │                     │                           │   │
│  │            ┌────────▼────────┐                  │   │
│  │            │ Connection Pool │                  │   │
│  │            │  Set<GatewayWs> │                  │   │
│  │            └────────┬────────┘                  │   │
│  │                     │                           │   │
│  │            ┌────────▼────────┐                  │   │
│  │            │ Message Router  │                  │   │
│  │            └─────────────────┘                  │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘
```

### Connection Lifecycle

```typescript
// Server-side connection handling
wss.on("connection", (socket, upgradeReq) => {
  let client: GatewayWsClient | null = null;
  let closed = false;
  const openedAt = Date.now();
  const connId = randomUUID();

  // Extract client metadata from handshake
  const remoteAddr = socket._socket?.remoteAddress;
  const requestHost = upgradeReq.headers.host;
  const requestOrigin = upgradeReq.headers.origin;
  const requestUserAgent = upgradeReq.headers["user-agent"];
  const forwardedFor = upgradeReq.headers["x-forwarded-for"];

  // Send connect challenge (authentication handshake)
  const connectNonce = randomUUID();
  send({
    type: "event",
    event: "connect.challenge",
    payload: { nonce: connectNonce, ts: Date.now() },
  });

  // Set handshake timeout
  const handshakeTimer = setTimeout(() => {
    if (handshakeState === "pending") {
      close(4001, "Handshake timeout");
    }
  }, getHandshakeTimeoutMs());

  // Track connection in global pool
  clients.add(client);
});
```

### Connection Metadata

```typescript
interface GatewayWsClient {
  socket: WebSocket;
  connId: string; // Unique connection identifier
  remoteAddr?: string; // Client IP (after proxy resolution)
  openedAt: number; // Connection timestamp
  lastFrameType?: string; // Last message type for debugging
  lastFrameMethod?: string; // Last RPC method called
  lastFrameId?: string; // Last request ID
  authenticated?: boolean; // Auth completed
  sessionKey?: string; // Associated session
}
```

### Broadcast Pattern

```typescript
function broadcast(
  event: string,
  payload: unknown,
  opts?: {
    dropIfSlow?: boolean;
    stateVersion?: { presence?: number; health?: number };
  },
): void {
  for (const client of clients) {
    if (opts?.dropIfSlow && client.sendQueue?.length > MAX_QUEUE_SIZE) {
      logWsControl.warn(`dropping broadcast to slow client ${client.connId}`);
      continue;
    }

    try {
      client.socket.send(
        JSON.stringify({
          type: "event",
          event,
          payload,
          stateVersion: opts?.stateVersion,
        }),
      );
    } catch (err) {
      logWsControl.warn(`broadcast failed to ${client.connId}: ${err}`);
      // Client will be removed on next error/close event
    }
  }
}
```

---

## Heartbeat/Keepalive Algorithm

### Purpose

Heartbeats serve multiple purposes:

1. **Connection liveness detection** - Detect stale/broken connections
2. **NAT mapping maintenance** - Keep NAT/firewall ports open
3. **Load balancer health** - Signal to LB that connection is active
4. **Session continuity** - Trigger reconnection before session expires

### Protocol Design

```
┌─────────────────────────────────────────────────────────┐
│  Heartbeat Exchange                                     │
│                                                         │
│  Client                          Gateway                │
│     │                               │                   │
│     │────── ping (seq: N) ─────────►│                   │
│     │                               │                   │
│     │◄────── pong (seq: N) ─────────│                   │
│     │                               │                   │
│     │  [idle for heartbeatSeconds] │                   │
│     │                               │                   │
│     │────── ping (seq: N+1) ───────►│                   │
│     │                               │                   │
│     │◄────── pong (seq: N+1) ───────│                   │
│     │                               │                   │
└─────────────────────────────────────────────────────────┘
```

### Configuration

```typescript
// Default heartbeat interval
const DEFAULT_HEARTBEAT_SECONDS = 60;

// Resolved from config
function resolveHeartbeatSeconds(cfg: SophiaClawConfig, overrideSeconds?: number): number {
  const candidate = overrideSeconds ?? cfg.web?.heartbeatSeconds;
  if (typeof candidate === "number" && candidate > 0) {
    return candidate;
  }
  return DEFAULT_HEARTBEAT_SECONDS;
}
```

### Server-Side Implementation

```typescript
// Connection handler tracks last activity
let lastActivityAt = Date.now();

socket.on("message", (data) => {
  lastActivityAt = Date.now();
  // Process message...
});

// Periodic liveness check
const livenessTimer = setInterval(() => {
  const idleMs = Date.now() - lastActivityAt;
  const timeoutMs = heartbeatSeconds * 1000 * 2; // Allow 2x grace period

  if (idleMs > timeoutMs) {
    logWsControl.warn(`connection ${connId} stale, closing`);
    close(4002, "Connection stale");
  }
}, heartbeatSeconds * 1000);
```

### Client-Side Implementation

```typescript
// In web/reconnect.ts
export const DEFAULT_RECONNECT_POLICY: ReconnectPolicy = {
  initialMs: 2_000,
  maxMs: 30_000,
  factor: 1.8,
  jitter: 0.25,
  maxAttempts: 12,
};

class WebSocketClient {
  private heartbeatTimer: NodeJS.Timeout | null = null;
  private lastPongAt: number = 0;

  connect() {
    this.socket = new WebSocket(url);

    this.socket.addEventListener("open", () => {
      this.startHeartbeat();
    });

    this.socket.addEventListener("message", (event) => {
      const message = JSON.parse(event.data);
      if (message.type === "pong") {
        this.lastPongAt = Date.now();
      }
    });
  }

  private startHeartbeat() {
    this.heartbeatTimer = setInterval(() => {
      const idleMs = Date.now() - this.lastPongAt;
      const timeoutMs = this.heartbeatSeconds * 1000 * 2;

      if (idleMs > timeoutMs) {
        console.warn("Heartbeat timeout, reconnecting...");
        this.reconnect();
      } else {
        this.send({ type: "ping", seq: this.pingSeq++ });
      }
    }, this.heartbeatSeconds * 1000);
  }

  private stopHeartbeat() {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
  }
}
```

### NAT Traversal

```
┌─────────────────────────────────────────────────────────┐
│  NAT Keepalive Mechanism                                │
│                                                         │
│  Client (NAT)                    Public Internet        │
│  192.168.1.100                                            │
│       │                                                   │
│       │  [NAT table entry created]                       │
│       │  Internal: 192.168.1.100:54321                   │
│       │  External: 203.0.113.5:12345                     │
│       │                                                   │
│       │────── WebSocket ────────────────────────────►    │
│       │      (regular traffic keeps entry alive)         │
│       │                                                   │
│       │  [Without heartbeat: NAT entry expires]          │
│       │  External: 203.0.113.5:12345 → (removed)         │
│       │                                                   │
│       │  [With heartbeat: entry stays active]            │
│       │◄────── Pong ────────────────────────────────     │
│       │                                                   │
└─────────────────────────────────────────────────────────┘
```

---

## Reconnection Algorithm

### Exponential Backoff Policy

```typescript
// In web/reconnect.ts
export interface ReconnectPolicy {
  initialMs: number; // Initial delay: 2000ms
  maxMs: number; // Maximum delay: 30000ms
  factor: number; // Multiplier: 1.8
  jitter: number; // Randomization: 0.25 (±25%)
  maxAttempts: number; // Max retries: 12
}

export function computeBackoff(
  attempt: number,
  policy: ReconnectPolicy,
): { delayMs: number; jitterMs: number } {
  // Exponential backoff: initialMs * (factor ^ attempt)
  const baseDelay = policy.initialMs * Math.pow(policy.factor, attempt);

  // Cap at maximum
  const cappedDelay = Math.min(baseDelay, policy.maxMs);

  // Add jitter to prevent thundering herd
  const jitterRange = cappedDelay * policy.jitter;
  const jitter = (Math.random() - 0.5) * 2 * jitterRange;

  return {
    delayMs: Math.max(0, cappedDelay + jitter),
    jitterMs: jitterRange,
  };
}
```

### Backoff Timeline Example

```
┌─────────────────────────────────────────────────────────┐
│  Exponential Backoff Timeline (DEFAULT_RECONNECT_POLICY)│
│                                                         │
│  Attempt 0: 2000ms  ±500ms  (1.5s - 2.5s)              │
│  Attempt 1: 3600ms  ±900ms  (2.7s - 4.5s)              │
│  Attempt 2: 6480ms  ±1620ms (4.9s - 8.1s)              │
│  Attempt 3: 11664ms ±2916ms (8.7s - 14.6s)             │
│  Attempt 4: 20995ms ±5249ms (15.7s - 26.2s)            │
│  Attempt 5: 30000ms ±7500ms (22.5s - 37.5s) [capped]   │
│  Attempt 6+: 30000ms ±7500ms (capped at maxMs)         │
│                                                         │
│  After 12 attempts (~6 minutes total), give up          │
└─────────────────────────────────────────────────────────┘
```

### Reconnection State Machine

```typescript
enum ReconnectState {
  Connected = "connected",
  Disconnecting = "disconnecting",
  Disconnected = "disconnected",
  Reconnecting = "reconnecting",
  Failed = "failed",
}

class ReconnectManager {
  private state: ReconnectState = ReconnectState.Disconnected;
  private attempt = 0;
  private reconnectTimer: NodeJS.Timeout | null = null;

  async reconnect(): Promise<void> {
    if (this.state === ReconnectState.Reconnecting) {
      return; // Already reconnecting
    }

    this.state = ReconnectState.Reconnecting;
    this.attempt++;

    if (this.attempt > this.policy.maxAttempts) {
      this.state = ReconnectState.Failed;
      this.onError(new Error("Max reconnection attempts exceeded"));
      return;
    }

    const { delayMs } = computeBackoff(this.attempt, this.policy);

    log.info(`Reconnecting in ${delayMs}ms (attempt ${this.attempt})`);

    await sleepWithAbort(delayMs, this.abortSignal);

    try {
      await this.connect();
      this.state = ReconnectState.Connected;
      this.attempt = 0; // Reset on success
    } catch (err) {
      log.warn(`Reconnection failed: ${err}`);
      await this.reconnect(); // Retry
    }
  }

  disconnect() {
    this.state = ReconnectState.Disconnecting;
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }
    this.socket?.close();
    this.state = ReconnectState.Disconnected;
  }
}
```

### Graceful Disconnect

```typescript
function close(code = 1000, reason?: string) {
  if (closed) return;
  closed = true;

  // Clear timers
  clearTimeout(handshakeTimer);
  clearInterval(livenessTimer);

  // Remove from connection pool
  if (client) {
    clients.delete(client);
    removeRemoteNodeInfo(client.connId);
  }

  // Broadcast presence update
  broadcastPresenceSnapshot(clients);
  incrementPresenceVersion();

  // Close WebSocket with proper code
  try {
    socket.close(code, reason);
  } catch {
    // Ignore close errors
  }

  // Log disconnect
  const durationMs = Date.now() - openedAt;
  logWs("out", "close", {
    connId,
    remoteAddr,
    durationMs,
    code,
    reason,
    closeCause,
    ...closeMeta,
  });
}
```

### Close Codes

```typescript
// Custom close codes for SophiaClaw
const CloseCodes = {
  Normal: 1000,
  GoingAway: 1001,
  ProtocolError: 1002,
  UnsupportedData: 1003,
  AbnormalClosure: 1006,
  InvalidPayload: 1007,
  PolicyViolation: 1008,
  MessageTooBig: 1009,
  MandatoryExtension: 1010,
  InternalServerError: 1011,

  // SophiaClaw custom codes
  AuthFailed: 4000,
  HandshakeTimeout: 4001,
  ConnectionStale: 4002,
  ServerShutdown: 4003,
};
```

---

## Connection Recovery

### Session State Preservation

```typescript
// Preserve session state across reconnections
interface SessionState {
  lastMessageId: string; // Last processed message
  pendingRequests: Map<
    string,
    {
      resolve: (result: unknown) => void;
      reject: (error: Error) => void;
      timeout: NodeJS.Timeout;
    }
  >;
  subscriptionState: Set<string>; // Active subscriptions
}

class SessionManager {
  private state = new Map<string, SessionState>();

  onDisconnect(connId: string) {
    const session = this.state.get(connId);
    if (session) {
      // Preserve state for recovery
      this.pendingSessions.set(connId, {
        state: session,
        timestamp: Date.now(),
      });
    }
  }

  async onReconnect(connId: string, lastMessageId: string): Promise<SessionState | null> {
    const pending = this.pendingSessions.get(connId);
    if (!pending) {
      return null; // No state to recover
    }

    // Check if state is still valid (within recovery window)
    const age = Date.now() - pending.timestamp;
    if (age > RECOVERY_WINDOW_MS) {
      this.pendingSessions.delete(connId);
      return null;
    }

    // Replay missed messages
    await this.replayMessages(connId, lastMessageId);

    return pending.state;
  }
}
```

### Message Replay

```typescript
async function replayMessages(connId: string, lastMessageId: string): Promise<void> {
  const messages = await messageStore.getAfter(lastMessageId, {
    limit: MAX_REPLAY_MESSAGES,
  });

  for (const message of messages) {
    sendToClient(connId, message);
  }

  log.info(`Replayed ${messages.length} messages to ${connId}`);
}
```

---

## Error Handling

### Connection Error Categories

```typescript
// In src/web/session.ts
export function formatError(err: unknown): string {
  if (err instanceof Error) {
    return err.message;
  }

  // Extract Boom-like error details (common in Baileys)
  const boom = extractBoomDetails(err);
  const status = boom?.statusCode ?? getStatusCode(err);
  const code = (err as { code?: unknown })?.code;

  const pieces: string[] = [];
  if (typeof status === "number") {
    pieces.push(`status=${status}`);
  }
  if (boom?.error) {
    pieces.push(boom.error);
  }
  if (boom?.message) {
    pieces.push(boom.message);
  }
  if (typeof code === "string" || typeof code === "number") {
    pieces.push(`code=${code}`);
  }

  return pieces.join(" ") || String(err);
}
```

### Disconnect Reason Mapping

```typescript
// Map Baileys disconnect reasons to reconnection behavior
function shouldReconnect(reason: DisconnectReason): boolean {
  switch (reason) {
    case DisconnectReason.loggedOut:
      return false; // User logged out, don't reconnect
    case DisconnectReason.connectionClosed:
      return true; // Network issue, reconnect
    case DisconnectReason.connectionLost:
      return true; // Network issue, reconnect
    case DisconnectReason.restartRequired:
      return true; // Server restart, reconnect after delay
    default:
      return false; // Unknown, play it safe
  }
}
```

---

## Performance Considerations

### Connection Pooling

```typescript
// Limit concurrent connections
const MAX_CONNECTIONS = 1000;

function handleConnection(socket: WebSocket) {
  if (clients.size >= MAX_CONNECTIONS) {
    socket.close(4008, "Server at capacity");
    return;
  }

  // Accept connection...
}
```

### Memory Management

```typescript
// Periodic cleanup of stale connections
function cleanupStaleConnections() {
  const now = Date.now();
  const maxAge = CONNECTION_MAX_AGE_MS;

  for (const client of clients) {
    const age = now - client.openedAt;
    if (age > maxAge) {
      close(client.connId, 4003, "Connection expired");
    }
  }
}

setInterval(cleanupStaleConnections, CLEANUP_INTERVAL_MS);
```

### Backpressure Handling

```typescript
interface GatewayWsClient {
  socket: WebSocket;
  sendQueue: Array<unknown>;
  isSlow: boolean;
}

function sendWithBackpressure(client: GatewayWsClient, message: unknown): boolean {
  if (client.sendQueue.length > MAX_QUEUE_SIZE) {
    client.isSlow = true;
    return false; // Drop message
  }

  if (client.socket.readyState === WebSocket.OPEN) {
    try {
      client.socket.send(JSON.stringify(message));
      return true;
    } catch {
      // Queue for later
      client.sendQueue.push(message);
      return true;
    }
  }

  // Connection not ready, queue message
  client.sendQueue.push(message);
  return true;
}
```

---

## Testing Strategies

### Connection Lifecycle Tests

```typescript
describe("WebSocket connection", () => {
  it("completes handshake within timeout", async () => {
    const client = new WebSocketClient(url);

    await expect(client.connect()).resolves.toBeDefined();
    expect(client.state).toBe("connected");
  });

  it("closes on handshake timeout", async () => {
    const server = createMockServer({ handshakeDelay: 10_000 });
    const client = new WebSocketClient(server.url, { handshakeTimeout: 1000 });

    await expect(client.connect()).rejects.toThrow("Handshake timeout");
  });

  it("reconnects after disconnection", async () => {
    const client = new WebSocketClient(url, {
      reconnect: { maxAttempts: 3 },
    });

    await client.connect();
    server.simulateDisconnect();

    await waitFor(() => {
      expect(client.state).toBe("connected");
      expect(client.attempts).toBe(1);
    });
  });
});
```

### Heartbeat Tests

```typescript
describe("heartbeat", () => {
  it("detects stale connections", async () => {
    const client = new WebSocketClient(url, { heartbeatSeconds: 1 });
    await client.connect();

    // Block pong responses
    server.blockPongs();

    await waitFor(() => {
      expect(client.state).toBe("reconnecting");
    });
  });

  it("maintains NAT mappings", async () => {
    const client = new WebSocketClient(url, { heartbeatSeconds: 30 });
    await client.connect();

    await sleep(60_000); // Wait 2 heartbeat intervals

    expect(client.pingsSent).toBeGreaterThanOrEqual(2);
    expect(client.pongsReceived).toBeGreaterThanOrEqual(2);
  });
});
```

---

## Related Documentation

- [Gateway Protocol](/gateway/protocol)
- [Gateway Heartbeat](/gateway/heartbeat)
- [Remote Gateway](/gateway/remote)
- [Multiple Gateways](/gateway/multiple-gateways)