meshcentral-openclaw-bridge/implementation/src/meshctrl-client.ts

/**
 * MeshCentral protocol client using the official meshctrl.js WebSocket interface.
 *
 * Implements typed framing forMeshCtrl actions and responses.
 * Uses secure TLS validation (no rejectUnauthorized: false).
 *
 * @see RESEARCH_INTEGRATION.md for protocol analysis
 */

import { Logger, getGlobalLogger } from './logger';

// Token bucket rate limiter and concurrency limiter are defined below.

/** Safe default values for rate limiting and concurrency controls. */
export const RATE_LIMIT_DEFAULTS = {
  /** Maximum tokens in bucket (burst capacity) */
  burstCapacity: 5,
  /** Maximum requests per second */
  maxRequestsPerSecond: 10,
  /** Maximum time to wait for rate limit (ms) */
  rateLimitTimeoutMs: 30_000,
} as const;

export const CONCURRENCY_DEFAULTS = {
  /** Maximum concurrent requests */
  maxConcurrentRequests: 5,
} as const;

/** Token bucket rate limiter for bounded request concurrency.
 *  Allows up to `maxTokens` requests per time window, with refill rate of `refillRate` tokens/second.
 */
export class TokenBucket {
  private tokens: number;
  private lastRefill: number;
  private readonly maxTokens: number;
  private readonly refillRate: number; // tokens per second
  private readonly timeoutMs: number;
  private readonly logger?: Logger;

/**
   * Create a TokenBucket with validated safe defaults.
   * @param maxTokens - Maximum burst capacity. Must be finite positive integer >= 1 and <= 1000.
   * @param refillRate - Tokens added per second. Must be finite positive number >= 0.01 and <= 10000.
   * @param timeoutMs - Maximum wait time in ms. Must be finite positive integer >= 100 and <= 300_000.
   * @param logger - Optional logger for validation warnings. Uses console.warn if not provided.
   */
  constructor(
    maxTokens: number,
    refillRate: number,
    timeoutMs: number,
    logger?: Logger,
  ) {
    // Validate maxTokens
    if (typeof maxTokens !== 'number' || !isFinite(maxTokens) || !Number.isInteger(maxTokens) || maxTokens < 1) {
      throw new Error(
        `maxTokens must be a finite positive integer >= 1, got ${maxTokens}`,
      );
    }
    if (maxTokens > 1000) {
      const msg = `maxTokens ${maxTokens} exceeds recommended max 1000, using 1000 for safety`;
      logger?.warn(msg);
      if (!logger) {
        // eslint-disable-next-line no-console
        console.warn(msg);
      }
      this.maxTokens = 1000;
    } else {
      this.maxTokens = maxTokens;
    }

    // Validate refillRate
    if (typeof refillRate !== 'number' || !isFinite(refillRate) || refillRate < 0.01) {
      throw new Error(
        `refillRate must be a finite positive number >= 0.01, got ${refillRate}`,
      );
    }
    if (refillRate > 10000) {
      const msg = `refillRate ${refillRate} exceeds recommended max 10000, using 10000 for safety`;
      logger?.warn(msg);
      if (!logger) {
        // eslint-disable-next-line no-console
        console.warn(msg);
      }
      this.refillRate = 10000;
    } else {
      this.refillRate = refillRate;
    }

    // Validate timeoutMs
    if (typeof timeoutMs !== 'number' || !isFinite(timeoutMs) || !Number.isInteger(timeoutMs) || timeoutMs < 100) {
      throw new Error(
        `timeoutMs must be a finite positive integer >= 100ms, got ${timeoutMs}`,
      );
    }
    if (timeoutMs > 300_000) {
      const msg = `timeoutMs ${timeoutMs} exceeds recommended max 300000ms, using 300000 for safety`;
      logger?.warn(msg);
      if (!logger) {
        // eslint-disable-next-line no-console
        console.warn(msg);
      }
      this.timeoutMs = 300_000;
    } else {
      this.timeoutMs = timeoutMs;
    }

    this.tokens = this.maxTokens;
    this.lastRefill = Date.now();
  }

  /** Try to consume a token. Returns true if successful, false if rate limited. */
  acquire(): boolean {
    this._refill();

    if (this.tokens >= 1) {
      this.tokens--;
      return true;
    }
    return false;
  }

