diff --git a/README.md b/README.md new file mode 100644 index 0000000..ee5050e --- /dev/null +++ b/README.md @@ -0,0 +1,125 @@ +--- +name: clawshow-gateway-connect +description: Install and activate @bowong/clawshow-gateway in OpenClaw, then migrate existing Gateway channel configuration to ClawShow with rollback safety. Use when a user wants one-step plugin install plus config migration without repeating requirements each run, applying changes via Config RPC config.apply full replace with minimal file access and no default outbound test traffic. +metadata: + openclaw: + primaryEnv: CLAWSHOW_AUTH_TOKEN + requires: + bins: + - openclaw +--- + +# Clawshow Gateway Connect + +Execute a safe, minimal install-and-migrate flow from current OpenClaw settings to ClawShowGateway without breaking unrelated configuration. + +## Runtime Inputs + +- Required: `authToken` +- Optional: `relayUrl` (default `wss://clawshow-api.bowong.cc`) +- If `authToken` is missing, ask once and stop before editing. + +## Workflow + +1. Install ws dependency +- Run `npm i ws` in the project root before proceeding. +- Confirm installation succeeds before continuing. + +2. Identify current config source +- Search only OpenClaw gateway config files (`*.json`, `*.yaml`, `*.yml`) in the active project path. +- Do not read `.env`, secrets files, deployment manifests, or unrelated workspace files. +- Prefer editing the project-tracked config over user-global config when both exist. +- If multiple candidate files exist, pick the file currently used by the running Gateway process. + +3. Create rollback copy before edits +- Save a sibling backup copy of the target config file with timestamp suffix (for example `.bak.20260313-1405`). +- Never delete the backup in the same task. + +4. Check OpenClaw version gate (before plugin install) +- Query current OpenClaw version using the supported command only: + - `openclaw version` +- Compare against `2026.3.2` using OpenClaw's version ordering. +- If current version is earlier than `2026.3.2`, upgrade OpenClaw using `openclaw update` first, then re-run: + - `openclaw version` +- Continue only after version check/upgrade is complete. + +5. Install plugin/package +- Install using OpenClaw plugin manager only: + - `openclaw plugins install @bowong/clawshow-gateway` +- Verify installation: + - `openclaw plugins list` +- Confirm package id is exactly `@bowong/clawshow-gateway`. +- If already installed, do not reinstall unless version repair is needed. + +6. Migrate channel config to ClawShow +- Ensure package id references `@bowong/clawshow-gateway` where plugin/package declarations are required. +- Ensure channel section exists under `channels.clawshow`. +- Set: + - `channels.clawshow.enabled: true` (required) + - `channels.clawshow.relayUrl: "wss://clawshow-api.bowong.cc"` (optional) + - `channels.clawshow.authToken: ""` (required) + - `channels.clawshow.name: "ClawShow Gateway"` (required) + - `channels.clawshow.dmPolicy: "open"` (required) + - `channels.clawshow.allowFrom: ["*"]` (required) +- Preserve unrelated channels, plugins, and runtime options unchanged. + +7. Remove conflicting legacy routing only when necessary +- If older channel config routes the same outbound traffic target, disable only the overlapping route. +- Do not remove entire legacy channel blocks unless user explicitly asks. + +8. Validate config shape +- Confirm JSON/YAML syntax after edit. +- Confirm required keys exist for ClawShow (`authToken` at minimum). +- Confirm package/plugin id and channel id are consistent (`clawshow`). +- Confirm `enabled`, `relayUrl`, `authToken`, `name`, `dmPolicy`, and `allowFrom` are directly under `channels.clawshow` (no `default` layer). + +9. Apply config (Config RPC only) +- Follow `Config RPC (programmatic updates)` exactly, using `config.apply (full replace)`. +- Do not use direct file-edit-only activation, `config.patch`, or manual restart as the primary path. +- Execute: + - `openclaw gateway call config.get --params '{}'` + - Capture `payload.hash` as `baseHash`. + - Build the full post-migration config as one JSON5 string in `raw`. + - `openclaw gateway call config.apply --params '{ "raw": "", "baseHash": "", "note": "migrate to @bowong/clawshow-gateway" }'` +- Respect control-plane rate limits for write RPCs (`config.apply`, `config.patch`, `update.run`): 3 requests per 60 seconds per `deviceId+clientIp`. +- If apply fails due to stale hash/conflict, call `config.get` again, rebase on newest config, and retry once. + +10. Verify runtime behavior +- Do not force a manual restart here; `config.apply` already validates, writes, and restarts in one step. +- Run status/probe command if available. +- Do not send outbound test messages by default. +- Only send an external test message if the user explicitly asks for network verification. + +11. Report outcome +- Include the backup file path. +- Include verification command results. +- If verification is blocked, state the blocker and provide the exact next command. + +## Canonical Target Snippet + +Use this structure when writing or repairing JSON config: + +```json +{ + "channels": { + "clawshow": { + "enabled": true, + "relayUrl": "wss://clawshow-api.bowong.cc", + "authToken": "YOUR_CLAWSHOW_AUTH_TOKEN", + "name": "ClawShow Gateway", + "dmPolicy": "open", + "allowFrom": [ + "*" + ] + } + } +} +``` + +## Guardrails + +- Keep edits minimal and reversible. +- Never invent secrets; leave placeholders for missing keys. +- Never exfiltrate secrets from local files. +- Never rewrite formatting style of the whole file. +- Never remove unrelated keys to "clean up". diff --git a/index.ts b/index.ts new file mode 100644 index 0000000..f2fa483 --- /dev/null +++ b/index.ts @@ -0,0 +1,15 @@ +import type { OpenClawPluginApi } from 'openclaw/plugin-sdk' +import { emptyPluginConfigSchema } from 'openclaw/plugin-sdk' +import { webChannelPlugin } from './src/plugin.js' + +const plugin = { + id: 'clawshow-gateway', + name: 'clawshow-gateway', + description: 'Web channel plugin — browser chat via WebSocket relay', + configSchema: emptyPluginConfigSchema(), + register(api: OpenClawPluginApi) { + api.registerChannel({ plugin: webChannelPlugin }) + }, +} + +export default plugin diff --git a/openclaw.plugin.json b/openclaw.plugin.json new file mode 100644 index 0000000..7d952a3 --- /dev/null +++ b/openclaw.plugin.json @@ -0,0 +1,9 @@ +{ + "id": "clawshow-gateway", + "channels": ["clawshow"], + "configSchema": { + "type": "object", + "additionalProperties": false, + "properties": {} + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..569fb13 --- /dev/null +++ b/package.json @@ -0,0 +1,43 @@ +{ + "name": "@bowong/clawshow-gateway", + "version": "2026.3.16-12", + "description": "OpenClaw Web channel plugin — browser-based chat via WebSocket relay", + "type": "module", + "main": "dist/src/index.js", + "types": "dist/src/index.d.ts", + "publishConfig": { + "access": "public", + "registry": "https://registry.npmjs.org/" + }, + "exports": { + ".": { + "import": "./dist/src/index.js", + "types": "./dist/src/index.d.ts" + } + }, + "files": [ + "dist", + "openclaw.plugin.json", + "package.json", + "README.md" + ], + "openclaw": { + "extensions": [ + "./dist/index.js" + ] + }, + "scripts": { + "build": "node -e \"require('fs').rmSync('dist',{ recursive: true, force: true })\" && bun x tsc -p tsconfig.build.json", + "typecheck": "bun x tsc --noEmit", + "clean": "node -e \"require('fs').rmSync('dist',{ recursive: true, force: true })\"" + }, + "peerDependencies": { + "openclaw": "^2026.3.2" + }, + "dependencies": { + "ws": "^8.19.0" + }, + "devDependencies": { + "@types/ws": "^8.18.1" + } +} \ No newline at end of file diff --git a/src/config.ts b/src/config.ts new file mode 100644 index 0000000..ec2bfd4 --- /dev/null +++ b/src/config.ts @@ -0,0 +1,103 @@ +/** + * Web channel configuration adapter. + * Resolves account configuration from OpenClaw config. + */ + +import type { + ChannelAccountSnapshot, + ChannelConfigAdapter, + OpenClawConfig, +} from 'openclaw/plugin-sdk' +import type { ResolvedWebAccount, WebAccountConfig } from './types.js' + +const DEFAULT_ACCOUNT_ID = 'default' + +type WebChannelConfig = WebAccountConfig & { + accounts?: Record +} + +export function listWebAccountIds(cfg: OpenClawConfig): string[] { + const webConfig = cfg.channels?.clawshow as WebChannelConfig | undefined + if (!webConfig) return [] + + const ids: string[] = [] + // Always include default if top-level config exists + if (webConfig.relayUrl) { + ids.push(DEFAULT_ACCOUNT_ID) + } + // Additional named accounts + if (webConfig.accounts) { + for (const id of Object.keys(webConfig.accounts)) { + if (!ids.includes(id)) { + ids.push(id) + } + } + } + return ids +} + +export function resolveWebAccount( + cfg: OpenClawConfig, + accountId?: string | null, +): ResolvedWebAccount { + const id = accountId ?? DEFAULT_ACCOUNT_ID + const webConfig = cfg.channels?.clawshow as WebChannelConfig | undefined + const accountConfig = + id !== DEFAULT_ACCOUNT_ID ? webConfig?.accounts?.[id] : undefined + + const base = webConfig + ? { + enabled: webConfig.enabled, + relayUrl: webConfig.relayUrl, + authToken: webConfig.authToken, + dmPolicy: webConfig.dmPolicy, + allowFrom: webConfig.allowFrom, + name: webConfig.name, + } + : {} + const merged: WebAccountConfig = { + relayUrl: '', + ...base, + ...accountConfig, + } + + const authToken = merged.authToken ?? process.env.WEB_RELAY_AUTH_TOKEN ?? '' + + return { + accountId: id, + name: merged.name, + enabled: merged.enabled !== false, + relayUrl: merged.relayUrl ?? '', + authToken, + config: merged, + } +} + +export function isWebConfigured(account: ResolvedWebAccount): boolean { + return Boolean(account.relayUrl && account.authToken) +} + +export const webConfigAdapter: ChannelConfigAdapter = { + listAccountIds: listWebAccountIds, + resolveAccount: resolveWebAccount, + defaultAccountId: (_cfg: OpenClawConfig) => DEFAULT_ACCOUNT_ID, + isEnabled: (account: ResolvedWebAccount, _cfg: OpenClawConfig) => + account.enabled, + isConfigured: (account: ResolvedWebAccount, _cfg: OpenClawConfig) => + isWebConfigured(account), + unconfiguredReason: (account: ResolvedWebAccount, _cfg: OpenClawConfig) => { + if (!account.relayUrl) return 'relayUrl not configured' + if (!account.authToken) + return 'authToken not configured (set WEB_RELAY_AUTH_TOKEN or channels.clawshow.authToken)' + return 'not configured' + }, + describeAccount: ( + account: ResolvedWebAccount, + _cfg: OpenClawConfig, + ): ChannelAccountSnapshot => ({ + accountId: account.accountId, + name: account.name, + enabled: account.enabled, + configured: isWebConfigured(account), + }), +} diff --git a/src/gateway.ts b/src/gateway.ts new file mode 100644 index 0000000..003f4ff --- /dev/null +++ b/src/gateway.ts @@ -0,0 +1,580 @@ +/** + * 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() +const sessionActiveDispatches = new Map() +const sessionQueuedDispatches = new Map() + +const KNOWN_QUEUE_MODES = new Set([ + '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, +): Promise { + const { account, abortSignal, log } = ctx + + let reconnectAttempts = 0 + let reconnectMs = INITIAL_RECONNECT_MS + let reconnectTimer: ReturnType | null = null // [R3] track timer for abort cleanup + + const connect = (): Promise => + new Promise((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 | 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, + msg: ChatMessage, +): Promise { + 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, +): SessionQueueMode { + const byChannel = ctx.cfg.messages?.queue?.byChannel as + | Record + | 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 = { + startAccount: async (ctx: ChannelGatewayContext) => { + return startWebGateway(ctx) + }, +} diff --git a/src/index.ts b/src/index.ts new file mode 100644 index 0000000..ff987ec --- /dev/null +++ b/src/index.ts @@ -0,0 +1,23 @@ +/** + * @sker-pro/openclaw-channel-web + * + * OpenClaw ChannelPlugin for browser-based chat via WebSocket relay. + */ + +export { webChannelPlugin } from './plugin.js' +export type { ResolvedWebAccount, WebAccountConfig } from './types.js' +export { + sendTextToSession, + sendTypingToSession, + getActiveWebSocket, +} from './gateway.js' +export { + type RelayMessage, + type ChatMessage, + type AuthMessage, + type SessionEventMessage, + generateMessageId, + parseRelayMessage, + serializeRelayMessage, + HEARTBEAT_INTERVAL_MS, +} from './protocol.js' diff --git a/src/outbound.ts b/src/outbound.ts new file mode 100644 index 0000000..0d5efb7 --- /dev/null +++ b/src/outbound.ts @@ -0,0 +1,33 @@ +/** + * Web channel outbound adapter — sends messages via the gateway WebSocket. + */ + +import type { + ChannelOutboundAdapter, + ChannelOutboundContext, +} from 'openclaw/plugin-sdk' +import { generateMessageId } from './protocol.js' +import { sendTextToSession } from './gateway.js' + +export const webOutboundAdapter: ChannelOutboundAdapter = { + deliveryMode: 'gateway' as const, + textChunkLimit: 4000, + + sendText: async (ctx: ChannelOutboundContext) => { + const ok = sendTextToSession(ctx.to, ctx.text, ctx.accountId ?? undefined) + return { + channel: 'clawshow', + messageId: ok ? generateMessageId() : '', + } + }, + + sendMedia: async (ctx: ChannelOutboundContext) => { + // Web channel doesn't support native media — send as text link + const text = ctx.mediaUrl ? `${ctx.text}\n\n${ctx.mediaUrl}` : ctx.text + const ok = sendTextToSession(ctx.to, text, ctx.accountId ?? undefined) + return { + channel: 'clawshow', + messageId: ok ? generateMessageId() : '', + } + }, +} diff --git a/src/plugin.ts b/src/plugin.ts new file mode 100644 index 0000000..bc77277 --- /dev/null +++ b/src/plugin.ts @@ -0,0 +1,44 @@ +/** + * Web ChannelPlugin — full plugin definition following the ChannelPlugin interface. + * + * Pattern follows Telegram/WhatsApp channel plugins. + */ + +import type { ChannelPlugin } from 'openclaw/plugin-sdk' +import { webConfigAdapter } from './config.js' +import { webGatewayAdapter } from './gateway.js' +import { webOutboundAdapter } from './outbound.js' +import { webSecurityAdapter } from './security.js' +import { webSetupAdapter } from './setup.js' +import type { ResolvedWebAccount } from './types.js' + +export const webChannelPlugin: ChannelPlugin = { + id: 'clawshow', + meta: { + id: 'clawshow', + label: 'Clawshow', + selectionLabel: 'Clawshow (browser chat)', + docsPath: '/channels/clawshow', + blurb: 'Browser chat via WebSocket relay', + order: 50, + }, + capabilities: { + chatTypes: ['direct'], + media: false, + reactions: false, + }, + defaults: { + queue: { + debounceMs: 500, + }, + }, + reload: { configPrefixes: ['channels.clawshow'] }, + config: webConfigAdapter, + gateway: webGatewayAdapter, + outbound: webOutboundAdapter, + security: webSecurityAdapter, + setup: webSetupAdapter, + pairing: { + idLabel: 'sessionId', + }, +} diff --git a/src/protocol.ts b/src/protocol.ts new file mode 100644 index 0000000..be2873a --- /dev/null +++ b/src/protocol.ts @@ -0,0 +1,394 @@ +/** + * Shared WebSocket message protocol for Web Channel. + * Used by: web-relay (Cloudflare Worker), openclaw-channel-web (plugin), web-chat (frontend). + */ + +// ── Roles ─────────────────────────────────────────────────────────── + +export type WebSocketRole = 'browser' | 'openclaw' + +// ── Message Types ─────────────────────────────────────────────────── + +export type AuthMessage = { + type: 'auth' + role: WebSocketRole + token: string + sessionId?: string // required for browser role +} + +export type AuthResultMessage = { + type: 'auth_result' + ok: boolean + error?: string + sessionId?: string + userId?: string + userName?: string + lobsterName?: string +} + +export type ChatMessage = { + type: 'message' + id: string + sessionId: string + from: 'user' | 'ai' + text: string + timestamp: number + userId?: string + userName?: string + targetLobsterId?: string +} + +export type AckMessage = { + type: 'ack' + messageId: string + sessionId: string +} + +export type TypingMessage = { + type: 'typing' + sessionId: string + from: 'user' | 'ai' + isTyping: boolean +} + +export type HistoryRequestMessage = { + type: 'history_request' + sessionId: string + before?: number // timestamp cursor for pagination + limit?: number +} + +export type HistoryResponseMessage = { + type: 'history_response' + sessionId: string + messages: ChatMessage[] + hasMore: boolean +} + +export type PingMessage = { + type: 'ping' + timestamp: number +} + +export type PongMessage = { + type: 'pong' + timestamp: number +} + +export type ErrorMessage = { + type: 'error' + code: string + message: string + sessionId?: string +} + +export type SessionQueueMode = + | 'collect' + | 'followup' + | 'steer' + | 'interrupt' + | 'steer-backlog' + +export type QueueStatusState = 'processing' | 'enqueued' | 'idle' + +export type QueueStatusSessionEvent = { + event: 'queue_status' + state: QueueStatusState + mode?: SessionQueueMode + depth?: number + messageId?: string +} + +export type SessionEventPayload = QueueStatusSessionEvent + +export type SessionEventMessage = { + type: 'session_event' + sessionId: string + timestamp: number + payload: SessionEventPayload +} + +// Lobster 上/下线推送给 browser +export type LobsterStatusMessage = { + type: 'lobster_status' + action: 'online' | 'offline' | 'list' + lobsters: Array<{ lobsterId: string; lobsterName: string }> +} + +// Browser 上/下线推送给 openclaw +export type UserStatusMessage = { + type: 'user_status' + action: 'online' | 'offline' + sessionId: string + userId: string + userName?: string +} + +export type HistoryClearMessage = { + type: 'history_clear' + sessionId: string +} + +export type HistoryClearResultMessage = { + type: 'history_clear_result' + sessionId: string + ok: boolean + error?: string +} + +// ── Union Type ────────────────────────────────────────────────────── + +export type RelayMessage = + | AuthMessage + | AuthResultMessage + | ChatMessage + | AckMessage + | TypingMessage + | HistoryRequestMessage + | HistoryResponseMessage + | PingMessage + | PongMessage + | ErrorMessage + | SessionEventMessage + | LobsterStatusMessage + | UserStatusMessage + | HistoryClearMessage + | HistoryClearResultMessage + +// ── Helpers ───────────────────────────────────────────────────────── + +export function parseRelayMessage(raw: string): RelayMessage | null { + try { + const parsed = JSON.parse(raw) as unknown + if (typeof parsed !== 'object' || parsed === null || !('type' in parsed)) { + return null + } + + const msg = parsed as { type: string } + + switch (msg.type) { + case 'auth': + return isValidAuthMessage(parsed) ? (parsed as AuthMessage) : null + case 'auth_result': + return isValidAuthResultMessage(parsed) ? (parsed as AuthResultMessage) : null + case 'message': + return isValidChatMessage(parsed) ? (parsed as ChatMessage) : null + case 'ack': + return isValidAckMessage(parsed) ? (parsed as AckMessage) : null + case 'typing': + return isValidTypingMessage(parsed) ? (parsed as TypingMessage) : null + case 'history_request': + return isValidHistoryRequestMessage(parsed) ? (parsed as HistoryRequestMessage) : null + case 'history_response': + return isValidHistoryResponseMessage(parsed) ? (parsed as HistoryResponseMessage) : null + case 'ping': + return isValidPingMessage(parsed) ? (parsed as PingMessage) : null + case 'pong': + return isValidPongMessage(parsed) ? (parsed as PongMessage) : null + case 'error': + return isValidErrorMessage(parsed) ? (parsed as ErrorMessage) : null + case 'session_event': + return isValidSessionEventMessage(parsed) ? (parsed as SessionEventMessage) : null + case 'lobster_status': + return isValidLobsterStatusMessage(parsed) ? (parsed as LobsterStatusMessage) : null + case 'user_status': + return isValidUserStatusMessage(parsed) ? (parsed as UserStatusMessage) : null + case 'history_clear': + return isValidHistoryClearMessage(parsed) ? (parsed as HistoryClearMessage) : null + case 'history_clear_result': + return isValidHistoryClearResultMessage(parsed) ? (parsed as HistoryClearResultMessage) : null + default: + return null + } + } catch { + return null + } +} + +// Type guards for each message type +function isValidAuthMessage(msg: unknown): msg is AuthMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as AuthMessage).type === 'auth' && + typeof (msg as AuthMessage).role === 'string' && + ['browser', 'openclaw'].includes((msg as AuthMessage).role) && + typeof (msg as AuthMessage).token === 'string' && + ((msg as AuthMessage).sessionId === undefined || typeof (msg as AuthMessage).sessionId === 'string') + ) +} + +function isValidAuthResultMessage(msg: unknown): msg is AuthResultMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as AuthResultMessage).type === 'auth_result' && + typeof (msg as AuthResultMessage).ok === 'boolean' && + ((msg as AuthResultMessage).error === undefined || typeof (msg as AuthResultMessage).error === 'string') && + ((msg as AuthResultMessage).sessionId === undefined || typeof (msg as AuthResultMessage).sessionId === 'string') + ) +} + +function isValidChatMessage(msg: unknown): msg is ChatMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as ChatMessage).type === 'message' && + typeof (msg as ChatMessage).id === 'string' && + typeof (msg as ChatMessage).sessionId === 'string' && + ['user', 'ai'].includes((msg as ChatMessage).from) && + typeof (msg as ChatMessage).text === 'string' && + typeof (msg as ChatMessage).timestamp === 'number' + ) +} + +function isValidAckMessage(msg: unknown): msg is AckMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as AckMessage).type === 'ack' && + typeof (msg as AckMessage).messageId === 'string' && + typeof (msg as AckMessage).sessionId === 'string' + ) +} + +function isValidTypingMessage(msg: unknown): msg is TypingMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as TypingMessage).type === 'typing' && + typeof (msg as TypingMessage).sessionId === 'string' && + ['user', 'ai'].includes((msg as TypingMessage).from) && + typeof (msg as TypingMessage).isTyping === 'boolean' + ) +} + +function isValidHistoryRequestMessage(msg: unknown): msg is HistoryRequestMessage { + const m = msg as HistoryRequestMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'history_request' && + typeof m.sessionId === 'string' && + (m.before === undefined || typeof m.before === 'number') && + (m.limit === undefined || typeof m.limit === 'number') + ) +} + +function isValidHistoryResponseMessage(msg: unknown): msg is HistoryResponseMessage { + const m = msg as HistoryResponseMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'history_response' && + typeof m.sessionId === 'string' && + Array.isArray(m.messages) && + typeof m.hasMore === 'boolean' + ) +} + +function isValidPingMessage(msg: unknown): msg is PingMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as PingMessage).type === 'ping' && + typeof (msg as PingMessage).timestamp === 'number' + ) +} + +function isValidPongMessage(msg: unknown): msg is PongMessage { + return ( + typeof msg === 'object' && msg !== null && + (msg as PongMessage).type === 'pong' && + typeof (msg as PongMessage).timestamp === 'number' + ) +} + +function isValidErrorMessage(msg: unknown): msg is ErrorMessage { + const m = msg as ErrorMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'error' && + typeof m.code === 'string' && + typeof m.message === 'string' && + (m.sessionId === undefined || typeof m.sessionId === 'string') + ) +} + +function isValidSessionEventMessage(msg: unknown): msg is SessionEventMessage { + const m = msg as SessionEventMessage + return ( + typeof msg === 'object' && + msg !== null && + m.type === 'session_event' && + typeof m.sessionId === 'string' && + typeof m.timestamp === 'number' && + isValidSessionEventPayload(m.payload) + ) +} + +function isValidSessionEventPayload(payload: unknown): payload is SessionEventPayload { + if (typeof payload !== 'object' || payload === null) return false + const typed = payload as SessionEventPayload + if (typed.event !== 'queue_status') return false + if (!['processing', 'enqueued', 'idle'].includes(typed.state)) return false + if ( + typed.mode !== undefined && + !['collect', 'followup', 'steer', 'interrupt', 'steer-backlog'].includes(typed.mode) + ) { + return false + } + return ( + (typed.depth === undefined || typeof typed.depth === 'number') && + (typed.messageId === undefined || typeof typed.messageId === 'string') + ) +} + +function isValidLobsterStatusMessage(msg: unknown): msg is LobsterStatusMessage { + const m = msg as LobsterStatusMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'lobster_status' && + typeof m.action === 'string' && + ['online', 'offline', 'list'].includes(m.action) && + Array.isArray(m.lobsters) + ) +} + +function isValidUserStatusMessage(msg: unknown): msg is UserStatusMessage { + const m = msg as UserStatusMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'user_status' && + typeof m.action === 'string' && + ['online', 'offline'].includes(m.action) && + typeof m.sessionId === 'string' && + typeof m.userId === 'string' + ) +} + +function isValidHistoryClearMessage(msg: unknown): msg is HistoryClearMessage { + const m = msg as HistoryClearMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'history_clear' && + typeof m.sessionId === 'string' + ) +} + +function isValidHistoryClearResultMessage(msg: unknown): msg is HistoryClearResultMessage { + const m = msg as HistoryClearResultMessage + return ( + typeof msg === 'object' && msg !== null && + m.type === 'history_clear_result' && + typeof m.sessionId === 'string' && + typeof m.ok === 'boolean' && + (m.error === undefined || typeof m.error === 'string') + ) +} + +export function serializeRelayMessage(msg: RelayMessage): string { + return JSON.stringify(msg) +} + +export function generateMessageId(): string { + return `msg-${Date.now()}-${Math.random().toString(36).slice(2, 10)}` +} + +// ── Constants ─────────────────────────────────────────────────────── + +export const HEARTBEAT_INTERVAL_MS = 30_000 +export const MAX_HISTORY_PER_SESSION = 200 +export const DEFAULT_HISTORY_LIMIT = 50 diff --git a/src/security.ts b/src/security.ts new file mode 100644 index 0000000..3e9a9c7 --- /dev/null +++ b/src/security.ts @@ -0,0 +1,25 @@ +/** + * Web channel DM security adapter. + */ + +import type { + ChannelSecurityAdapter, + ChannelSecurityContext, + ChannelSecurityDmPolicy, +} from 'openclaw/plugin-sdk' +import type { ResolvedWebAccount } from './types.js' + +export const webSecurityAdapter: ChannelSecurityAdapter = { + resolveDmPolicy: ( + ctx: ChannelSecurityContext, + ): ChannelSecurityDmPolicy => { + const policy = ctx.account.config.dmPolicy ?? 'open' + return { + policy, + allowFrom: ctx.account.config.allowFrom ?? null, + policyPath: 'channels.clawshow.dmPolicy', + allowFromPath: 'channels.clawshow.allowFrom', + approveHint: 'Add the session ID to channels.clawshow.allowFrom', + } + }, +} diff --git a/src/setup.ts b/src/setup.ts new file mode 100644 index 0000000..be8c146 --- /dev/null +++ b/src/setup.ts @@ -0,0 +1,77 @@ +/** + * Web channel setup adapter — applies account configuration. + */ + +import type { + ChannelSetupAdapter, + ChannelSetupInput, + OpenClawConfig, +} from 'openclaw/plugin-sdk' + +export const webSetupAdapter: ChannelSetupAdapter = { + applyAccountConfig: (params: { + cfg: OpenClawConfig + accountId: string + input: ChannelSetupInput + }): OpenClawConfig => { + const { cfg, accountId, input } = params + + if (accountId === 'default') { + return { + ...cfg, + channels: { + ...cfg.channels, + clawshow: { + ...(cfg.channels?.clawshow as Record | undefined), + enabled: true, + ...(input.url ? { relayUrl: input.url } : {}), + ...(input.token ? { authToken: input.token } : {}), + ...(input.name ? { name: input.name } : {}), + }, + }, + } + } + + const webConfig = (cfg.channels?.clawshow as Record) ?? {} + const accounts = (webConfig.accounts as Record) ?? {} + + return { + ...cfg, + channels: { + ...cfg.channels, + clawshow: { + ...webConfig, + enabled: true, + accounts: { + ...accounts, + [accountId]: { + ...(accounts[accountId] as Record | undefined), + enabled: true, + ...(input.url ? { relayUrl: input.url } : {}), + ...(input.token ? { authToken: input.token } : {}), + ...(input.name ? { name: input.name } : {}), + }, + }, + }, + }, + } + }, + + validateInput: (params: { + cfg: OpenClawConfig + accountId: string + input: ChannelSetupInput + }): string | null => { + if (!params.input.url && !params.input.useEnv) { + return 'Web channel requires relayUrl (--url) or --use-env.' + } + if ( + !params.input.token && + !params.input.useEnv && + !process.env.WEB_RELAY_AUTH_TOKEN + ) { + return 'Web channel requires authToken (--token), --use-env, or WEB_RELAY_AUTH_TOKEN env var.' + } + return null + }, +} diff --git a/src/types.ts b/src/types.ts new file mode 100644 index 0000000..68c91f3 --- /dev/null +++ b/src/types.ts @@ -0,0 +1,21 @@ +/** + * Web channel account configuration and resolved types. + */ + +export interface WebAccountConfig { + enabled?: boolean + relayUrl: string + authToken?: string + dmPolicy?: 'open' | 'allowlist' + allowFrom?: string[] + name?: string +} + +export interface ResolvedWebAccount { + accountId: string + name?: string + enabled: boolean + relayUrl: string + authToken: string + config: WebAccountConfig +} diff --git a/tsconfig.build.json b/tsconfig.build.json new file mode 100644 index 0000000..8ad24df --- /dev/null +++ b/tsconfig.build.json @@ -0,0 +1,12 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "noEmit": false, + "declaration": true, + "declarationMap": true, + "sourceMap": true, + "outDir": "dist", + "rootDir": "." + }, + "include": ["index.ts", "src/**/*.ts"] +} diff --git a/tsconfig.json b/tsconfig.json new file mode 100644 index 0000000..81cb018 --- /dev/null +++ b/tsconfig.json @@ -0,0 +1,16 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "strict": true, + "skipLibCheck": true, + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "resolveJsonModule": true, + "isolatedModules": true, + "noEmit": true, + "baseUrl": "." + }, + "include": ["index.ts", "src/**/*.ts"] +}