581 lines
18 KiB
TypeScript
581 lines
18 KiB
TypeScript
/**
|
|
* Web channel gateway — reverse WebSocket connection to relay.
|
|
*
|
|
* Follows the Telegram monitorTelegramProvider pattern:
|
|
* - Connects to relay as "openclaw" role
|
|
* - Listens for browser messages → dispatches AI reply
|
|
* - Auto-reconnects with exponential backoff
|
|
* - Responds to AbortSignal for graceful shutdown
|
|
* - Detects zombie connections via pong timeout
|
|
*/
|
|
|
|
import WebSocket from 'ws'
|
|
import type {
|
|
ChannelAccountSnapshot,
|
|
ChannelGatewayAdapter,
|
|
ChannelGatewayContext,
|
|
} from 'openclaw/plugin-sdk'
|
|
import {
|
|
type ChatMessage,
|
|
generateMessageId,
|
|
HEARTBEAT_INTERVAL_MS,
|
|
parseRelayMessage,
|
|
type SessionEventMessage,
|
|
type SessionQueueMode,
|
|
serializeRelayMessage,
|
|
} from './protocol.js'
|
|
import type { ResolvedWebAccount } from './types.js'
|
|
|
|
const MAX_RECONNECT_ATTEMPTS = 10
|
|
const INITIAL_RECONNECT_MS = 2_000
|
|
const MAX_RECONNECT_MS = 60_000
|
|
const BACKOFF_FACTOR = 2
|
|
const PONG_TIMEOUT_MS = HEARTBEAT_INTERVAL_MS * 2.5 // 75s — allow 2 missed pongs
|
|
const COOLDOWN_MS = 5 * 60_000 // 5 min cooldown after max reconnect attempts
|
|
|
|
/** Per-account gateway WebSocket references for outbound message sending. */
|
|
const activeWsMap = new Map<string, WebSocket>()
|
|
const sessionActiveDispatches = new Map<string, number>()
|
|
const sessionQueuedDispatches = new Map<string, number>()
|
|
|
|
const KNOWN_QUEUE_MODES = new Set<SessionQueueMode>([
|
|
'collect',
|
|
'followup',
|
|
'steer',
|
|
'interrupt',
|
|
'steer-backlog',
|
|
])
|
|
|
|
export function getActiveWebSocket(accountId?: string): WebSocket | null {
|
|
if (accountId) return activeWsMap.get(accountId) ?? null
|
|
// Fallback: return the first connected one (backward compat for single-account)
|
|
for (const ws of activeWsMap.values()) {
|
|
if (ws.readyState === WebSocket.OPEN) return ws
|
|
}
|
|
return null
|
|
}
|
|
|
|
export async function startWebGateway(
|
|
ctx: ChannelGatewayContext<ResolvedWebAccount>,
|
|
): Promise<void> {
|
|
const { account, abortSignal, log } = ctx
|
|
|
|
let reconnectAttempts = 0
|
|
let reconnectMs = INITIAL_RECONNECT_MS
|
|
let reconnectTimer: ReturnType<typeof setTimeout> | null = null // [R3] track timer for abort cleanup
|
|
|
|
const connect = (): Promise<void> =>
|
|
new Promise<void>((resolve, reject) => {
|
|
if (abortSignal.aborted) {
|
|
resolve()
|
|
return
|
|
}
|
|
|
|
const baseUrl = account.relayUrl.replace(/\/+$/, '')
|
|
const wsUrl = baseUrl.endsWith('/openclaw')
|
|
? baseUrl
|
|
: `${baseUrl}/openclaw`
|
|
log?.info(`[web:${account.accountId}] connecting to relay: ${wsUrl}`)
|
|
|
|
const ws = new WebSocket(wsUrl)
|
|
let heartbeatTimer: ReturnType<typeof setInterval> | null = null
|
|
let authenticated = false
|
|
let lastPongAt = Date.now() // [H1] track last pong for zombie detection
|
|
|
|
const cleanup = () => {
|
|
activeWsMap.delete(account.accountId)
|
|
if (heartbeatTimer) {
|
|
clearInterval(heartbeatTimer)
|
|
heartbeatTimer = null
|
|
}
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
connected: false,
|
|
running: true,
|
|
lastStopAt: Date.now(),
|
|
} satisfies ChannelAccountSnapshot)
|
|
}
|
|
|
|
ws.on('open', () => {
|
|
log?.info(`[web:${account.accountId}] connected, authenticating...`)
|
|
|
|
// Send auth handshake
|
|
ws.send(
|
|
serializeRelayMessage({
|
|
type: 'auth',
|
|
role: 'openclaw',
|
|
token: account.authToken,
|
|
}),
|
|
)
|
|
})
|
|
|
|
// [E1] Wrap entire message handler in void async IIFE with try/catch
|
|
// to prevent unhandled promise rejections from crashing the process
|
|
ws.on('message', (data) => {
|
|
void (async () => {
|
|
try {
|
|
const raw = typeof data === 'string' ? data : data.toString('utf-8')
|
|
const msg = parseRelayMessage(raw)
|
|
if (!msg) return
|
|
|
|
switch (msg.type) {
|
|
case 'auth_result': {
|
|
if (msg.ok) {
|
|
log?.info(`[web:${account.accountId}] authenticated successfully`)
|
|
authenticated = true
|
|
activeWsMap.set(account.accountId, ws)
|
|
reconnectAttempts = 0
|
|
reconnectMs = INITIAL_RECONNECT_MS
|
|
lastPongAt = Date.now() // [H1] reset pong timer on fresh connection
|
|
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
connected: true,
|
|
running: true,
|
|
lastStartAt: Date.now(),
|
|
lastError: null,
|
|
restartPending: false, // [M2] clear restartPending on successful reconnect
|
|
} satisfies ChannelAccountSnapshot)
|
|
|
|
// [H1] Start heartbeat with pong timeout detection
|
|
heartbeatTimer = setInterval(() => {
|
|
if (ws.readyState !== WebSocket.OPEN) return
|
|
|
|
// Check if pong timed out — zombie connection detection
|
|
const elapsed = Date.now() - lastPongAt
|
|
if (elapsed > PONG_TIMEOUT_MS) {
|
|
log?.error(
|
|
`[web:${account.accountId}] pong timeout (${elapsed}ms elapsed, limit ${PONG_TIMEOUT_MS}ms), forcing reconnect`,
|
|
)
|
|
ws.close(4000, 'Pong timeout')
|
|
return
|
|
}
|
|
|
|
ws.send(
|
|
serializeRelayMessage({
|
|
type: 'ping',
|
|
timestamp: Date.now(),
|
|
}),
|
|
)
|
|
}, HEARTBEAT_INTERVAL_MS)
|
|
} else {
|
|
log?.error(`[web:${account.accountId}] auth failed: ${msg.error}`)
|
|
ws.close(4001, 'Authentication failed')
|
|
}
|
|
break
|
|
}
|
|
|
|
case 'message': {
|
|
if (!authenticated) break
|
|
// Incoming message from browser → dispatch to AI
|
|
await handleIncomingMessage(ctx, msg as ChatMessage)
|
|
break
|
|
}
|
|
|
|
case 'typing': {
|
|
// Browser typing indicator — can log or pass through
|
|
break
|
|
}
|
|
|
|
case 'pong': {
|
|
// [H1] Update last pong timestamp for zombie detection
|
|
lastPongAt = Date.now()
|
|
break
|
|
}
|
|
|
|
case 'user_status': {
|
|
// Browser user online/offline notification
|
|
log?.info(
|
|
`[web:${account.accountId}] user_status: ${msg.action} userId=${msg.userId} sessionId=${msg.sessionId}${msg.userName ? ` name=${msg.userName}` : ''}`,
|
|
)
|
|
break
|
|
}
|
|
|
|
default:
|
|
break
|
|
}
|
|
} catch (err) {
|
|
// [E1] Catch all async errors to prevent unhandled promise rejection
|
|
log?.error(
|
|
`[web:${account.accountId}] message handler error: ${err instanceof Error ? err.message : String(err)}`,
|
|
)
|
|
}
|
|
})()
|
|
})
|
|
|
|
ws.on('close', (code, reason) => {
|
|
log?.warn(
|
|
`[web:${account.accountId}] disconnected (code=${code}, reason=${reason.toString('utf-8')})`,
|
|
)
|
|
cleanup()
|
|
|
|
if (abortSignal.aborted) {
|
|
resolve()
|
|
return
|
|
}
|
|
|
|
// [R1] Authentication failure is unrecoverable — do not reconnect
|
|
if (code === 4001) {
|
|
const errMessage = 'Authentication failed — not reconnecting'
|
|
log?.error(`[web:${account.accountId}] ${errMessage}`)
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
lastError: errMessage,
|
|
running: false,
|
|
} satisfies ChannelAccountSnapshot)
|
|
reject(new Error(errMessage))
|
|
return
|
|
}
|
|
|
|
// Schedule reconnect
|
|
if (reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
|
|
reconnectAttempts++
|
|
const jitter = Math.random() * 0.2 * reconnectMs
|
|
const delay = reconnectMs + jitter
|
|
log?.info(
|
|
`[web:${account.accountId}] reconnecting in ${Math.round(delay)}ms (attempt ${reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS})`,
|
|
)
|
|
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
restartPending: true,
|
|
reconnectAttempts,
|
|
} satisfies ChannelAccountSnapshot)
|
|
|
|
reconnectTimer = setTimeout(() => {
|
|
reconnectTimer = null
|
|
reconnectMs = Math.min(
|
|
reconnectMs * BACKOFF_FACTOR,
|
|
MAX_RECONNECT_MS,
|
|
)
|
|
connect().then(resolve, reject)
|
|
}, delay)
|
|
} else {
|
|
// [R2] Max attempts reached — cooldown then retry instead of permanent failure
|
|
log?.warn(
|
|
`[web:${account.accountId}] max reconnect attempts (${MAX_RECONNECT_ATTEMPTS}) reached, cooling down for ${COOLDOWN_MS / 1000}s`,
|
|
)
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
lastError: `Max reconnect attempts reached, cooling down for ${COOLDOWN_MS / 1000}s`,
|
|
restartPending: true,
|
|
} satisfies ChannelAccountSnapshot)
|
|
|
|
reconnectTimer = setTimeout(() => {
|
|
reconnectTimer = null
|
|
if (abortSignal.aborted) {
|
|
resolve()
|
|
return
|
|
}
|
|
log?.info(`[web:${account.accountId}] cooldown complete, resetting reconnect counter`)
|
|
reconnectAttempts = 0
|
|
reconnectMs = INITIAL_RECONNECT_MS
|
|
connect().then(resolve, reject)
|
|
}, COOLDOWN_MS)
|
|
}
|
|
})
|
|
|
|
// [M1] Update lastError status in error handler
|
|
ws.on('error', err => {
|
|
log?.error(`[web:${account.accountId}] WebSocket error: ${err.message}`)
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
lastError: err.message,
|
|
} satisfies ChannelAccountSnapshot)
|
|
// Error triggers close event, so reconnect logic runs there
|
|
})
|
|
|
|
// Listen for abort signal
|
|
const onAbort = () => {
|
|
log?.info(`[web:${account.accountId}] shutting down (abort signal)`)
|
|
cleanup()
|
|
// [R3] Clear pending reconnect timer on abort
|
|
if (reconnectTimer) {
|
|
clearTimeout(reconnectTimer)
|
|
reconnectTimer = null
|
|
}
|
|
if (
|
|
ws.readyState === WebSocket.OPEN ||
|
|
ws.readyState === WebSocket.CONNECTING
|
|
) {
|
|
ws.close(1000, 'Shutting down')
|
|
}
|
|
resolve()
|
|
}
|
|
|
|
abortSignal.addEventListener('abort', onAbort, { once: true })
|
|
})
|
|
|
|
return connect()
|
|
}
|
|
|
|
// ── Inbound Message Handling ──────────────────────────────────────
|
|
|
|
async function handleIncomingMessage(
|
|
ctx: ChannelGatewayContext<ResolvedWebAccount>,
|
|
msg: ChatMessage,
|
|
): Promise<void> {
|
|
const { log } = ctx
|
|
const sessionKey = buildSessionStateKey(ctx.accountId, msg.sessionId)
|
|
const queueMode = resolveQueueMode(ctx)
|
|
const activeCount = sessionActiveDispatches.get(sessionKey) ?? 0
|
|
const isBusy = activeCount > 0
|
|
const nextQueuedCount = isBusy
|
|
? (sessionQueuedDispatches.get(sessionKey) ?? 0) + 1
|
|
: 0
|
|
|
|
if (isBusy) {
|
|
sessionQueuedDispatches.set(sessionKey, nextQueuedCount)
|
|
sendSessionEventToSession(
|
|
msg.sessionId,
|
|
{
|
|
event: 'queue_status',
|
|
state: 'enqueued',
|
|
mode: queueMode,
|
|
depth: nextQueuedCount,
|
|
messageId: msg.id,
|
|
},
|
|
ctx.accountId,
|
|
)
|
|
} else {
|
|
sendSessionEventToSession(
|
|
msg.sessionId,
|
|
{
|
|
event: 'queue_status',
|
|
state: 'processing',
|
|
mode: queueMode,
|
|
depth: 0,
|
|
messageId: msg.id,
|
|
},
|
|
ctx.accountId,
|
|
)
|
|
}
|
|
|
|
sessionActiveDispatches.set(sessionKey, activeCount + 1)
|
|
|
|
log?.info(
|
|
`[web:${ctx.account.accountId}] message from session ${msg.sessionId}: ${msg.text.slice(0, 80)}...`,
|
|
)
|
|
|
|
ctx.setStatus({
|
|
...ctx.getStatus(),
|
|
lastInboundAt: Date.now(),
|
|
} satisfies ChannelAccountSnapshot)
|
|
|
|
try {
|
|
const core = ctx.channelRuntime
|
|
if (!core) {
|
|
log?.warn(
|
|
`[web:${ctx.account.accountId}] channelRuntime not available — cannot dispatch AI reply`,
|
|
)
|
|
// 版本太低 openclaw 版本太低去升级吧
|
|
void sendTextToSession(
|
|
msg.sessionId,
|
|
'AI is not available at the moment. Please try again later.',
|
|
ctx.accountId,
|
|
)
|
|
return
|
|
}
|
|
|
|
// 1. Resolve agent routing
|
|
const route = core.routing.resolveAgentRoute({
|
|
cfg: ctx.cfg,
|
|
channel: 'clawshow',
|
|
accountId: ctx.accountId,
|
|
peer: { kind: 'direct', id: msg.sessionId },
|
|
})
|
|
|
|
log?.info(
|
|
`[web:${ctx.account.accountId}] route: agent=${route.agentId} session=${route.sessionKey}`,
|
|
)
|
|
|
|
// 2. Build MsgContext with PascalCase fields (will be finalized by dispatchInboundMessageWithBufferedDispatcher)
|
|
const ctxPayload = {
|
|
Body: msg.text,
|
|
BodyForAgent: msg.text,
|
|
RawBody: msg.text,
|
|
CommandBody: msg.text,
|
|
From: `web:${msg.sessionId}`,
|
|
To: msg.sessionId,
|
|
SessionKey: route.sessionKey,
|
|
AccountId: ctx.accountId,
|
|
ChatType: 'direct' as const,
|
|
Provider: 'clawshow',
|
|
Surface: 'clawshow',
|
|
MessageSid: msg.id,
|
|
Timestamp: msg.timestamp || Date.now(),
|
|
CommandAuthorized: false,
|
|
}
|
|
|
|
// 3. Use the high-level dispatch API that handles all lifecycle internally
|
|
sendTypingToSession(msg.sessionId, true, ctx.accountId)
|
|
const result = await core.reply.dispatchReplyWithBufferedBlockDispatcher({
|
|
ctx: ctxPayload,
|
|
cfg: ctx.cfg,
|
|
dispatcherOptions: {
|
|
deliver: async (payload: { text?: string }, _info: { kind: string }) => {
|
|
if (payload.text) {
|
|
void sendTextToSession(msg.sessionId, payload.text, ctx.accountId)
|
|
}
|
|
},
|
|
onError: (err: unknown, info: { kind: string }) => {
|
|
log?.error(
|
|
`[web:${ctx.account.accountId}] reply delivery error (${info.kind}): ${err}`,
|
|
)
|
|
},
|
|
},
|
|
})
|
|
|
|
sendTypingToSession(msg.sessionId, false, ctx.accountId)
|
|
|
|
log?.info(
|
|
`[web:${ctx.account.accountId}] dispatch complete (queuedFinal=${result.queuedFinal}, replies=${result.counts.final})`,
|
|
)
|
|
} catch (err) {
|
|
sendTypingToSession(msg.sessionId, false, ctx.accountId)
|
|
// [M3] Log full stack trace but only send generic error to user
|
|
const fullMsg =
|
|
err instanceof Error ? `${err.message}\n${err.stack}` : String(err)
|
|
log?.error(`[web:${ctx.account.accountId}] dispatch error: ${fullMsg}`)
|
|
void sendTextToSession(
|
|
msg.sessionId,
|
|
`Sorry, an error occurred while processing your message. Please try again.`,
|
|
ctx.accountId,
|
|
)
|
|
} finally {
|
|
const remainingActive = Math.max(
|
|
0,
|
|
(sessionActiveDispatches.get(sessionKey) ?? 1) - 1,
|
|
)
|
|
if (remainingActive > 0) {
|
|
sessionActiveDispatches.set(sessionKey, remainingActive)
|
|
} else {
|
|
sessionActiveDispatches.delete(sessionKey)
|
|
}
|
|
|
|
if (isBusy) {
|
|
const remainingQueued = Math.max(
|
|
0,
|
|
(sessionQueuedDispatches.get(sessionKey) ?? 1) - 1,
|
|
)
|
|
if (remainingQueued > 0) {
|
|
sessionQueuedDispatches.set(sessionKey, remainingQueued)
|
|
} else {
|
|
sessionQueuedDispatches.delete(sessionKey)
|
|
}
|
|
}
|
|
|
|
if (remainingActive === 0) {
|
|
sendSessionEventToSession(
|
|
msg.sessionId,
|
|
{
|
|
event: 'queue_status',
|
|
state: 'idle',
|
|
mode: queueMode,
|
|
depth: sessionQueuedDispatches.get(sessionKey) ?? 0,
|
|
messageId: msg.id,
|
|
},
|
|
ctx.accountId,
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Outbound Helpers ────────────────────────────────────────────────
|
|
|
|
// [E2] Try/catch around ws.send to handle race between readyState check and send
|
|
export function sendTextToSession(
|
|
sessionId: string,
|
|
text: string,
|
|
accountId?: string,
|
|
): boolean {
|
|
const ws = getActiveWebSocket(accountId)
|
|
if (!ws || ws.readyState !== WebSocket.OPEN) return false
|
|
|
|
try {
|
|
const msg: ChatMessage = {
|
|
type: 'message',
|
|
id: generateMessageId(),
|
|
sessionId,
|
|
from: 'ai',
|
|
text,
|
|
timestamp: Date.now(),
|
|
}
|
|
|
|
ws.send(serializeRelayMessage(msg))
|
|
return true
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
export function sendTypingToSession(
|
|
sessionId: string,
|
|
isTyping: boolean,
|
|
accountId?: string,
|
|
): boolean {
|
|
const ws = getActiveWebSocket(accountId)
|
|
if (!ws || ws.readyState !== WebSocket.OPEN) return false
|
|
|
|
try {
|
|
ws.send(
|
|
serializeRelayMessage({
|
|
type: 'typing',
|
|
sessionId,
|
|
from: 'ai',
|
|
isTyping,
|
|
}),
|
|
)
|
|
return true
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
export function sendSessionEventToSession(
|
|
sessionId: string,
|
|
payload: SessionEventMessage['payload'],
|
|
accountId?: string,
|
|
): boolean {
|
|
const ws = getActiveWebSocket(accountId)
|
|
if (!ws || ws.readyState !== WebSocket.OPEN) return false
|
|
|
|
try {
|
|
const msg: SessionEventMessage = {
|
|
type: 'session_event',
|
|
sessionId,
|
|
timestamp: Date.now(),
|
|
payload,
|
|
}
|
|
|
|
ws.send(serializeRelayMessage(msg))
|
|
return true
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
function resolveQueueMode(
|
|
ctx: ChannelGatewayContext<ResolvedWebAccount>,
|
|
): SessionQueueMode {
|
|
const byChannel = ctx.cfg.messages?.queue?.byChannel as
|
|
| Record<string, unknown>
|
|
| undefined
|
|
const rawMode = byChannel?.clawshow ?? ctx.cfg.messages?.queue?.mode
|
|
|
|
if (typeof rawMode !== 'string') return 'collect'
|
|
return KNOWN_QUEUE_MODES.has(rawMode as SessionQueueMode)
|
|
? (rawMode as SessionQueueMode)
|
|
: 'collect'
|
|
}
|
|
|
|
function buildSessionStateKey(accountId: string, sessionId: string): string {
|
|
return `${accountId}:${sessionId}`
|
|
}
|
|
|
|
export const webGatewayAdapter: ChannelGatewayAdapter<ResolvedWebAccount> = {
|
|
startAccount: async (ctx: ChannelGatewayContext<ResolvedWebAccount>) => {
|
|
return startWebGateway(ctx)
|
|
},
|
|
}
|