  /** Wait for a token to become available. Resolves when acquired, rejects on timeout. */
  async acquireWithWait(): Promise<void> {
    const startTime = Date.now();

    // eslint-disable-next-line no-constant-condition
    while (true) {
      if (this.acquire()) {
        return;
      }

      const elapsed = Date.now() - startTime;
      if (elapsed >= this.timeoutMs) {
        throw new Error(`Rate limit wait timeout after ${elapsed}ms`);
      }

      // Calculate how long until next token (in ms)
      const tokensNeeded = 1 - this.tokens;
      const waitMs = Math.min((tokensNeeded / this.refillRate) * 1000, this.timeoutMs - elapsed);
      await new Promise((resolve) => setTimeout(resolve, Math.ceil(waitMs)));
    }
  }

  private _refill(): void {
    const now = Date.now();
    const elapsedSeconds = (now - this.lastRefill) / 1000;
    const tokensToAdd = elapsedSeconds * this.refillRate;

    if (tokensToAdd > 0) {
      this.tokens = Math.min(this.maxTokens, this.tokens + tokensToAdd);
      this.lastRefill = now;
    }
  }
}

/** Bounded concurrency controller using a semaphore pattern. */
export class ConcurrencyLimiter {
  private pending: number;
  private readonly maxPending: number;
  private waiting: Array<() => void> = [];
  private readonly logger?: Logger;

  /**
   * Create a concurrency limiter with safe defaults.
   * @param maxPending - Maximum concurrent requests. Must be finite positive integer >= 1 and <= 100. Defaults to 5.
   * @param logger - Optional logger for validation warnings. Uses console.warn if not provided.
   */
  constructor(maxPending: number = CONCURRENCY_DEFAULTS.maxConcurrentRequests, logger?: Logger) {
    this.logger = logger;
    // Validate maxPending
    if (typeof maxPending !== 'number' || !isFinite(maxPending) || !Number.isInteger(maxPending) || maxPending < 1) {
      throw new Error(
        `maxConcurrentRequests must be a finite positive integer >= 1, got ${maxPending}`,
      );
    }
    if (maxPending > 100) {
      const msg = `maxConcurrentRequests ${maxPending} exceeds recommended max 100, using 100 for safety`;
      this.logger?.warn(msg);
      if (!this.logger) {
        // eslint-disable-next-line no-console
        console.warn(msg);
      }
      this.maxPending = 100;
    } else {
      this.maxPending = maxPending;
    }
    this.pending = 0;
  }

  /** Try to enter a critical section. Returns true if allowed, false if at capacity. */
  tryEnter(): boolean {
    if (this.pending < this.maxPending) {
      this.pending++;
      return true;
    }
    return false;
  }

  /** Wait until we can enter the critical section. Resolves when slot is available. */
  async enter(): Promise<void> {
    return new Promise<void>((resolve) => {
      // Try to acquire immediately
      if (this.tryEnter()) {
        resolve();
        return;
      }

      // Queue the resolver to be called when a slot opens
      this.waiting.push(resolve);
    });
  }

  /** Exit the critical section and wake a waiting caller if any. */
  exit(): void {
    // Check if there's a waiting caller who can proceed
    const resolve = this.waiting.shift();
    if (resolve) {
      // A caller was waiting - they will proceed and acquire a slot
      // pending stays the same (one slot freed, one waiter takes it)
      resolve();
    } else if (this.pending > 0) {
      // No waiters, just decrement pending
      this.pending--;
    }
  }

  get pendingCount(): number {
    return this.pending;
  }

  /** Get the maximum concurrent requests allowed. */
  getMaxPending(): number {
    return this.maxPending;
  }
}

import { WebSocket } from 'ws';
import { z } from 'zod';
import { validateCredentials } from './contracts';

// ========== Protocol Constants ==========
export const CONTROL_PATH = '/control.ashx';
export const DEFAULT_TIMEOUT_MS = 30_000;
export const MAX_RESPONSE_SIZE = 1_048_576; // 1 MB

// ========== Action Types ==========
export enum MeshCtrlAction {
  ServerVersion = 'serverversion',
  UserInfo = 'userinfo',
  Meshes = 'meshes',
  Nodes = 'nodes',
  Events = 'events',
  DeviceInfo = 'deviceinfo',
}

// ========== Request Types ==========
export interface MeshCtrlRequest {
  action: string;
  /** MeshCentral's control protocol correlates requests with responseid. */
  responseid?: string;
  /** Retained for compatibility with older fixtures; not used for wire correlation. */
  id?: number;
  [key: string]: unknown;
}

