feat(qq): add QQ Bot channel plugin (Official Bot API)

- Implement QQ Bot API client with token caching
- Add WebSocket monitor for event handling
- Support C2C (single chat) and group messages
- Include pairing mechanism for DM authorization

Also fix memory-core peerDependencies to use workspace:*

extensions/qq/src/monitor.ts

@@ -0,0 +1,534 @@
+/**
+ * QQ Bot WebSocket Monitor
+ *
+ * Handles WebSocket connection to QQ Bot Gateway for receiving events.
+ */
+
+import WebSocket from "ws";
+import type { MoltbotConfig, MarkdownTableMode } from "clawdbot/plugin-sdk";
+
+import type { ResolvedQQAccount } from "./accounts.js";
+import {
+    getAccessToken,
+    getGatewayUrl,
+    sendC2CMessage,
+    sendGroupMessage,
+} from "./api.js";
+import { getQQRuntime } from "./runtime.js";
+import { chunkQQText } from "./send.js";
+import {
+    OpCode,
+    EventType,
+    DEFAULT_INTENTS,
+    type GatewayPayload,
+    type HelloData,
+    type ReadyData,
+    type QQMessageEvent,
+    type IdentifyData,
+    type ResumeData,
+} from "./types.js";
+
+export interface QQMonitorOptions {
+    account: ResolvedQQAccount;
+    config: MoltbotConfig;
+    abortSignal: AbortSignal;
+    statusSink?: (patch: {
+        lastInboundAt?: number;
+        lastOutboundAt?: number;
+        sessionId?: string;
+    }) => void;
+    log?: {
+        info: (msg: string) => void;
+        error: (msg: string) => void;
+    };
+}
+
+export interface QQMonitorResult {
+    stop: () => void;
+}
+
+const QQ_TEXT_LIMIT = 2000;
+const DEFAULT_MEDIA_MAX_MB = 5;
+
+type QQCoreRuntime = ReturnType<typeof getQQRuntime>;
+
+/**
+ * Start QQ Bot WebSocket monitor
+ */
+export async function monitorQQProvider(
+    options: QQMonitorOptions,
+): Promise<QQMonitorResult> {
+    const { account, config, abortSignal, statusSink, log } = options;
+
+    if (!account.appId || !account.appSecret) {
+        throw new Error("QQ appId and appSecret are required");
+    }
+
+    const core = getQQRuntime();
+    let stopped = false;
+    let ws: WebSocket | null = null;
+    let heartbeatInterval: ReturnType<typeof setInterval> | null = null;
+    let sessionId: string | null = null;
+    let lastSeq: number | null = null;
+    let reconnectAttempts = 0;
+    const maxReconnectAttempts = 10;
+
+    const stop = () => {
+        stopped = true;
+        if (heartbeatInterval) {
+            clearInterval(heartbeatInterval);
+            heartbeatInterval = null;
+        }
+        if (ws) {
+            ws.close();
+            ws = null;
+        }
+    };
+
+    abortSignal.addEventListener("abort", stop, { once: true });
+
+    const connect = async () => {
+        if (stopped || abortSignal.aborted) return;
+
+        try {
+            const token = await getAccessToken(account.appId!, account.appSecret!);
+            const gatewayUrl = await getGatewayUrl(token);
+
+            log?.info(`[${account.accountId}] Connecting to gateway: ${gatewayUrl}`);
+
+            ws = new WebSocket(gatewayUrl);
+
+            ws.on("open", () => {
+                log?.info(`[${account.accountId}] WebSocket connected`);
+                reconnectAttempts = 0;
+            });
+
+            ws.on("message", async (data) => {
+                try {
+                    const payload = JSON.parse(data.toString()) as GatewayPayload;
+                    await handlePayload(payload, token);
+                } catch (err) {
+                    log?.error(
+                        `[${account.accountId}] Failed to parse message: ${String(err)}`,
+                    );
+                }
+            });
+
+            ws.on("close", (code, reason) => {
+                log?.info(
+                    `[${account.accountId}] WebSocket closed: ${code} ${reason.toString()}`,
+                );
+                if (heartbeatInterval) {
+                    clearInterval(heartbeatInterval);
+                    heartbeatInterval = null;
+                }
+                scheduleReconnect();
+            });
+
+            ws.on("error", (err) => {
+                log?.error(`[${account.accountId}] WebSocket error: ${String(err)}`);
+            });
+        } catch (err) {
+            log?.error(`[${account.accountId}] Failed to connect: ${String(err)}`);
+            scheduleReconnect();
+        }
+    };
+
+    const scheduleReconnect = () => {
+        if (stopped || abortSignal.aborted) return;
+        if (reconnectAttempts >= maxReconnectAttempts) {
+            log?.error(`[${account.accountId}] Max reconnect attempts reached`);
+            return;
+        }
+
+        reconnectAttempts++;
+        const delay = Math.min(1000 * Math.pow(2, reconnectAttempts), 30000);
+        log?.info(
+            `[${account.accountId}] Reconnecting in ${delay}ms (attempt ${reconnectAttempts})`,
+        );
+        setTimeout(connect, delay);
+    };
+
+    const handlePayload = async (
+        payload: GatewayPayload,
+        token: string,
+    ): Promise<void> => {
+        // Update sequence number
+        if (payload.s !== undefined) {
+            lastSeq = payload.s;
+        }
+
+        switch (payload.op) {
+            case OpCode.Hello: {
+                const hello = payload.d as HelloData;
+                log?.info(
+                    `[${account.accountId}] Received Hello, heartbeat interval: ${hello.heartbeat_interval}ms`,
+                );
+
+                // Start heartbeat
+                if (heartbeatInterval) clearInterval(heartbeatInterval);
+                heartbeatInterval = setInterval(() => {
+                    sendHeartbeat();
+                }, hello.heartbeat_interval);
+
+                // Send Identify or Resume
+                if (sessionId && lastSeq !== null) {
+                    sendResume(token);
+                } else {
+                    sendIdentify(token);
+                }
+                break;
+            }
+
+            case OpCode.HeartbeatAck:
+                // Heartbeat acknowledged
+                break;
+
+            case OpCode.Dispatch:
+                statusSink?.({ lastInboundAt: Date.now() });
+                await handleEvent(payload.t!, payload.d, token);
+                break;
+
+            case OpCode.Reconnect:
+                log?.info(`[${account.accountId}] Received Reconnect, reconnecting...`);
+                ws?.close();
+                break;
+
+            case OpCode.InvalidSession: {
+                const resumable = payload.d as boolean;
+                log?.info(
+                    `[${account.accountId}] Invalid session, resumable: ${resumable}`,
+                );
+                if (!resumable) {
+                    sessionId = null;
+                    lastSeq = null;
+                }
+                // Wait a bit before reconnecting
+                setTimeout(() => {
+                    if (resumable && sessionId) {
+                        sendResume(token);
+                    } else {
+                        sendIdentify(token);
+                    }
+                }, 1000 + Math.random() * 4000);
+                break;
+            }
+        }
+    };
+
+    const sendHeartbeat = () => {
+        if (!ws || ws.readyState !== WebSocket.OPEN) return;
+
+        const payload: GatewayPayload<number | null> = {
+            op: OpCode.Heartbeat,
+            d: lastSeq,
+        };
+
+        ws.send(JSON.stringify(payload));
+    };
+
+    const sendIdentify = async (token: string) => {
+        if (!ws || ws.readyState !== WebSocket.OPEN) return;
+
+        const intents = account.config.intents ?? DEFAULT_INTENTS;
+
+        const payload: GatewayPayload<IdentifyData> = {
+            op: OpCode.Identify,
+            d: {
+                token: `QQBot ${token}`,
+                intents,
+                properties: {
+                    $os: "linux",
+                    $browser: "moltbot",
+                    $device: "moltbot",
+                },
+            },
+        };
+
+        log?.info(`[${account.accountId}] Sending Identify with intents: ${intents}`);
+        ws.send(JSON.stringify(payload));
+    };
+
+    const sendResume = (token: string) => {
+        if (!ws || ws.readyState !== WebSocket.OPEN || !sessionId) return;
+
+        const payload: GatewayPayload<ResumeData> = {
+            op: OpCode.Resume,
+            d: {
+                token: `QQBot ${token}`,
+                session_id: sessionId,
+                seq: lastSeq ?? 0,
+            },
+        };
+
+        log?.info(`[${account.accountId}] Sending Resume for session: ${sessionId}`);
+        ws.send(JSON.stringify(payload));
+    };
+
+    const handleEvent = async (
+        eventType: string,
+        data: unknown,
+        token: string,
+    ): Promise<void> => {
+        switch (eventType) {
+            case EventType.READY: {
+                const ready = data as ReadyData;
+                sessionId = ready.session_id;
+                statusSink?.({ sessionId });
+                log?.info(
+                    `[${account.accountId}] Ready! Bot: ${ready.user.username} (${ready.user.id})`,
+                );
+                break;
+            }
+
+            case EventType.RESUMED:
+                log?.info(`[${account.accountId}] Session resumed`);
+                break;
+
+            case EventType.C2C_MESSAGE_CREATE:
+                await handleC2CMessage(data as QQMessageEvent, token);
+                break;
+
+            case EventType.GROUP_AT_MESSAGE_CREATE:
+                await handleGroupMessage(data as QQMessageEvent, token);
+                break;
+
+            case EventType.DIRECT_MESSAGE_CREATE:
+            case EventType.AT_MESSAGE_CREATE:
+            case EventType.MESSAGE_CREATE:
+                // TODO: Implement channel message handling
+                log?.info(`[${account.accountId}] Received channel event: ${eventType}`);
+                break;
+
+            default:
+                log?.info(`[${account.accountId}] Unhandled event: ${eventType}`);
+        }
+    };
+
+    const handleC2CMessage = async (
+        message: QQMessageEvent,
+        token: string,
+    ): Promise<void> => {
+        const senderId = message.author.user_openid;
+        if (!senderId) return;
+
+        log?.info(
+            `[${account.accountId}] C2C message from ${senderId}: ${message.content?.slice(0, 50)}`,
+        );
+
+        await processMessageWithPipeline({
+            message,
+            token,
+            chatType: "c2c",
+            chatId: senderId,
+            senderId,
+            isGroup: false,
+        });
+    };
+
+    const handleGroupMessage = async (
+        message: QQMessageEvent,
+        token: string,
+    ): Promise<void> => {
+        const groupId = message.group_openid;
+        const senderId = message.author.member_openid;
+        if (!groupId || !senderId) return;
+
+        log?.info(
+            `[${account.accountId}] Group message in ${groupId} from ${senderId}: ${message.content?.slice(0, 50)}`,
+        );
+
+        await processMessageWithPipeline({
+            message,
+            token,
+            chatType: "group",
+            chatId: groupId,
+            senderId,
+            isGroup: true,
+        });
+    };
+
+    const processMessageWithPipeline = async (params: {
+        message: QQMessageEvent;
+        token: string;
+        chatType: "c2c" | "group";
+        chatId: string;
+        senderId: string;
+        isGroup: boolean;
+    }): Promise<void> => {
+        const { message, token, chatType, chatId, senderId, isGroup } = params;
+
+        const rawBody = message.content?.trim() || "";
+        if (!rawBody) return;
+
+        const dmPolicy = account.config.dmPolicy ?? "open";
+        const configAllowFrom = (account.config.allowFrom ?? []).map((v) =>
+            String(v),
+        );
+
+        // Check authorization for DMs
+        if (!isGroup && dmPolicy !== "open") {
+            const allowed =
+                configAllowFrom.includes("*") ||
+                configAllowFrom.some(
+                    (entry) => entry.toLowerCase() === senderId.toLowerCase(),
+                );
+
+            if (!allowed) {
+                if (dmPolicy === "pairing") {
+                    const { code, created } = await core.channel.pairing.upsertPairingRequest({
+                        channel: "qq",
+                        id: senderId,
+                        meta: {},
+                    });
+
+                    if (created) {
+                        log?.info(`[${account.accountId}] Pairing request from ${senderId}`);
+                        const replyText = core.channel.pairing.buildPairingReply({
+                            channel: "qq",
+                            idLine: `Your QQ OpenID: ${senderId}`,
+                            code,
+                        });
+
+                        await sendC2CMessage(token, senderId, {
+                            content: replyText,
+                            msg_type: 0,
+                            msg_id: message.id,
+                        });
+                        statusSink?.({ lastOutboundAt: Date.now() });
+                    }
+                }
+                return;
+            }
+        }
+
+        // Resolve agent route
+        const route = core.channel.routing.resolveAgentRoute({
+            cfg: config,
+            channel: "qq",
+            accountId: account.accountId,
+            peer: {
+                kind: isGroup ? "group" : "dm",
+                id: chatId,
+            },
+        });
+
+        const fromLabel = isGroup ? `group:${chatId}` : `user:${senderId}`;
+        const storePath = core.channel.session.resolveStorePath(config.session?.store, {
+            agentId: route.agentId,
+        });
+        const envelopeOptions = core.channel.reply.resolveEnvelopeFormatOptions(config);
+        const previousTimestamp = core.channel.session.readSessionUpdatedAt({
+            storePath,
+            sessionKey: route.sessionKey,
+        });
+
+        const body = core.channel.reply.formatAgentEnvelope({
+            channel: "QQ",
+            from: fromLabel,
+            timestamp: new Date(message.timestamp).getTime(),
+            previousTimestamp,
+            envelope: envelopeOptions,
+            body: rawBody,
+        });
+
+        const ctxPayload = core.channel.reply.finalizeInboundContext({
+            Body: body,
+            RawBody: rawBody,
+            CommandBody: rawBody,
+            From: isGroup ? `qq:group:${chatId}` : `qq:${senderId}`,
+            To: `qq:${chatId}`,
+            SessionKey: route.sessionKey,
+            AccountId: route.accountId,
+            ChatType: isGroup ? "group" : "direct",
+            ConversationLabel: fromLabel,
+            SenderId: senderId,
+            Provider: "qq",
+            Surface: "qq",
+            MessageSid: message.id,
+            OriginatingChannel: "qq",
+            OriginatingTo: `qq:${chatId}`,
+        });
+
+        await core.channel.session.recordInboundSession({
+            storePath,
+            sessionKey: ctxPayload.SessionKey ?? route.sessionKey,
+            ctx: ctxPayload,
+            onRecordError: (err) => {
+                log?.error(`[${account.accountId}] Failed updating session: ${String(err)}`);
+            },
+        });
+
+        const tableMode = core.channel.text.resolveMarkdownTableMode({
+            cfg: config,
+            channel: "qq",
+            accountId: account.accountId,
+        });
+
+        await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({
+            ctx: ctxPayload,
+            cfg: config,
+            dispatcherOptions: {
+                deliver: async (payload) => {
+                    await deliverQQReply({
+                        payload,
+                        token,
+                        chatType,
+                        chatId,
+                        msgId: message.id,
+                        tableMode,
+                    });
+                },
+                onError: (err, info) => {
+                    log?.error(
+                        `[${account.accountId}] QQ ${info.kind} reply failed: ${String(err)}`,
+                    );
+                },
+            },
+        });
+    };
+
+    const deliverQQReply = async (params: {
+        payload: { text?: string; mediaUrls?: string[]; mediaUrl?: string };
+        token: string;
+        chatType: "c2c" | "group";
+        chatId: string;
+        msgId?: string;
+        tableMode?: MarkdownTableMode;
+    }): Promise<void> => {
+        const { payload, token, chatType, chatId, msgId } = params;
+        const tableMode = params.tableMode ?? "code";
+        const text = core.channel.text.convertMarkdownTables(payload.text ?? "", tableMode);
+
+        if (!text) return;
+
+        const chunkMode = core.channel.text.resolveChunkMode(config, "qq", account.accountId);
+        const chunks = core.channel.text.chunkMarkdownTextWithMode(
+            text,
+            QQ_TEXT_LIMIT,
+            chunkMode,
+        );
+
+        const sendFn = chatType === "c2c" ? sendC2CMessage : sendGroupMessage;
+
+        for (let i = 0; i < chunks.length; i++) {
+            const chunk = chunks[i];
+            try {
+                await sendFn(token, chatId, {
+                    content: chunk,
+                    msg_type: 0,
+                    msg_id: i === 0 ? msgId : undefined, // Only use msg_id for first chunk
+                    msg_seq: i + 1,
+                });
+                statusSink?.({ lastOutboundAt: Date.now() });
+            } catch (err) {
+                log?.error(`[${account.accountId}] QQ message send failed: ${String(err)}`);
+            }
+        }
+    };
+
+    // Start initial connection
+    await connect();
+
+    return { stop };
+}