Files
zclaw_openfang/desktop/src/lib/gateway-client.ts
iven 74dbf42644 refactor(startup): simplify stack to Tauri-managed OpenFang + optional ChromeDriver
- Remove OpenFang CLI dependency from startup scripts
- OpenFang now bundled with Tauri and managed via gateway_start/gateway_status commands
- Add bootstrap screen in App.tsx to auto-start local gateway before UI loads
- Update Makefile: replace start-no-gateway with start-desktop-only
- Fix gateway config endpoints: use /api/config instead of /api/config/quick
- Add Playwright dependencies for future E2E testing
2026-03-17 14:08:03 +08:00

1969 lines
58 KiB
TypeScript

/**
* ZCLAW Gateway Client (Browser/Tauri side)
*
* WebSocket client for OpenFang Kernel protocol, designed to run
* in the Tauri React frontend. Uses native browser WebSocket API.
* Supports Ed25519 device authentication + JWT.
*
* OpenFang Configuration:
* - Port: 50051
* - WebSocket path: /ws
* - REST API: http://127.0.0.1:50051/api/*
* - Config format: TOML
*
* Security:
* - Device keys stored in OS keyring when available
* - Supports WSS (WebSocket Secure) for production
*/
import nacl from 'tweetnacl';
import type { GatewayConfigSnapshot, GatewayModelChoice } from './gateway-config';
import {
storeDeviceKeys,
getDeviceKeys,
deleteDeviceKeys,
} from './secure-storage';
import {
getQuickConfigFallback,
getWorkspaceInfoFallback,
getUsageStatsFallback,
getPluginStatusFallback,
getScheduledTasksFallback,
getSecurityStatusFallback,
isNotFoundError,
} from './api-fallbacks';
// === WSS Configuration ===
/**
* Whether to use WSS (WebSocket Secure) instead of WS.
* - Production: defaults to WSS for security
* - Development: defaults to WS for convenience
* - Override: set VITE_USE_WSS=false to force WS in production
*/
const USE_WSS = import.meta.env.VITE_USE_WSS !== 'false' && import.meta.env.PROD;
/**
* Default protocol based on WSS configuration.
*/
const DEFAULT_WS_PROTOCOL = USE_WSS ? 'wss://' : 'ws://';
/**
* Check if a URL points to localhost.
*/
function isLocalhost(url: string): boolean {
try {
const parsed = new URL(url);
return parsed.hostname === 'localhost' ||
parsed.hostname === '127.0.0.1' ||
parsed.hostname === '[::1]';
} catch {
return false;
}
}
// OpenFang endpoints (actual port is 50051, not 4200)
// Note: REST API uses relative path to leverage Vite proxy for CORS bypass
export const DEFAULT_GATEWAY_URL = `${DEFAULT_WS_PROTOCOL}127.0.0.1:50051/ws`;
export const REST_API_URL = ''; // Empty = use relative path (Vite proxy)
export const FALLBACK_GATEWAY_URLS = [
DEFAULT_GATEWAY_URL,
`${DEFAULT_WS_PROTOCOL}127.0.0.1:4200/ws`,
];
const GATEWAY_URL_STORAGE_KEY = 'zclaw_gateway_url';
const GATEWAY_TOKEN_STORAGE_KEY = 'zclaw_gateway_token';
// === Protocol Types ===
export interface GatewayRequest {
type: 'req';
id: string;
method: string;
params?: Record<string, unknown>;
}
export interface GatewayError {
code?: string;
message?: string;
details?: unknown;
}
export interface GatewayResponse {
type: 'res';
id: string;
ok: boolean;
payload?: unknown;
error?: GatewayError;
}
export interface GatewayEvent {
type: 'event';
event: string;
payload?: unknown;
seq?: number;
}
export interface GatewayPong {
type: 'pong';
timestamp?: number;
}
export type GatewayFrame = GatewayRequest | GatewayResponse | GatewayEvent | GatewayPong;
function createIdempotencyKey(): string {
if (typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function') {
return crypto.randomUUID();
}
return `idem_${Date.now()}_${Math.random().toString(36).slice(2, 10)}`;
}
export interface AgentStreamDelta {
stream: 'assistant' | 'tool' | 'lifecycle' | 'hand' | 'workflow';
delta?: string;
content?: string;
tool?: string;
toolInput?: string;
toolOutput?: string;
phase?: 'start' | 'end' | 'error';
runId?: string;
error?: string;
// Hand event fields
handName?: string;
handStatus?: string;
handResult?: unknown;
// Workflow event fields
workflowId?: string;
workflowStep?: string;
workflowStatus?: string;
workflowResult?: unknown;
}
/** OpenFang WebSocket stream event types */
export interface OpenFangStreamEvent {
type: 'text_delta' | 'phase' | 'response' | 'typing' | 'tool_call' | 'tool_result' | 'hand' | 'workflow' | 'error' | 'connected' | 'agents_updated';
content?: string;
phase?: 'streaming' | 'done';
state?: 'start' | 'stop';
tool?: string;
input?: unknown;
output?: string;
result?: unknown;
hand_name?: string;
hand_status?: string;
hand_result?: unknown;
workflow_id?: string;
workflow_step?: string;
workflow_status?: string;
workflow_result?: unknown;
message?: string;
code?: string;
agent_id?: string;
agents?: Array<{ id: string; name: string; status: string }>;
}
export type ConnectionState = 'disconnected' | 'connecting' | 'handshaking' | 'connected' | 'reconnecting';
type EventCallback = (payload: unknown) => void;
export function getStoredGatewayUrl(): string {
try {
const stored = localStorage.getItem(GATEWAY_URL_STORAGE_KEY);
return normalizeGatewayUrl(stored || DEFAULT_GATEWAY_URL);
} catch {
return DEFAULT_GATEWAY_URL;
}
}
export function setStoredGatewayUrl(url: string): string {
const normalized = normalizeGatewayUrl(url || DEFAULT_GATEWAY_URL);
try {
localStorage.setItem(GATEWAY_URL_STORAGE_KEY, normalized);
} catch { /* ignore localStorage failures */ }
return normalized;
}
export function getStoredGatewayToken(): string {
try {
return localStorage.getItem(GATEWAY_TOKEN_STORAGE_KEY) || '';
} catch {
return '';
}
}
export function setStoredGatewayToken(token: string): string {
const normalized = token.trim();
try {
if (normalized) {
localStorage.setItem(GATEWAY_TOKEN_STORAGE_KEY, normalized);
} else {
localStorage.removeItem(GATEWAY_TOKEN_STORAGE_KEY);
}
} catch {
/* ignore localStorage failures */
}
return normalized;
}
// === URL Normalization ===
/**
* Normalize a gateway URL to ensure correct protocol and path.
* - Ensures ws:// or wss:// protocol based on configuration
* - Ensures /ws path suffix
* - Handles both localhost and IP addresses
*/
export function normalizeGatewayUrl(url: string): string {
let normalized = url.trim();
// Remove trailing slashes except for protocol
normalized = normalized.replace(/\/+$/, '');
// Ensure protocol
if (!normalized.startsWith('ws://') && !normalized.startsWith('wss://')) {
normalized = USE_WSS ? `wss://${normalized}` : `ws://${normalized}`;
}
// Ensure /ws path
if (!normalized.endsWith('/ws')) {
normalized = `${normalized}/ws`;
}
return normalized;
}
// === Device Auth ===
interface DeviceKeys {
deviceId: string;
publicKey: Uint8Array;
secretKey: Uint8Array;
publicKeyBase64: string;
}
export interface LocalDeviceIdentity {
deviceId: string;
publicKeyBase64: string;
}
async function deriveDeviceId(publicKey: Uint8Array): Promise<string> {
const stableBytes = Uint8Array.from(publicKey);
const digest = await crypto.subtle.digest('SHA-256', stableBytes.buffer);
return Array.from(new Uint8Array(digest))
.map((byte) => byte.toString(16).padStart(2, '0'))
.join('');
}
async function generateDeviceKeys(): Promise<DeviceKeys> {
const keyPair = nacl.sign.keyPair();
const deviceId = await deriveDeviceId(keyPair.publicKey);
return {
deviceId,
publicKey: keyPair.publicKey,
secretKey: keyPair.secretKey,
publicKeyBase64: b64Encode(keyPair.publicKey),
};
}
/**
* Load device keys from secure storage.
* Uses OS keyring when available, falls back to localStorage.
*/
async function loadDeviceKeys(): Promise<DeviceKeys> {
// Try to load from secure storage (keyring or localStorage fallback)
const storedKeys = await getDeviceKeys();
if (storedKeys) {
try {
const deviceId = await deriveDeviceId(storedKeys.publicKey);
return {
deviceId,
publicKey: storedKeys.publicKey,
secretKey: storedKeys.secretKey,
publicKeyBase64: b64Encode(storedKeys.publicKey),
};
} catch (e) {
console.warn('[GatewayClient] Failed to load stored keys:', e);
// Invalid stored keys, clear and regenerate
await deleteDeviceKeys();
}
}
// Generate new keys
const keys = await generateDeviceKeys();
// Store in secure storage (keyring when available, localStorage fallback)
await storeDeviceKeys(keys.publicKey, keys.secretKey);
return keys;
}
export async function getLocalDeviceIdentity(): Promise<LocalDeviceIdentity> {
const keys = await loadDeviceKeys();
return {
deviceId: keys.deviceId,
publicKeyBase64: keys.publicKeyBase64,
};
}
/**
* Clear cached device keys to force regeneration on next connect.
* Useful when device signature validation fails repeatedly.
*/
export async function clearDeviceKeys(): Promise<void> {
try {
await deleteDeviceKeys();
console.log('[GatewayClient] Device keys cleared');
} catch (e) {
console.warn('[GatewayClient] Failed to clear device keys:', e);
}
}
function b64Encode(bytes: Uint8Array): string {
let binary = '';
for (let i = 0; i < bytes.length; i++) {
binary += String.fromCharCode(bytes[i]);
}
return btoa(binary).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
}
function buildDeviceAuthPayload(params: {
clientId: string;
clientMode: string;
deviceId: string;
nonce: string;
role: string;
scopes: string[];
signedAt: number;
token?: string;
}) {
return [
'v2',
params.deviceId,
params.clientId,
params.clientMode,
params.role,
params.scopes.join(','),
String(params.signedAt),
params.token || '',
params.nonce,
].join('|');
}
function signDeviceAuth(params: {
clientId: string;
clientMode: string;
deviceId: string;
nonce: string;
role: string;
scopes: string[];
secretKey: Uint8Array;
token?: string;
}): { signature: string; signedAt: number } {
const signedAt = Date.now();
const message = buildDeviceAuthPayload({
clientId: params.clientId,
clientMode: params.clientMode,
deviceId: params.deviceId,
nonce: params.nonce,
role: params.role,
scopes: params.scopes,
signedAt,
token: params.token,
});
const messageBytes = new TextEncoder().encode(message);
const signature = nacl.sign.detached(messageBytes, params.secretKey);
return {
signature: b64Encode(signature),
signedAt,
};
}
// === Client ===
export class GatewayClient {
private ws: WebSocket | null = null;
private openfangWs: WebSocket | null = null; // OpenFang stream WebSocket
private state: ConnectionState = 'disconnected';
private requestId = 0;
private pendingRequests = new Map<string, {
resolve: (value: unknown) => void;
reject: (reason: unknown) => void;
timer: number;
}>();
private eventListeners = new Map<string, Set<EventCallback>>();
private reconnectAttempts = 0;
private reconnectTimer: number | null = null;
private deviceKeysPromise: Promise<DeviceKeys>;
private streamCallbacks = new Map<string, {
onDelta: (delta: string) => void;
onTool?: (tool: string, input: string, output: string) => void;
onHand?: (name: string, status: string, result?: unknown) => void;
onComplete: () => void;
onError: (error: string) => void;
}>();
// Options
private url: string;
private token: string;
private autoReconnect: boolean;
private reconnectInterval: number;
private requestTimeout: number;
// Heartbeat
private heartbeatInterval: number | null = null;
private heartbeatTimeout: number | null = null;
private missedHeartbeats: number = 0;
private static readonly HEARTBEAT_INTERVAL = 30000; // 30 seconds
private static readonly HEARTBEAT_TIMEOUT = 10000; // 10 seconds
private static readonly MAX_MISSED_HEARTBEATS = 3;
// State change callbacks
onStateChange?: (state: ConnectionState) => void;
onLog?: (level: string, message: string) => void;
constructor(opts?: {
url?: string;
token?: string;
autoReconnect?: boolean;
reconnectInterval?: number;
requestTimeout?: number;
}) {
this.url = normalizeGatewayUrl(opts?.url || getStoredGatewayUrl());
this.token = opts?.token ?? getStoredGatewayToken();
this.autoReconnect = opts?.autoReconnect ?? true;
this.reconnectInterval = opts?.reconnectInterval || 3000;
this.requestTimeout = opts?.requestTimeout || 30000;
this.deviceKeysPromise = loadDeviceKeys();
}
updateOptions(opts?: {
url?: string;
token?: string;
autoReconnect?: boolean;
reconnectInterval?: number;
requestTimeout?: number;
}) {
if (!opts) return;
if (opts.url) {
this.url = normalizeGatewayUrl(opts.url);
}
if (opts.token !== undefined) {
this.token = opts.token;
}
if (opts.autoReconnect !== undefined) {
this.autoReconnect = opts.autoReconnect;
}
if (opts.reconnectInterval !== undefined) {
this.reconnectInterval = opts.reconnectInterval;
}
if (opts.requestTimeout !== undefined) {
this.requestTimeout = opts.requestTimeout;
}
}
getState(): ConnectionState {
return this.state;
}
// === Connection ===
/** Connect using REST API only (for OpenFang mode) */
async connectRest(): Promise<void> {
if (this.state === 'connected') {
return;
}
this.setState('connecting');
try {
// Check if OpenFang API is healthy
const health = await this.restGet<{ status: string; version?: string }>('/api/health');
if (health.status === 'ok') {
this.reconnectAttempts = 0;
this.setState('connected');
this.startHeartbeat(); // Start heartbeat after successful connection
this.log('info', `Connected to OpenFang via REST API${health.version ? ` (v${health.version})` : ''}`);
this.emitEvent('connected', { version: health.version });
} else {
throw new Error('Health check failed');
}
} catch (err: unknown) {
this.setState('disconnected');
const errorMessage = err instanceof Error ? err.message : String(err);
throw new Error(`Failed to connect to OpenFang: ${errorMessage}`);
}
}
connect(): Promise<void> {
if (this.state === 'connected' || this.state === 'connecting' || this.state === 'handshaking') {
return Promise.resolve();
}
// Check if URL is for OpenFang (port 50051) - use REST mode
if (this.url.includes(':50051')) {
return this.connectRest();
}
// Security warning: non-localhost with insecure WebSocket
if (!this.url.startsWith('wss://') && !isLocalhost(this.url)) {
console.warn('[Gateway] Connecting to non-localhost with insecure WebSocket (ws://). Consider using WSS in production.');
}
this.autoReconnect = true;
this.setState('connecting');
return new Promise((resolve, reject) => {
let settled = false;
const settleResolve = () => {
if (settled) return;
settled = true;
resolve();
};
const settleReject = (error: Error) => {
if (settled) return;
settled = true;
reject(error);
};
const handshakeTimer = window.setTimeout(() => {
this.log('error', `Handshake timed out after ${this.requestTimeout}ms`);
this.cleanup();
settleReject(new Error(`Gateway handshake timed out after ${this.requestTimeout}ms`));
}, this.requestTimeout);
try {
this.ws = new WebSocket(this.url);
this.ws.onopen = () => {
this.setState('handshaking');
};
this.ws.onmessage = (evt) => {
try {
const frame: GatewayFrame = JSON.parse(evt.data);
this.handleFrame(frame, () => {
clearTimeout(handshakeTimer);
settleResolve();
}, (error) => {
clearTimeout(handshakeTimer);
settleReject(error);
});
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
this.log('error', `Parse error: ${errorMessage}`);
}
};
this.ws.onclose = (evt) => {
const wasConnected = this.state === 'connected';
const closedDuringConnect = !wasConnected && !settled;
this.cleanup();
if (wasConnected && this.autoReconnect) {
this.scheduleReconnect();
}
this.emitEvent('close', { code: evt.code, reason: evt.reason });
if (closedDuringConnect) {
clearTimeout(handshakeTimer);
settleReject(new Error(evt.reason || `WebSocket closed before handshake completed (code: ${evt.code})`));
}
};
this.ws.onerror = () => {
if (this.state === 'connecting' || this.state === 'handshaking') {
clearTimeout(handshakeTimer);
this.cleanup();
settleReject(new Error('WebSocket connection failed'));
}
};
} catch (err) {
clearTimeout(handshakeTimer);
this.cleanup();
settleReject(err instanceof Error ? err : new Error(String(err)));
}
});
}
disconnect() {
this.autoReconnect = false;
this.cancelReconnect();
if (this.ws) {
this.ws.close(1000, 'Client disconnect');
}
this.cleanup();
}
// === Request/Response ===
async request(method: string, params?: Record<string, unknown>): Promise<unknown> {
if (this.state !== 'connected') {
throw new Error(`Not connected (state: ${this.state})`);
}
const id = `req_${++this.requestId}`;
const frame: GatewayRequest = { type: 'req', id, method, params };
return new Promise((resolve, reject) => {
const timer = window.setTimeout(() => {
this.pendingRequests.delete(id);
reject(new Error(`Request ${method} timed out`));
}, this.requestTimeout);
this.pendingRequests.set(id, { resolve, reject, timer });
this.send(frame);
});
}
// === High-level API ===
// Default agent ID for OpenFang (will be set dynamically from /api/agents)
private defaultAgentId: string = 'f77004c8-418f-4132-b7d4-7ecb9d66f44c';
/** Set the default agent ID */
setDefaultAgentId(agentId: string): void {
this.defaultAgentId = agentId;
this.log('info', `Default agent set to: ${agentId}`);
}
/** Get the current default agent ID */
getDefaultAgentId(): string {
return this.defaultAgentId;
}
/** Send message to agent (OpenFang chat API) */
async chat(message: string, opts?: {
sessionKey?: string;
agentId?: string;
idempotencyKey?: string;
extraSystemPrompt?: string;
model?: string;
temperature?: number;
maxTokens?: number;
}): Promise<{ runId: string; sessionId?: string; response?: string }> {
// OpenFang uses /api/agents/{agentId}/message endpoint
const agentId = opts?.agentId || this.defaultAgentId;
const result = await this.restPost<{ response?: string; input_tokens?: number; output_tokens?: number }>(`/api/agents/${agentId}/message`, {
message,
session_id: opts?.sessionKey,
});
// OpenFang returns { response, input_tokens, output_tokens }
return {
runId: createIdempotencyKey(),
sessionId: opts?.sessionKey,
response: result.response,
};
}
/** Send message with streaming response (OpenFang WebSocket) */
async chatStream(
message: string,
callbacks: {
onDelta: (delta: string) => void;
onTool?: (tool: string, input: string, output: string) => void;
onHand?: (name: string, status: string, result?: unknown) => void;
onComplete: () => void;
onError: (error: string) => void;
},
opts?: {
sessionKey?: string;
agentId?: string;
}
): Promise<{ runId: string }> {
const agentId = opts?.agentId || this.defaultAgentId;
const runId = createIdempotencyKey();
const sessionId = opts?.sessionKey || `session_${Date.now()}`;
// Store callbacks for this run
this.streamCallbacks.set(runId, callbacks);
// Connect to OpenFang WebSocket if not connected
this.connectOpenFangStream(agentId, runId, sessionId, message);
return { runId };
}
/** Connect to OpenFang streaming WebSocket */
private connectOpenFangStream(
agentId: string,
runId: string,
sessionId: string,
message: string
): void {
// Close existing connection if any
if (this.openfangWs && this.openfangWs.readyState !== WebSocket.CLOSED) {
this.openfangWs.close();
}
// Build WebSocket URL
// In dev mode, use Vite proxy; in production, use direct connection
let wsUrl: string;
if (typeof window !== 'undefined' && window.location.port === '1420') {
// Dev mode: use Vite proxy with relative path
wsUrl = `ws://${window.location.host}/api/agents/${agentId}/ws`;
} else {
// Production: extract from stored URL
const httpUrl = this.getRestBaseUrl();
wsUrl = httpUrl.replace(/^http/, 'ws') + `/api/agents/${agentId}/ws`;
}
this.log('info', `Connecting to OpenFang stream: ${wsUrl}`);
try {
this.openfangWs = new WebSocket(wsUrl);
this.openfangWs.onopen = () => {
this.log('info', 'OpenFang WebSocket connected');
// Send chat message using OpenFang actual protocol
const chatRequest = {
type: 'message',
content: message,
session_id: sessionId,
};
this.openfangWs?.send(JSON.stringify(chatRequest));
};
this.openfangWs.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
this.handleOpenFangStreamEvent(runId, data, sessionId);
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
this.log('error', `Failed to parse stream event: ${errorMessage}`);
}
};
this.openfangWs.onerror = (_event) => {
this.log('error', 'OpenFang WebSocket error');
const callbacks = this.streamCallbacks.get(runId);
if (callbacks) {
callbacks.onError('WebSocket connection failed');
this.streamCallbacks.delete(runId);
}
};
this.openfangWs.onclose = (event) => {
this.log('info', `OpenFang WebSocket closed: ${event.code} ${event.reason}`);
const callbacks = this.streamCallbacks.get(runId);
if (callbacks && event.code !== 1000) {
callbacks.onError(`Connection closed: ${event.reason || 'unknown'}`);
}
this.streamCallbacks.delete(runId);
this.openfangWs = null;
};
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
this.log('error', `Failed to create WebSocket: ${errorMessage}`);
const callbacks = this.streamCallbacks.get(runId);
if (callbacks) {
callbacks.onError(errorMessage);
this.streamCallbacks.delete(runId);
}
}
}
/** Handle OpenFang stream events */
private handleOpenFangStreamEvent(runId: string, data: OpenFangStreamEvent, sessionId: string): void {
const callbacks = this.streamCallbacks.get(runId);
if (!callbacks) return;
switch (data.type) {
// OpenFang actual event types
case 'text_delta':
// Stream delta content
if (data.content) {
callbacks.onDelta(data.content);
}
break;
case 'phase':
// Phase change: streaming | done
if (data.phase === 'done') {
callbacks.onComplete();
this.streamCallbacks.delete(runId);
if (this.openfangWs) {
this.openfangWs.close(1000, 'Stream complete');
}
}
break;
case 'response':
// Final response with tokens info
if (data.content) {
// If we haven't received any deltas yet, send the full response
// This handles non-streaming responses
}
// Mark complete if phase done wasn't sent
callbacks.onComplete();
this.streamCallbacks.delete(runId);
if (this.openfangWs) {
this.openfangWs.close(1000, 'Stream complete');
}
break;
case 'typing':
// Typing indicator: { state: 'start' | 'stop' }
// Can be used for UI feedback
break;
case 'tool_call':
// Tool call event
if (callbacks.onTool && data.tool) {
callbacks.onTool(data.tool, JSON.stringify(data.input || {}), data.output || '');
}
break;
case 'tool_result':
if (callbacks.onTool && data.tool) {
callbacks.onTool(data.tool, '', String(data.result || data.output || ''));
}
break;
case 'hand':
if (callbacks.onHand && data.hand_name) {
callbacks.onHand(data.hand_name, data.hand_status || 'triggered', data.hand_result);
}
break;
case 'error':
callbacks.onError(data.message || data.code || data.content || 'Unknown error');
this.streamCallbacks.delete(runId);
if (this.openfangWs) {
this.openfangWs.close(1011, 'Error');
}
break;
case 'connected':
// Connection established
this.log('info', `OpenFang agent connected: ${data.agent_id}`);
break;
case 'agents_updated':
// Agents list updated
this.log('debug', 'Agents list updated');
break;
default:
// Emit unknown events for debugging
this.log('debug', `Stream event: ${data.type}`);
}
// Also emit to general 'agent' event listeners
this.emitEvent('agent', {
stream: data.type === 'text_delta' ? 'assistant' : data.type,
delta: data.content,
content: data.content,
runId,
sessionId,
...data,
});
}
/** Cancel an ongoing stream */
cancelStream(runId: string): void {
const callbacks = this.streamCallbacks.get(runId);
if (callbacks) {
callbacks.onError('Stream cancelled');
this.streamCallbacks.delete(runId);
}
if (this.openfangWs && this.openfangWs.readyState === WebSocket.OPEN) {
this.openfangWs.close(1000, 'User cancelled');
}
}
/** Get Gateway health info */
async health(): Promise<any> {
return this.request('health');
}
/** Get Gateway status */
async status(): Promise<any> {
return this.request('status');
}
// === REST API Helpers (OpenFang) ===
private getRestBaseUrl(): string {
// In browser dev mode, use Vite proxy (empty string = relative path)
// In production Tauri, extract HTTP URL from WebSocket URL
if (typeof window !== 'undefined' && window.location.port === '1420') {
// Dev mode: use Vite proxy (requests go to /api/* which Vite proxies to backend)
return '';
}
// Production: extract HTTP URL from WebSocket URL
const wsUrl = this.url;
return wsUrl.replace(/^ws/, 'http').replace(/\/ws$/, '');
}
private async restGet<T>(path: string): Promise<T> {
const baseUrl = this.getRestBaseUrl();
const response = await fetch(`${baseUrl}${path}`);
if (!response.ok) {
// For 404 errors, throw with status code so callers can handle gracefully
const error = new Error(`REST API error: ${response.status} ${response.statusText}`);
(error as any).status = response.status;
throw error;
}
return response.json();
}
private async restPost<T>(path: string, body?: unknown): Promise<T> {
const baseUrl = this.getRestBaseUrl();
const response = await fetch(`${baseUrl}${path}`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: body ? JSON.stringify(body) : undefined,
});
if (!response.ok) {
throw new Error(`REST API error: ${response.status} ${response.statusText}`);
}
return response.json();
}
private async restPut<T>(path: string, body?: unknown): Promise<T> {
const baseUrl = this.getRestBaseUrl();
const response = await fetch(`${baseUrl}${path}`, {
method: 'PUT',
headers: { 'Content-Type': 'application/json' },
body: body ? JSON.stringify(body) : undefined,
});
if (!response.ok) {
throw new Error(`REST API error: ${response.status} ${response.statusText}`);
}
return response.json();
}
private async restDelete<T>(path: string): Promise<T> {
const baseUrl = this.getRestBaseUrl();
const response = await fetch(`${baseUrl}${path}`, {
method: 'DELETE',
});
if (!response.ok) {
throw new Error(`REST API error: ${response.status} ${response.statusText}`);
}
return response.json();
}
private async restPatch<T>(path: string, body?: unknown): Promise<T> {
const baseUrl = this.getRestBaseUrl();
const response = await fetch(`${baseUrl}${path}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: body ? JSON.stringify(body) : undefined,
});
if (!response.ok) {
throw new Error(`REST API error: ${response.status} ${response.statusText}`);
}
return response.json();
}
// === ZCLAW / Agent Methods (OpenFang REST API) ===
async listClones(): Promise<any> {
return this.restGet('/api/agents');
}
async createClone(opts: {
name: string;
role?: string;
nickname?: string;
scenarios?: string[];
model?: string;
workspaceDir?: string;
restrictFiles?: boolean;
privacyOptIn?: boolean;
userName?: string;
userRole?: string;
emoji?: string;
personality?: string;
communicationStyle?: string;
notes?: string;
}): Promise<any> {
// Build manifest_toml for OpenClaw Gateway
const lines: string[] = [];
lines.push(`name = "${opts.nickname || opts.name}"`);
lines.push(`model_provider = "bailian"`);
lines.push(`model_name = "${opts.model || 'qwen3.5-plus'}"`);
// Add identity section
lines.push('');
lines.push('[identity]');
if (opts.emoji) {
lines.push(`emoji = "${opts.emoji}"`);
}
if (opts.personality) {
lines.push(`personality = "${opts.personality}"`);
}
if (opts.communicationStyle) {
lines.push(`communication_style = "${opts.communicationStyle}"`);
}
// Add scenarios
if (opts.scenarios && opts.scenarios.length > 0) {
lines.push('');
lines.push('scenarios = [');
opts.scenarios.forEach((s, i) => {
lines.push(` "${s}"${i < opts.scenarios!.length - 1 ? ',' : ''}`);
});
lines.push(']');
}
// Add user context
if (opts.userName || opts.userRole) {
lines.push('');
lines.push('[user_context]');
if (opts.userName) {
lines.push(`name = "${opts.userName}"`);
}
if (opts.userRole) {
lines.push(`role = "${opts.userRole}"`);
}
}
const manifestToml = lines.join('\n');
return this.restPost('/api/agents', {
manifest_toml: manifestToml,
});
}
async updateClone(id: string, updates: Record<string, any>): Promise<any> {
return this.restPut(`/api/agents/${id}`, updates);
}
async deleteClone(id: string): Promise<any> {
return this.restDelete(`/api/agents/${id}`);
}
async getUsageStats(): Promise<any> {
try {
return await this.restGet('/api/stats/usage');
} catch (error) {
// Return structured fallback if API not available (404)
if (isNotFoundError(error)) {
return getUsageStatsFallback([]);
}
// Return minimal stats for other errors
return {
totalMessages: 0,
totalTokens: 0,
sessionsCount: 0,
agentsCount: 0,
};
}
}
async getSessionStats(): Promise<any> {
try {
return await this.restGet('/api/stats/sessions');
} catch {
return { sessions: [] };
}
}
async getWorkspaceInfo(): Promise<any> {
try {
return await this.restGet('/api/workspace');
} catch (error) {
// Return structured fallback if API not available (404)
if (isNotFoundError(error)) {
return getWorkspaceInfoFallback();
}
// Return minimal info for other errors
return {
rootDir: process.env.HOME || process.env.USERPROFILE || '~',
skillsDir: null,
handsDir: null,
configDir: null,
};
}
}
async getPluginStatus(): Promise<any> {
try {
return await this.restGet('/api/plugins/status');
} catch (error) {
// Return structured fallback if API not available (404)
if (isNotFoundError(error)) {
const plugins = getPluginStatusFallback([]);
return { plugins, loaded: plugins.length, total: plugins.length };
}
return { plugins: [], loaded: 0, total: 0 };
}
}
async getQuickConfig(): Promise<any> {
try {
// Use /api/config endpoint (OpenFang's actual config endpoint)
const config = await this.restGet('/api/config');
// Map OpenFang config to frontend expected format
return {
quickConfig: {
agentName: 'ZCLAW',
agentRole: 'AI 助手',
userName: '用户',
userRole: '用户',
agentNickname: 'ZCLAW',
scenarios: ['通用对话', '代码助手', '文档编写'],
workspaceDir: config.data_dir || config.home_dir,
gatewayUrl: this.baseUrl,
defaultModel: config.default_model?.model,
defaultProvider: config.default_model?.provider,
theme: 'dark',
showToolCalls: true,
autoSaveContext: true,
fileWatching: true,
privacyOptIn: false,
}
};
} catch (error) {
// Return structured fallback if API not available (404)
if (isNotFoundError(error)) {
return { quickConfig: getQuickConfigFallback() };
}
return {};
}
}
async saveQuickConfig(config: Record<string, any>): Promise<any> {
// Use /api/config endpoint for saving config
// Map frontend config back to OpenFang format
const openfangConfig = {
data_dir: config.workspaceDir,
default_model: config.defaultModel ? {
model: config.defaultModel,
provider: config.defaultProvider || 'bailian',
} : undefined,
};
return this.restPut('/api/config', openfangConfig);
}
async listSkills(): Promise<any> {
return this.restGet('/api/skills');
}
async getSkill(id: string): Promise<any> {
return this.restGet(`/api/skills/${id}`);
}
async createSkill(skill: {
name: string;
description?: string;
triggers: Array<{ type: string; pattern?: string }>;
actions: Array<{ type: string; params?: Record<string, unknown> }>;
enabled?: boolean;
}): Promise<any> {
return this.restPost('/api/skills', skill);
}
async updateSkill(id: string, updates: {
name?: string;
description?: string;
triggers?: Array<{ type: string; pattern?: string }>;
actions?: Array<{ type: string; params?: Record<string, unknown> }>;
enabled?: boolean;
}): Promise<any> {
return this.restPut(`/api/skills/${id}`, updates);
}
async deleteSkill(id: string): Promise<any> {
return this.restDelete(`/api/skills/${id}`);
}
async listChannels(): Promise<any> {
return this.restGet('/api/channels');
}
async getChannel(id: string): Promise<any> {
return this.restGet(`/api/channels/${id}`);
}
async createChannel(channel: {
type: string;
name: string;
config: Record<string, unknown>;
enabled?: boolean;
}): Promise<any> {
return this.restPost('/api/channels', channel);
}
async updateChannel(id: string, updates: {
name?: string;
config?: Record<string, unknown>;
enabled?: boolean;
}): Promise<any> {
return this.restPut(`/api/channels/${id}`, updates);
}
async deleteChannel(id: string): Promise<any> {
return this.restDelete(`/api/channels/${id}`);
}
async getFeishuStatus(): Promise<any> {
return this.restGet('/api/channels/feishu/status');
}
async listScheduledTasks(): Promise<any> {
try {
return await this.restGet('/api/scheduler/tasks');
} catch (error) {
// Return structured fallback if API not available (404)
if (isNotFoundError(error)) {
const tasks = getScheduledTasksFallback([]);
return { tasks, total: tasks.length };
}
// Return empty tasks list for other errors
return { tasks: [], total: 0 };
}
}
/** Create a scheduled task */
async createScheduledTask(task: {
name: string;
schedule: string;
scheduleType: 'cron' | 'interval' | 'once';
target?: {
type: 'agent' | 'hand' | 'workflow';
id: string;
};
description?: string;
enabled?: boolean;
}): Promise<{ id: string; name: string; schedule: string; status: string }> {
return this.restPost('/api/scheduler/tasks', task);
}
/** Delete a scheduled task */
async deleteScheduledTask(id: string): Promise<void> {
return this.restDelete(`/api/scheduler/tasks/${id}`);
}
/** Toggle a scheduled task (enable/disable) */
async toggleScheduledTask(id: string, enabled: boolean): Promise<{ id: string; enabled: boolean }> {
return this.restPatch(`/api/scheduler/tasks/${id}`, { enabled });
}
// === OpenFang Hands API ===
/** List available Hands */
async listHands(): Promise<{
hands: {
id?: string;
name: string;
description?: string;
status?: string;
requirements_met?: boolean;
category?: string;
icon?: string;
tool_count?: number;
tools?: string[];
metric_count?: number;
metrics?: string[];
}[]
}> {
return this.restGet('/api/hands');
}
/** Get Hand details */
async getHand(name: string): Promise<{
id?: string;
name?: string;
description?: string;
status?: string;
requirements_met?: boolean;
category?: string;
icon?: string;
provider?: string;
model?: string;
requirements?: { description?: string; name?: string; met?: boolean; satisfied?: boolean; details?: string; hint?: string }[];
tools?: string[];
metrics?: string[];
config?: Record<string, unknown>;
tool_count?: number;
metric_count?: number;
}> {
return this.restGet(`/api/hands/${name}`);
}
/** Trigger a Hand */
async triggerHand(name: string, params?: Record<string, unknown>): Promise<{ runId: string; status: string }> {
// OpenFang uses /activate endpoint, not /trigger
const result = await this.restPost<{
instance_id: string;
status: string;
}>(`/api/hands/${name}/activate`, params || {});
return { runId: result.instance_id, status: result.status };
}
/** Get Hand execution status */
async getHandStatus(name: string, runId: string): Promise<{ status: string; result?: unknown }> {
return this.restGet(`/api/hands/${name}/runs/${runId}`);
}
/** Approve a Hand execution (for needs_approval status) */
async approveHand(name: string, runId: string, approved: boolean, reason?: string): Promise<{ status: string }> {
return this.restPost(`/api/hands/${name}/runs/${runId}/approve`, { approved, reason });
}
/** Cancel a running Hand execution */
async cancelHand(name: string, runId: string): Promise<{ status: string }> {
return this.restPost(`/api/hands/${name}/runs/${runId}/cancel`, {});
}
/** List Hand execution runs */
async listHandRuns(name: string, opts?: { limit?: number; offset?: number }): Promise<{ runs: { runId: string; status: string; startedAt: string }[] }> {
const params = new URLSearchParams();
if (opts?.limit) params.set('limit', String(opts.limit));
if (opts?.offset) params.set('offset', String(opts.offset));
return this.restGet(`/api/hands/${name}/runs?${params}`);
}
// === OpenFang Workflows API ===
/** List available workflows */
async listWorkflows(): Promise<{ workflows: { id: string; name: string; steps: number }[] }> {
return this.restGet('/api/workflows');
}
/** Get workflow details */
async getWorkflow(id: string): Promise<{ id: string; name: string; steps: unknown[] }> {
return this.restGet(`/api/workflows/${id}`);
}
/** Execute a workflow */
async executeWorkflow(id: string, input?: Record<string, unknown>): Promise<{ runId: string; status: string }> {
return this.restPost(`/api/workflows/${id}/execute`, input);
}
/** Get workflow execution status */
async getWorkflowRun(workflowId: string, runId: string): Promise<{ status: string; step: string; result?: unknown }> {
return this.restGet(`/api/workflows/${workflowId}/runs/${runId}`);
}
/** List workflow execution runs */
async listWorkflowRuns(workflowId: string, opts?: { limit?: number; offset?: number }): Promise<{
runs: Array<{
runId: string;
status: string;
startedAt: string;
completedAt?: string;
step?: string;
result?: unknown;
error?: string;
}>;
}> {
const params = new URLSearchParams();
if (opts?.limit) params.set('limit', String(opts.limit));
if (opts?.offset) params.set('offset', String(opts.offset));
return this.restGet(`/api/workflows/${workflowId}/runs?${params}`);
}
/** Cancel a workflow execution */
async cancelWorkflow(workflowId: string, runId: string): Promise<{ status: string }> {
return this.restPost(`/api/workflows/${workflowId}/runs/${runId}/cancel`, {});
}
/** Create a new workflow */
async createWorkflow(workflow: {
name: string;
description?: string;
steps: Array<{
handName: string;
name?: string;
params?: Record<string, unknown>;
condition?: string;
}>;
}): Promise<{ id: string; name: string }> {
return this.restPost('/api/workflows', workflow);
}
/** Update a workflow */
async updateWorkflow(id: string, updates: {
name?: string;
description?: string;
steps?: Array<{
handName: string;
name?: string;
params?: Record<string, unknown>;
condition?: string;
}>;
}): Promise<{ id: string; name: string }> {
return this.restPut(`/api/workflows/${id}`, updates);
}
/** Delete a workflow */
async deleteWorkflow(id: string): Promise<{ status: string }> {
return this.restDelete(`/api/workflows/${id}`);
}
// === OpenFang Session API ===
/** List all sessions */
async listSessions(opts?: { limit?: number; offset?: number }): Promise<{
sessions: Array<{
id: string;
agent_id: string;
created_at: string;
updated_at?: string;
message_count?: number;
status?: 'active' | 'archived' | 'expired';
}>;
}> {
const params = new URLSearchParams();
if (opts?.limit) params.set('limit', String(opts.limit));
if (opts?.offset) params.set('offset', String(opts.offset));
return this.restGet(`/api/sessions?${params}`);
}
/** Get session details */
async getSession(sessionId: string): Promise<{
id: string;
agent_id: string;
created_at: string;
updated_at?: string;
message_count?: number;
status?: 'active' | 'archived' | 'expired';
metadata?: Record<string, unknown>;
}> {
return this.restGet(`/api/sessions/${sessionId}`);
}
/** Create a new session */
async createSession(opts: {
agent_id: string;
metadata?: Record<string, unknown>;
}): Promise<{
id: string;
agent_id: string;
created_at: string;
}> {
return this.restPost('/api/sessions', opts);
}
/** Delete a session */
async deleteSession(sessionId: string): Promise<{ status: string }> {
return this.restDelete(`/api/sessions/${sessionId}`);
}
/** Get session messages */
async getSessionMessages(sessionId: string, opts?: {
limit?: number;
offset?: number;
}): Promise<{
messages: Array<{
id: string;
role: 'user' | 'assistant' | 'system';
content: string;
created_at: string;
tokens?: { input?: number; output?: number };
}>;
}> {
const params = new URLSearchParams();
if (opts?.limit) params.set('limit', String(opts.limit));
if (opts?.offset) params.set('offset', String(opts.offset));
return this.restGet(`/api/sessions/${sessionId}/messages?${params}`);
}
// === OpenFang Triggers API ===
/** List triggers */
async listTriggers(): Promise<{ triggers: { id: string; type: string; enabled: boolean }[] }> {
return this.restGet('/api/triggers');
}
/** Get trigger details */
async getTrigger(id: string): Promise<{
id: string;
type: string;
name?: string;
enabled: boolean;
config?: Record<string, unknown>;
}> {
return this.restGet(`/api/triggers/${id}`);
}
/** Create a new trigger */
async createTrigger(trigger: {
type: string;
name?: string;
enabled?: boolean;
config?: Record<string, unknown>;
handName?: string;
workflowId?: string;
}): Promise<{ id: string }> {
return this.restPost('/api/triggers', trigger);
}
/** Update a trigger */
async updateTrigger(id: string, updates: {
name?: string;
enabled?: boolean;
config?: Record<string, unknown>;
handName?: string;
workflowId?: string;
}): Promise<{ id: string }> {
return this.restPut(`/api/triggers/${id}`, updates);
}
/** Delete a trigger */
async deleteTrigger(id: string): Promise<{ status: string }> {
return this.restDelete(`/api/triggers/${id}`);
}
// === OpenFang Audit API ===
/** Get audit logs */
async getAuditLogs(opts?: { limit?: number; offset?: number }): Promise<{ logs: unknown[] }> {
const params = new URLSearchParams();
if (opts?.limit) params.set('limit', String(opts.limit));
if (opts?.offset) params.set('offset', String(opts.offset));
return this.restGet(`/api/audit/logs?${params}`);
}
/** Verify audit log chain for a specific log entry */
async verifyAuditLogChain(logId: string): Promise<{
valid: boolean;
chain_depth?: number;
root_hash?: string;
broken_at_index?: number;
}> {
return this.restGet(`/api/audit/verify/${logId}`);
}
// === OpenFang Security API ===
/** Get security status */
async getSecurityStatus(): Promise<{ layers: { name: string; enabled: boolean }[] }> {
try {
return await this.restGet('/api/security/status');
} catch (error) {
// Return structured fallback if API not available (404)
if (isNotFoundError(error)) {
const status = getSecurityStatusFallback();
return { layers: status.layers };
}
// Return minimal security layers for other errors
return {
layers: [
{ name: 'device_auth', enabled: true },
{ name: 'rbac', enabled: true },
{ name: 'audit_log', enabled: true },
],
};
}
}
/** Get capabilities (RBAC) */
async getCapabilities(): Promise<{ capabilities: string[] }> {
try {
return await this.restGet('/api/capabilities');
} catch {
return { capabilities: ['chat', 'agents', 'hands', 'workflows'] };
}
}
// === OpenFang Approvals API ===
/** List pending/approved/rejected approvals */
async listApprovals(status?: string): Promise<{
approvals: {
id: string;
hand_name: string;
run_id: string;
status: string;
requested_at: string;
requested_by?: string;
reason?: string;
action?: string;
params?: Record<string, unknown>;
responded_at?: string;
responded_by?: string;
response_reason?: string;
}[];
}> {
const params = status ? `?status=${status}` : '';
return this.restGet(`/api/approvals${params}`);
}
/** Respond to an approval (approve/reject) */
async respondToApproval(approvalId: string, approved: boolean, reason?: string): Promise<{ status: string }> {
return this.restPost(`/api/approvals/${approvalId}/respond`, { approved, reason });
}
async listModels(): Promise<{ models: GatewayModelChoice[] }> {
// OpenFang: 使用 REST API
return this.restGet('/api/models');
}
async getConfig(): Promise<GatewayConfigSnapshot | Record<string, any>> {
// OpenFang: 使用 REST API
return this.restGet('/api/config');
}
async applyConfig(raw: string, baseHash?: string, opts?: { sessionKey?: string; note?: string; restartDelayMs?: number }): Promise<any> {
return this.request('config.apply', {
raw,
baseHash,
sessionKey: opts?.sessionKey,
note: opts?.note,
restartDelayMs: opts?.restartDelayMs,
});
}
// === Event Subscription ===
/** Subscribe to a Gateway event (e.g., 'agent', 'chat', 'heartbeat') */
on(event: string, callback: EventCallback): () => void {
if (!this.eventListeners.has(event)) {
this.eventListeners.set(event, new Set());
}
this.eventListeners.get(event)!.add(callback);
// Return unsubscribe function
return () => {
this.eventListeners.get(event)?.delete(callback);
};
}
/** Subscribe to agent stream events */
onAgentStream(callback: (delta: AgentStreamDelta) => void): () => void {
return this.on('agent', (payload: unknown) => {
callback(payload as AgentStreamDelta);
});
}
// === Internal ===
private handleFrame(frame: GatewayFrame, connectResolve?: () => void, connectReject?: (error: Error) => void) {
// Handle pong responses for heartbeat
if (frame.type === 'pong') {
this.handlePong();
return;
}
if (frame.type === 'event') {
this.handleEvent(frame, connectResolve, connectReject);
} else if (frame.type === 'res') {
this.handleResponse(frame);
}
}
private handleEvent(event: GatewayEvent, connectResolve?: () => void, connectReject?: (error: Error) => void) {
// Handle connect challenge
if (event.event === 'connect.challenge' && this.state === 'handshaking') {
const payload = event.payload as { nonce?: string } | undefined;
this.performHandshake(payload?.nonce || '', connectResolve, connectReject);
return;
}
// Dispatch to listeners
this.emitEvent(event.event, event.payload);
}
private async performHandshake(challengeNonce: string | undefined, connectResolve?: () => void, connectReject?: (error: Error) => void) {
if (!challengeNonce) {
this.log('error', 'No challenge nonce received');
connectReject?.(new Error('Handshake failed: no challenge nonce'));
return;
}
const connectId = `connect_${Date.now()}`;
// Use a valid client ID from GATEWAY_CLIENT_ID_SET
// Valid IDs: gateway-client, cli, webchat, node-host, test
// 'cli' is for control UI / command-line clients
const clientId = 'cli';
// Valid modes: cli, webchat, backend, node
// 'cli' is for command-line/Control UI clients
const clientMode = 'cli';
const role = 'operator';
const scopes = ['operator.read', 'operator.write', 'operator.admin', 'operator.approvals', 'operator.pairing'];
// Debug: log token status
this.log('debug', `Handshake token: ${this.token ? `${this.token.substring(0, 8)}... (${this.token.length} chars)` : '(empty)'}`);
try {
const deviceKeys = await this.deviceKeysPromise;
// Debug: log device auth details
this.log('debug', `Device auth: deviceId=${deviceKeys.deviceId.substring(0, 8)}..., nonce=${challengeNonce.substring(0, 8)}...`);
const { signature, signedAt } = signDeviceAuth({
clientId,
clientMode,
deviceId: deviceKeys.deviceId,
nonce: challengeNonce,
role,
scopes,
secretKey: deviceKeys.secretKey,
token: this.token,
});
// Debug: log signature details
this.log('debug', `Signature created: signedAt=${signedAt}, sig=${signature.substring(0, 16)}...`);
const connectReq: GatewayRequest = {
type: 'req',
id: connectId,
method: 'connect',
params: {
minProtocol: 3,
maxProtocol: 3,
client: {
id: clientId,
version: '0.2.0',
platform: this.detectPlatform(),
mode: clientMode,
},
role,
scopes,
auth: this.token ? { token: this.token } : {},
locale: 'zh-CN',
userAgent: 'zclaw-tauri/0.2.0',
device: {
id: deviceKeys.deviceId,
publicKey: deviceKeys.publicKeyBase64,
signature,
signedAt,
nonce: challengeNonce,
},
},
};
const originalHandler = this.ws!.onmessage;
this.ws!.onmessage = (evt) => {
try {
const frame = JSON.parse(evt.data);
if (frame.type === 'res' && frame.id === connectId) {
this.ws!.onmessage = originalHandler;
if (frame.ok) {
this.setState('connected');
this.reconnectAttempts = 0;
this.startHeartbeat(); // Start heartbeat after successful connection
this.emitEvent('connected', frame.payload);
this.log('info', 'Connected to Gateway');
connectResolve?.();
} else {
const errorObj = frame.error;
const errorMessage = errorObj?.message || errorObj?.code || JSON.stringify(errorObj);
const error = new Error(`Handshake failed: ${errorMessage}`);
this.log('error', error.message);
// Check for signature-related errors and clear device keys if needed
if (errorMessage.includes('signature') || errorMessage.includes('device')) {
this.log('warn', 'Device signature failed, clearing cached keys for retry');
clearDeviceKeys();
}
this.cleanup();
connectReject?.(error);
}
} else {
originalHandler?.call(this.ws!, evt);
}
} catch {
// Ignore parse errors
}
};
this.send(connectReq);
} catch (err: unknown) {
const error = err instanceof Error ? err : new Error(String(err));
this.log('error', error.message);
this.cleanup();
connectReject?.(error);
}
}
private handleResponse(res: GatewayResponse) {
const pending = this.pendingRequests.get(res.id);
if (pending) {
clearTimeout(pending.timer);
this.pendingRequests.delete(res.id);
if (res.ok) {
pending.resolve(res.payload);
} else {
pending.reject(new Error(JSON.stringify(res.error)));
}
}
}
private send(frame: GatewayFrame) {
if (this.ws?.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(frame));
}
}
private emitEvent(event: string, payload: unknown) {
const listeners = this.eventListeners.get(event);
if (listeners) {
for (const cb of listeners) {
try { cb(payload); } catch { /* ignore listener errors */ }
}
}
// Also emit wildcard
const wildcardListeners = this.eventListeners.get('*');
if (wildcardListeners) {
for (const cb of wildcardListeners) {
try { cb({ event, payload }); } catch { /* ignore */ }
}
}
}
private setState(state: ConnectionState) {
this.state = state;
this.onStateChange?.(state);
this.emitEvent('state', state);
}
private cleanup() {
// Stop heartbeat on cleanup
this.stopHeartbeat();
for (const [, pending] of this.pendingRequests) {
clearTimeout(pending.timer);
pending.reject(new Error('Connection closed'));
}
this.pendingRequests.clear();
if (this.ws) {
this.ws.onopen = null;
this.ws.onmessage = null;
this.ws.onclose = null;
this.ws.onerror = null;
if (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING) {
try { this.ws.close(); } catch { /* ignore */ }
}
this.ws = null;
}
this.setState('disconnected');
}
// === Heartbeat Methods ===
/**
* Start heartbeat to keep connection alive.
* Called after successful connection.
*/
private startHeartbeat(): void {
this.stopHeartbeat();
this.missedHeartbeats = 0;
this.heartbeatInterval = window.setInterval(() => {
this.sendHeartbeat();
}, GatewayClient.HEARTBEAT_INTERVAL);
this.log('debug', 'Heartbeat started');
}
/**
* Stop heartbeat.
* Called on cleanup or disconnect.
*/
private stopHeartbeat(): void {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
if (this.heartbeatTimeout) {
clearTimeout(this.heartbeatTimeout);
this.heartbeatTimeout = null;
}
this.log('debug', 'Heartbeat stopped');
}
/**
* Send a ping heartbeat to the server.
*/
private sendHeartbeat(): void {
if (this.ws?.readyState !== WebSocket.OPEN) {
this.log('debug', 'Skipping heartbeat - WebSocket not open');
return;
}
this.missedHeartbeats++;
if (this.missedHeartbeats > GatewayClient.MAX_MISSED_HEARTBEATS) {
this.log('warn', `Max missed heartbeats (${GatewayClient.MAX_MISSED_HEARTBEATS}), reconnecting`);
this.stopHeartbeat();
this.ws.close(4000, 'Heartbeat timeout');
return;
}
// Send ping frame
try {
this.ws.send(JSON.stringify({ type: 'ping' }));
this.log('debug', `Ping sent (missed: ${this.missedHeartbeats})`);
// Set timeout for pong
this.heartbeatTimeout = window.setTimeout(() => {
this.log('warn', 'Heartbeat pong timeout');
// Don't reconnect immediately, let the next heartbeat check
}, GatewayClient.HEARTBEAT_TIMEOUT);
} catch (error) {
this.log('error', `Failed to send heartbeat: ${error instanceof Error ? error.message : String(error)}`);
}
}
/**
* Handle pong response from server.
*/
private handlePong(): void {
this.missedHeartbeats = 0;
if (this.heartbeatTimeout) {
clearTimeout(this.heartbeatTimeout);
this.heartbeatTimeout = null;
}
this.log('debug', 'Pong received, heartbeat reset');
}
private static readonly MAX_RECONNECT_ATTEMPTS = 10;
private scheduleReconnect() {
if (this.reconnectAttempts >= GatewayClient.MAX_RECONNECT_ATTEMPTS) {
this.log('error', `Max reconnect attempts (${GatewayClient.MAX_RECONNECT_ATTEMPTS}) reached. Please reconnect manually.`);
this.setState('disconnected');
this.emitEvent('reconnect_failed', {
attempts: this.reconnectAttempts,
maxAttempts: GatewayClient.MAX_RECONNECT_ATTEMPTS
});
return;
}
this.reconnectAttempts++;
this.setState('reconnecting');
const delay = Math.min(this.reconnectInterval * Math.pow(1.5, this.reconnectAttempts - 1), 30000);
this.log('info', `Scheduling reconnect attempt ${this.reconnectAttempts} in ${delay}ms`);
// Emit reconnecting event for UI
this.emitEvent('reconnecting', {
attempt: this.reconnectAttempts,
delay,
maxAttempts: GatewayClient.MAX_RECONNECT_ATTEMPTS
});
this.reconnectTimer = window.setTimeout(async () => {
try {
await this.connect();
} catch {
/* close handler will trigger another reconnect */
this.log('warn', `Reconnect attempt ${this.reconnectAttempts} failed`);
}
}, delay);
}
private cancelReconnect() {
if (this.reconnectTimer !== null) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
}
private detectPlatform(): string {
const ua = navigator.userAgent.toLowerCase();
if (ua.includes('win')) return 'windows';
if (ua.includes('mac')) return 'macos';
return 'linux';
}
private log(level: string, message: string) {
this.onLog?.(level, message);
}
}
// Singleton instance
let _client: GatewayClient | null = null;
export function getGatewayClient(opts?: ConstructorParameters<typeof GatewayClient>[0]): GatewayClient {
if (!_client) {
_client = new GatewayClient(opts);
} else if (opts) {
_client.updateOptions(opts);
}
return _client;
}