export interface ServerVersionRequest {
  action: MeshCtrlAction.ServerVersion;
}

export interface UserinfoRequest {
  action: MeshCtrlAction.UserInfo;
}

export interface MesheMeshesRequest {
  action: MeshCtrlAction.Meshes;
}

export interface NodesRequest {
  action: MeshCtrlAction.Nodes;
  meshid: string;
}

export interface DeviceInfoRequest {
  action: MeshCtrlAction.DeviceInfo;
  id: string;
}

export interface EventsRequest {
  action: MeshCtrlAction.Events;
  stop?: number;
  count?: number;
}

// ========== Response Types ==========
export interface MeshCtrlResponse {
  action: string;
  responseid?: string;
  result?: string;
  tags?: Record<string, unknown>;
  id?: number;
  err?: string;
  [key: string]: unknown;
}

export interface ServerVersionResponse {
  action: MeshCtrlAction.ServerVersion;
  ver?: string;
  err?: string;
}

export interface UserInfoResponse {
  action: MeshCtrlAction.UserInfo;
  user?: {
    id: string;
    name: string;
    domain: string;
    admin: number;
  };
  err?: string;
}

export interface MeshesResponse {
  action: MeshCtrlAction.Meshes;
  meshes?: Array<{
    id?: string;
    _id?: string;
    name: string;
    description?: string;
    desc?: string;
  }>;
  err?: string;
}

export interface NodesResponse {
  action: MeshCtrlAction.Nodes;
  nodes?: Record<string, Array<{
    id?: string;
    _id?: string;
    name: string;
    meshid?: string;
    online?: boolean;
    conn?: number;
    osdesc?: string;
    ip?: string[];
    agent?: unknown;
    device?: {
      os?: string;
      uptime?: number;
      lastseen?: number;
      ip?: string[];
      agentver?: string;
    };
  }>>;
  err?: string;
}

export interface DeviceInfoResponse {
  action: MeshCtrlAction.DeviceInfo;
  device?: {
    id: string;
    name: string;
    meshid: string;
    online: boolean;
    os?: string;
    uptime?: number;
    lastseen?: number;
    ip?: string[];
    agentver?: string;
  };
  err?: string;
}

export interface EventsResponse {
  action: MeshCtrlAction.Events;
  events?: Array<{
    id: string;
    time: number;
    type: string;
    msg: string;
  }>;
  err?: string;
}

const responseBase = {
  responseid: z.string().optional(),
  result: z.string().optional(),
  id: z.number().optional(),
  err: z.string().optional(),
};

const responseSchemas: Record<string, z.ZodTypeAny> = {
  [MeshCtrlAction.ServerVersion]: z.object({
    ...responseBase,
    action: z.literal(MeshCtrlAction.ServerVersion),
    ver: z.string().optional(),
  }).passthrough(),
  [MeshCtrlAction.UserInfo]: z.object({
    ...responseBase,
    action: z.literal(MeshCtrlAction.UserInfo),
    user: z.object({
      id: z.string(), name: z.string(), domain: z.string(), admin: z.number(),
    }).optional(),
  }).passthrough(),
  [MeshCtrlAction.Meshes]: z.object({
    ...responseBase,
    action: z.literal(MeshCtrlAction.Meshes),
    meshes: z.array(z.object({
      id: z.string().optional(), _id: z.string().optional(), name: z.string(),
      description: z.string().optional(), desc: z.string().optional(),
    }).passthrough()).optional(),
  }).passthrough(),
  [MeshCtrlAction.Nodes]: z.object({
    ...responseBase,
    action: z.literal(MeshCtrlAction.Nodes),
    nodes: z.record(z.array(z.object({
      id: z.string().optional(), _id: z.string().optional(), name: z.string(),
      meshid: z.string().optional(), online: z.boolean().optional(),
      conn: z.number().optional(), osdesc: z.string().optional(),
      ip: z.array(z.string()).optional(), agent: z.unknown().optional(),
      device: z.object({
        os: z.string().optional(), uptime: z.number().optional(), lastseen: z.number().optional(),
        ip: z.array(z.string()).optional(), agentver: z.string().optional(),
      }).passthrough().optional(),
    }).passthrough())).optional(),
  }).passthrough(),
  [MeshCtrlAction.DeviceInfo]: z.object({
    ...responseBase,
    action: z.literal(MeshCtrlAction.DeviceInfo),
    device: z.object({
      id: z.string(), name: z.string(), meshid: z.string(), online: z.boolean(),
      os: z.string().optional(), uptime: z.number().optional(), lastseen: z.number().optional(),
      ip: z.array(z.string()).optional(), agentver: z.string().optional(),
    }).passthrough().optional(),
  }).passthrough(),
  [MeshCtrlAction.Events]: z.object({
    ...responseBase,
    action: z.literal(MeshCtrlAction.Events),
    events: z.array(z.object({
      id: z.string(), time: z.number(), type: z.string(), msg: z.string(),
    }).passthrough()).optional(),
  }).passthrough(),
};

// ========== Authentication ==========
export interface AuthCredentials {
  username: string;
  password: string;
  totpToken?: string;
}

/**
 * Encode credentials for x-meshauth header.
 * Format: base64(username:password[,totp])
 */
export function encodeAuthHeader(credentials: AuthCredentials): string {
  validateCredentials(credentials);
  const { username, password, totpToken } = credentials;

  // MeshCentral's meshctrl.js wire format is three comma-separated base64
  // fields: username, password, and (when present) a base64 TOTP field.
  const header = `${Buffer.from(username).toString('base64')},${Buffer.from(password).toString('base64')}`;
  return totpToken
    ? `${header},${Buffer.from(totpToken).toString('base64')}`
    : header;
}

/**
 * Sanitize credentials for safe logging.
 * Never log actual credential values - only whether they exist.
 */
// Re-exported from contracts.ts for convenience
export { sanitizeCredentials } from './contracts';

// ========== Protocol Framing ==========
/**
 * Create a MeshCtrl request message.
 */
export function createRequest(action: string, payload?: Record<string, unknown>, id?: number): MeshCtrlRequest {
  const request: MeshCtrlRequest = { action, ...(payload ?? {}) };
  if (id !== undefined) {
    request.id = id;
  }
  return request;
}

/**
 * Parse a MeshCtrl response message with size validation.
 */
export function parseResponse(data: string, maxSize: number = MAX_RESPONSE_SIZE): MeshCtrlResponse {
  if (Buffer.byteLength(data, 'utf8') > maxSize) {
    throw new Error('MeshCentral response exceeds the configured size limit');
  }

  let parsed: unknown;
  try {
    parsed = JSON.parse(data);
  } catch {
    throw new Error('MeshCentral response is not valid JSON');
  }

  const result = z.object({
    action: z.string().min(1),
    // Unsolicited MeshCentral session/event messages may use different id
    // shapes. The request-specific schemas validate fields on matched
    // responses; the generic envelope must not reject bootstrap traffic.
    id: z.unknown().optional(),
    err: z.string().optional(),
  }).passthrough().safeParse(parsed);
  if (!result.success) {
    throw new Error('MeshCentral response failed protocol validation');
  }
  return result.data as MeshCtrlResponse;
}

export function validateResponseForAction(response: MeshCtrlResponse, action: string): MeshCtrlResponse {
  if (response.action !== action) {
    throw new Error('MeshCentral response action did not match the request');
  }

  const schema = responseSchemas[action];
  if (!schema) {
    return response;
  }

  const result = schema.safeParse(response);
  if (!result.success) {
    throw new Error('MeshCentral response failed schema validation');
  }
  return result.data as MeshCtrlResponse;
}

/**
 * Determine if a response indicates an error.
 */
export function isErrorResponse(response: MeshCtrlResponse): boolean {
  return response.err !== undefined && response.err !== '';
}

/**
 * Extract error message from a response, if any.
 */
export function getErrorResponseMessage(response: MeshCtrlResponse): string | null {
  if (isErrorResponse(response)) {
    return response.err ?? 'Unknown error';
  }
  return null;
}

// ========== Concurrency & Rate Limiting ==========
export interface RateLimitOptions {
  /** Maximum requests per second (RPS) */
  maxRequestsPerSecond?: number;
  /** Burst capacity for requests (max tokens in bucket) */
  burstCapacity?: number;
  /** Maximum time to wait for rate limit (ms) */
  rateLimitTimeoutMs?: number;
}

export interface ConcurrencyOptions {
  /** Maximum concurrent requests on a single connection */
  maxConcurrentRequests?: number;
}

export interface MeshCtrlClientOptions extends RateLimitOptions, ConcurrencyOptions {
  url: string;
  timeoutMs?: number;
  // TLS configuration
  caFile?: string; // Optional CA certificate file path for custom CAs
}

/**
 * Validate rate limit options with safe defaults and fail-closed errors.
 */
export function validateRateLimitOptions(options?: RateLimitOptions): {
  maxRequestsPerSecond: number;
  burstCapacity: number;
  rateLimitTimeoutMs: number;
} {
  const maxRequestsPerSecond = options?.maxRequestsPerSecond ?? RATE_LIMIT_DEFAULTS.maxRequestsPerSecond;
  const burstCapacity = options?.burstCapacity ?? RATE_LIMIT_DEFAULTS.burstCapacity;
  const rateLimitTimeoutMs = options?.rateLimitTimeoutMs ?? RATE_LIMIT_DEFAULTS.rateLimitTimeoutMs;

  // Validate maxRequestsPerSecond
  if (typeof maxRequestsPerSecond !== 'number' || !isFinite(maxRequestsPerSecond)) {
    throw new Error(`maxRequestsPerSecond must be a finite number, got ${maxRequestsPerSecond}`);
  }
  if (maxRequestsPerSecond < 1) {
    throw new Error(`maxRequestsPerSecond must be >= 1, got ${maxRequestsPerSecond}`);
  }
  if (maxRequestsPerSecond > 10000) {
    throw new Error(`maxRequestsPerSecond must be <= 10000 for safety, got ${maxRequestsPerSecond}`);
  }

  // Validate burstCapacity
  if (!Number.isInteger(burstCapacity) || burstCapacity < 1) {
    throw new Error(`burstCapacity must be a positive integer >= 1, got ${burstCapacity}`);
  }
  if (burstCapacity > 1000) {
    throw new Error(`burstCapacity must be <= 1000 for safety, got ${burstCapacity}`);
  }

  // Validate rateLimitTimeoutMs
  if (!Number.isInteger(rateLimitTimeoutMs) || rateLimitTimeoutMs < 100) {
    throw new Error(`rateLimitTimeoutMs must be an integer >= 100ms, got ${rateLimitTimeoutMs}`);
  }
  if (rateLimitTimeoutMs > 300_000) {
    throw new Error(`rateLimitTimeoutMs must be <= 300000ms for safety, got ${rateLimitTimeoutMs}`);
  }

  return {
    maxRequestsPerSecond,
    burstCapacity,
    rateLimitTimeoutMs,
  };
}

/**
 * Validate concurrency options with safe defaults and fail-closed errors.
 */
export function validateConcurrencyOptions(options?: ConcurrencyOptions): {
  maxConcurrentRequests: number;
} {
  const maxConcurrentRequests = options?.maxConcurrentRequests ?? CONCURRENCY_DEFAULTS.maxConcurrentRequests;

  // Validate maxConcurrentRequests
  if (!Number.isInteger(maxConcurrentRequests) || maxConcurrentRequests < 1) {
    throw new Error(`maxConcurrentRequests must be a positive integer >= 1, got ${maxConcurrentRequests}`);
  }
  if (maxConcurrentRequests > 100) {
    throw new Error(`maxConcurrentRequests must be <= 100 for safety, got ${maxConcurrentRequests}`);
  }

  return {
    maxConcurrentRequests,
  };
}

/**
 * Validate all MeshCtrlClient options with safe defaults and fail-closed errors.
 */
export function validateMeshCtrlClientOptions(options: MeshCtrlClientOptions): {
  url: string;
  timeoutMs: number;
  rateLimitConfig: {
    maxRequestsPerSecond: number;
    burstCapacity: number;
    rateLimitTimeoutMs: number;
  };
  concurrencyConfig: {
    maxConcurrentRequests: number;
  };
} {
  // Validate URL
  if (!options.url || typeof options.url !== 'string' || options.url.trim() === '') {
    throw new Error('url is required and must be a non-empty string');
  }

  // Validate timeoutMs
  const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
  if (!Number.isInteger(timeoutMs) || timeoutMs < 1000) {
    throw new Error(`timeoutMs must be an integer >= 1000ms, got ${timeoutMs}`);
  }
  if (timeoutMs > 60_000) {
    throw new Error(`timeoutMs must be <= 60000ms for safety, got ${timeoutMs}`);
  }

  const rateLimitConfig = validateRateLimitOptions(options);
  const concurrencyConfig = validateConcurrencyOptions(options);

  return {
    url: options.url.trim(),
    timeoutMs,
    rateLimitConfig,
    concurrencyConfig,
  };
}

/**
 * MeshCentral protocol client using the control.ashx WebSocket.
 *
 * Implements secure TLS validation and typed protocol framing.
 * Supports bounded request concurrency and rate limiting via token bucket algorithm.
 */
export class MeshCtrlClient {
  private readonly url: string;
  private readonly timeoutMs: number;
  private readonly caFile?: string;
  private ws: WebSocket | null = null;
  private connected = false;
  private requestSequence = 0;
  private connectionReadyAt = 0;
  private pendingRequests = new Map<string, {
    action: string;
    resolve: (response: MeshCtrlResponse) => void;
    reject: (error: Error) => void;
    timeout: NodeJS.Timeout;
  }>();

  // Concurrency and rate limiting
  private readonly concurrencyLimiter: ConcurrencyLimiter;
  private readonly rateLimiter: TokenBucket | null;

  // Logger
  private readonly logger: Logger;

  constructor(options: MeshCtrlClientOptions) {
    const validated = validateMeshCtrlClientOptions(options);

    this.url = validated.url;
    this.timeoutMs = validated.timeoutMs;
    this.caFile = options.caFile;
    // Use a dedicated logger for this client instance
    this.logger = getGlobalLogger();
    // Set the logger name to be more descriptive
    this.logger.setName('meshctrl-client');

    // Initialize concurrency limiter with validated config
    this.concurrencyLimiter = new ConcurrencyLimiter(validated.concurrencyConfig.maxConcurrentRequests, this.logger);

    // Initialize rate limiter with validated config (disabled if maxRps is 0)
    const { maxRequestsPerSecond, burstCapacity, rateLimitTimeoutMs } = validated.rateLimitConfig;
    if (maxRequestsPerSecond > 0 && burstCapacity > 0) {
      this.rateLimiter = new TokenBucket(
        burstCapacity,
        maxRequestsPerSecond,
        rateLimitTimeoutMs,
        this.logger,
      );
    } else {
      this.rateLimiter = null;
    }
  }

  /**
   * Connect to MeshCentral control channel.
   * @param credentials - Authentication credentials
   * @returns Promise that resolves when connected and authenticated
   */
  async connect(credentials: AuthCredentials): Promise<void> {
    if (this.connected) {
      throw new Error('Already connected');
    }

    // Log connection attempt (redacted)
    this.logger.audit('AUTH_ATTEMPT - PENDING');

    // Construct WebSocket URL
    const wsUrl = this._buildWebSocketUrl();

    // Set authentication header for x-meshauth
    const authHeader = encodeAuthHeader(credentials);

    // Create WebSocket with explicit secure TLS validation
    this.ws = new WebSocket(wsUrl, {
      perMessageDeflate: false,
      headers: {
        'x-meshauth': authHeader,
      },
      // Explicit TLS validation - never bypass
      rejectUnauthorized: true,
    });

    // Setup WebSocket close handler with diagnostics
    this.ws.on('close', (code: number, reason: Buffer) => {
      this.logger.debug('MeshCentral socket closed', { code, reasonBytes: reason.length });
      const reasonText = reason.toString();
      this.logger.warn(`WebSocket close event: code=${code}, reason="${reasonText}", url=${wsUrl}`);
      this.connected = false;
      for (const [requestId, pending] of this.pendingRequests) {
        clearTimeout(pending.timeout);
        pending.reject(new Error('MeshCentral connection closed'));
        this.pendingRequests.delete(requestId);
      }
      this.ws = null;
    });

    // Setup WebSocket error handler with diagnostics
    this.ws.on('error', (error: Error) => {
      const errorMessage = error instanceof Error ? error.message : 'unknown error';
      this.logger.error(`WebSocket error event: ${errorMessage}, url=${wsUrl}`);
      this.logger.debug('MeshCentral socket error state', {
        connected: this.connected,
        hasSocket: this.ws !== null,
      });
    });
    const handleMessage = (data: Buffer | string): void => {
      let message: unknown;
      try {
        message = JSON.parse(Buffer.isBuffer(data) ? data.toString() : data);
      } catch {
        return;
      }
      if (typeof message !== 'object' || message === null) return;
      const response = message as { action?: unknown; responseid?: unknown };
      if (typeof response.action !== 'string') return;
      const responseId = typeof response.responseid === 'string' ? response.responseid : undefined;
      const pending = responseId ? this.pendingRequests.get(responseId) : undefined;
      // Legacy `meshes` replies omit responseid. They are safe to correlate
      // only while this client has exactly one pending meshes request.
      const legacyMeshes = response.action === MeshCtrlAction.Meshes
        ? [...this.pendingRequests.entries()].filter(([, item]) => item.action === MeshCtrlAction.Meshes)
        : [];
      const target = pending ?? (legacyMeshes.length === 1 ? legacyMeshes[0][1] : undefined);
      const targetId = pending ? responseId : (legacyMeshes.length === 1 ? legacyMeshes[0][0] : undefined);
      if (!target || !targetId) return;

      try {
        const parsedResponse = parseResponse(Buffer.isBuffer(data) ? data.toString() : data);
        const validated = validateResponseForAction(parsedResponse, target.action);
        clearTimeout(target.timeout);
        this.pendingRequests.delete(targetId);
        if (isErrorResponse(validated)) target.reject(new Error('MeshCentral request failed'));
        else target.resolve(validated);
      } catch (error) {
        this.logger.error('MeshCentral dispatcher failed', {
          action: response.action,
          error: error instanceof Error ? error.message : 'unknown',
        });
        clearTimeout(target.timeout);
        this.pendingRequests.delete(targetId);
        target.reject(error instanceof Error ? error : new Error('Unknown error'));
      }
    };

    // Wait for connection
    await new Promise<void>((resolve, reject) => {
      const timeout = setTimeout(() => {
        this.connected = false;
        this._cleanup();
        reject(new Error(`Connection timed out after ${this.timeoutMs}ms`));
      }, this.timeoutMs);

      this.ws!.once('open', () => {
        clearTimeout(timeout);
        this.connected = true;
        // MeshCentral emits serverinfo/userinfo bootstrap traffic immediately
        // after upgrade. The upstream meshctrl client waits briefly before
        // its first command; mirror that bounded settling window here.
        this.connectionReadyAt = Date.now() + 750;

        // MeshCentral authenticates from the x-meshauth header during the
        // WebSocket upgrade; the first application message is a request,
        // correlated with responseid (not numeric id).
        // Log successful connection
        this.logger.audit('CONNECTION_ESTABLISHED - SUCCESS');

        // Match meshctrl.js: install the application dispatcher only after
        // the WebSocket upgrade has completed. Some MeshCentral deployments
        // close sessions when an application listener is active during the
        // upgrade/bootstrap transition.
        this.ws?.on('message', handleMessage);

        resolve();
      });

      this.ws!.once('error', (error: Error) => {
        clearTimeout(timeout);
        this._cleanup();

        // Log connection failure
        this.logger.audit(`AUTH_ATTEMPT - FAILURE: ${error.message}`);

        reject(new Error(`Connection failed: ${error instanceof Error ? error.message : 'unknown'}`));
      });
    });
  }

  /**
   * Disconnect from MeshCentral control channel.
   */
  async disconnect(): Promise<void> {
    if (!this.connected) {
      return;
    }

    this._cleanup();
    this.connected = false;

    // Log disconnection
    this.logger.audit('CONNECTION_CLOSED - SUCCESS');
  }

/**
 * Get current concurrency status.
 */
  getConcurrencyStatus(): { pending: number; max: number } {
    return {
      pending: this.concurrencyLimiter.pendingCount,
      max: (this.concurrencyLimiter as unknown as { maxPending: number }).maxPending,
    };
  }

  /**
   * Get current rate limiter status.
   */
  getRateLimitStatus(): { tokens: number; maxTokens: number } | null {
    if (!this.rateLimiter) {
      return null;
    }
    // Access private fields via type assertion for status reporting
    const bucket = this.rateLimiter as unknown as { tokens: number; maxTokens: number };
    return {
      tokens: Math.floor(bucket.tokens),
      maxTokens: bucket.maxTokens,
    };
  }

  /**
   * Send a MeshCtrl request and wait for response.
   * Enforces bounded concurrency and rate limiting.
   * @param request - The request to send
   * @returns Parsed response with schema validation
   */
  async sendRequest(request: MeshCtrlRequest): Promise<MeshCtrlResponse> {
    if (!this.connected) {
      throw new Error('Not connected');
    }

    // Enforce rate limiting before concurrency
    if (this.rateLimiter) {
      await this.rateLimiter.acquireWithWait();
    }

    // Enforce bounded concurrency
    await this.concurrencyLimiter.enter();

    try {
      return await this._sendWithRetry(request, 3);
    } finally {
      // Always exit concurrency slot, even on error
      this.concurrencyLimiter.exit();
    }
  }

  /**
   * Send a request with retry logic for transient failures.
   */
  private async _sendWithRetry(request: MeshCtrlRequest, maxRetries: number): Promise<MeshCtrlResponse> {
    let lastError: Error | null = null;

    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        return await this._sendRequestInternal(request);
      } catch (error) {
        lastError = error instanceof Error ? error : new Error('Unknown error');

        if (attempt < maxRetries) {
          // Wait before retrying
          await new Promise((resolve) => setTimeout(resolve, 100 * attempt));
        }
      }
    }

    throw new Error(`Request failed after ${maxRetries} attempts: ${lastError?.message}`);
  }

  /**
   * Internal send implementation with timeout and bounded response validation.
   */
  private async _sendRequestInternal(request: MeshCtrlRequest): Promise<MeshCtrlResponse> {
    if (!this.ws || !this.connected) {
      throw new Error('Not connected');
    }

    return new Promise((resolve, reject) => {
      const requestId = request.responseid ?? `openclaw-${Date.now()}-${this.requestSequence++}`;
      const finalRequest = { ...request, responseid: requestId };
      const sendDelay = Math.max(0, this.connectionReadyAt - Date.now());
      const timeout = setTimeout(() => {
        this.pendingRequests.delete(requestId);
        reject(new Error(`Request timed out after ${this.timeoutMs}ms`));
      }, this.timeoutMs);
      this.pendingRequests.set(requestId, { action: request.action, resolve, reject, timeout });

      // Install the pending request before sending. MeshCentral may answer
      // immediately once its bootstrap phase has settled.
      setTimeout(() => {
        try {
          if (this.ws && this.connected) {
            this.logger.debug('MeshCentral request sent', { action: request.action });
            this.ws.send(JSON.stringify(finalRequest));
          } else {
            this.pendingRequests.delete(requestId);
            clearTimeout(timeout);
            reject(new Error('Not connected'));
          }
        } catch (error) {
          this.pendingRequests.delete(requestId);
          clearTimeout(timeout);
          reject(new Error(`Failed to send request: ${error instanceof Error ? error.message : 'unknown'}`));
        }
      }, sendDelay);
    });
  }

  /**
   * Build WebSocket URL from HTTP(S) base URL.
   */
  private _buildWebSocketUrl(): string {
    let url = this.url;

    // Convert http:// to ws:// and https:// to wss://
    if (url.startsWith('http://')) {
      url = 'ws://' + url.substring(7);
    } else if (url.startsWith('https://')) {
      url = 'wss://' + url.substring(8);
    }

    // Ensure path ends with /control.ashx
    if (!url.endsWith(CONTROL_PATH)) {
      if (!url.endsWith('/')) {
        url += '/';
      }
      url += CONTROL_PATH;
    }

    return url;
  }

  /**
   * Cleanup WebSocket resources.
   */
  private _cleanup(): void {
    if (!this.ws) {
      return;
    }

    // Capture WebSocket state before cleanup
    const readyState = this.ws.readyState;
    const wsUrl = (this.ws as unknown as { _url?: string })._url ?? this.url;

    // Remove all listeners to prevent duplicate handling
    this.ws.removeAllListeners();

    // Log cleanup reason based on WebSocket state
    if (readyState === WebSocket.CLOSED) {
      this.logger.warn(`WebSocket cleanup: connection already closed (url=${wsUrl})`);
    } else if (readyState === WebSocket.CLOSING) {
      this.logger.warn(`WebSocket cleanup: connection closing (url=${wsUrl})`);
    } else if (readyState === WebSocket.OPEN || readyState === WebSocket.CONNECTING) {
      this.logger.warn(`WebSocket cleanup: terminating active connection (url=${wsUrl}, readyState=${readyState})`);
      try {
        this.ws.close(1000, 'Cleanup');
      } catch (closeError) {
        this.logger.error(`WebSocket cleanup close failed: ${closeError instanceof Error ? closeError.message : 'unknown'}`);
      }
    } else {
      this.logger.warn(`WebSocket cleanup: unknown state ${readyState} (url=${wsUrl})`);
    }

    this.ws = null;
  }

  /**
   * Get connection status.
   */
  isConnected(): boolean {
    return this.connected && this.ws !== null;
  }
}