From cbb920c7d9e4a6b7b2be395063ea1aa930c05938 Mon Sep 17 00:00:00 2001 From: Masato Hoshino Date: Sun, 5 Jul 2026 20:18:01 +0900 Subject: [PATCH] fix(qqbot): channel status keeps reporting connected after the gateway websocket dies (#100127) * fix(qqbot): publish disconnected channel status when the gateway closes or gives up The QQBot channel only ever set connected: true (onReady/onResumed); no close path updated the status, so a fatal close (bot banned or offline, 4914/4915) or reconnect exhaustion left channels.status claiming a live connection forever, and the channel-health monitor never saw connected=false for a dead gateway. Thread an onDisconnected callback from GatewayConnection.handleClose (fatal and pre-reconnect branches) and the reconnect-exhaustion path through the engine/bridge layers into channel.ts, which now records connected: false and, for fatal closes, the close reason as lastError. The running flag stays owned by the gateway lifecycle store, matching the sibling channel convention. Co-Authored-By: Claude Fable 5 * fix(qqbot): import ChannelAccountSnapshot from the channel-contract subpath The monolithic openclaw/plugin-sdk root entry is a legacy surface; plugin-sdk contract guardrails and extension boundary checks require focused subpath imports in bundled plugin sources. Co-Authored-By: Claude Fable 5 * fix(qqbot): ignore stale socket closes from superseded gateway connections A server-driven RECONNECT / INVALID_SESSION tears the old socket down and brings up a replacement; the old socket's close event can arrive after the replacement is live. Reacting to it again would tear down the new socket and regress the connected status, so the close handler now ignores closes from sockets that are no longer current. Co-Authored-By: Claude Fable 5 * fix(qqbot): harden disconnect status recovery * fix(qqbot): report opcode-driven reconnects * fix(qqbot): keep fatal disconnect health visible --------- Co-authored-by: Claude Fable 5 Co-authored-by: Vincent Koc --- extensions/qqbot/src/bridge/gateway.ts | 2 + .../qqbot/src/channel.gateway-status.test.ts | 116 ++++++++ extensions/qqbot/src/channel.ts | 15 + .../engine/gateway/gateway-connection.test.ts | 256 ++++++++++++++++++ .../src/engine/gateway/gateway-connection.ts | 27 ++ .../qqbot/src/engine/gateway/gateway.ts | 1 + extensions/qqbot/src/engine/gateway/types.ts | 6 + 7 files changed, 423 insertions(+) create mode 100644 extensions/qqbot/src/channel.gateway-status.test.ts create mode 100644 extensions/qqbot/src/engine/gateway/gateway-connection.test.ts diff --git a/extensions/qqbot/src/bridge/gateway.ts b/extensions/qqbot/src/bridge/gateway.ts index 171f5d1de06..5fa4c18985c 100644 --- a/extensions/qqbot/src/bridge/gateway.ts +++ b/extensions/qqbot/src/bridge/gateway.ts @@ -47,6 +47,7 @@ export interface GatewayContext { onReady?: (data: unknown) => void; onResumed?: (data: unknown) => void; onError?: (error: Error) => void; + onDisconnected?: (info: { reason?: string; fatal?: boolean }) => void; log?: { info: (msg: string) => void; error: (msg: string) => void; @@ -143,6 +144,7 @@ export async function startGateway(ctx: GatewayContext): Promise { onReady: ctx.onReady, onResumed: ctx.onResumed, onError: ctx.onError, + onDisconnected: ctx.onDisconnected, log: accountLogger, runtime, adapters: createEngineAdapters(), diff --git a/extensions/qqbot/src/channel.gateway-status.test.ts b/extensions/qqbot/src/channel.gateway-status.test.ts new file mode 100644 index 00000000000..315b1753332 --- /dev/null +++ b/extensions/qqbot/src/channel.gateway-status.test.ts @@ -0,0 +1,116 @@ +// Qqbot tests cover channel gateway status truth on disconnect. +import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { qqbotPlugin } from "./channel.js"; +import type { ResolvedQQBotAccount } from "./types.js"; + +const startGatewayMock = vi.hoisted(() => vi.fn()); + +vi.mock("./bridge/gateway.js", () => ({ + startGateway: startGatewayMock, +})); + +type StartGatewayOptions = { + onReady?: (data: unknown) => void; + onResumed?: (data: unknown) => void; + onError?: (error: Error) => void; + onDisconnected?: (info: { reason?: string; fatal?: boolean }) => void; +}; + +async function startAccountAndCaptureGatewayOptions() { + startGatewayMock.mockImplementation(() => new Promise(() => {})); + const statusWrites: ChannelAccountSnapshot[] = []; + let status: ChannelAccountSnapshot = { + accountId: "test-account", + running: true, + connected: false, + lastConnectedAt: null, + lastError: null, + }; + const account = { + accountId: "test-account", + appId: "test-app", + clientSecret: "test-secret", + enabled: true, + markdownSupport: false, + config: {}, + secretSource: "config", + } as unknown as ResolvedQQBotAccount; + const ctx = { + cfg: {}, + accountId: "test-account", + account, + runtime: {}, + abortSignal: new AbortController().signal, + getStatus: () => status, + setStatus: (next: ChannelAccountSnapshot) => { + status = next; + statusWrites.push(next); + }, + }; + const startAccount = qqbotPlugin.gateway?.startAccount; + expect(startAccount).toBeDefined(); + void startAccount?.(ctx as Parameters>[0]); + await vi.waitFor(() => { + expect(startGatewayMock).toHaveBeenCalled(); + }); + const options = startGatewayMock.mock.calls[0]?.[0] as StartGatewayOptions; + return { account, options, statusWrites, getStatus: () => status }; +} + +describe("qqbot channel gateway status", () => { + afterEach(() => { + vi.clearAllMocks(); + }); + + it("marks the account disconnected when the gateway reports a disconnect", async () => { + const { options, getStatus } = await startAccountAndCaptureGatewayOptions(); + + options.onReady?.({}); + expect(getStatus().connected).toBe(true); + + expect(options.onDisconnected).toBeDefined(); + options.onDisconnected?.({ reason: "close code 1006", fatal: false }); + expect(getStatus().connected).toBe(false); + expect(getStatus().running).toBe(true); + }); + + it("marks fatal disconnects unhealthy and records the close reason", async () => { + const { account, options, getStatus } = await startAccountAndCaptureGatewayOptions(); + + options.onReady?.({}); + options.onDisconnected?.({ reason: "banned", fatal: true }); + + expect(getStatus().connected).toBe(false); + // `running` is owned by the gateway lifecycle store: the account task + // stays held until an explicit stop/abort, so the plugin must not + // flip it here (a Start action would no-op against a held task). + expect(getStatus().running).toBe(true); + expect(getStatus().lastError).toBe("banned"); + + const publicStatus = await qqbotPlugin.status?.buildAccountSnapshot?.({ + account, + cfg: {}, + runtime: getStatus(), + }); + expect(publicStatus?.connected).toBe(false); + expect(publicStatus?.lastError).toBe("banned"); + }); + + it("clears fatal errors when the gateway becomes ready or resumes", async () => { + const { options, getStatus } = await startAccountAndCaptureGatewayOptions(); + + options.onReady?.({}); + options.onDisconnected?.({ reason: "banned", fatal: true }); + options.onResumed?.({}); + + expect(getStatus().connected).toBe(true); + expect(getStatus().lastError).toBeNull(); + + options.onDisconnected?.({ reason: "offline/sandbox-only", fatal: true }); + options.onReady?.({}); + + expect(getStatus().connected).toBe(true); + expect(getStatus().lastError).toBeNull(); + }); +}); diff --git a/extensions/qqbot/src/channel.ts b/extensions/qqbot/src/channel.ts index 5c3dc9a1dc6..fa453d43055 100644 --- a/extensions/qqbot/src/channel.ts +++ b/extensions/qqbot/src/channel.ts @@ -362,6 +362,7 @@ export const qqbotPlugin: ChannelPlugin = { running: true, connected: true, lastConnectedAt: Date.now(), + lastError: null, }); // Snapshot credentials so we can recover from the next hot // upgrade that might wipe openclaw.json mid-flight. @@ -374,6 +375,7 @@ export const qqbotPlugin: ChannelPlugin = { running: true, connected: true, lastConnectedAt: Date.now(), + lastError: null, }); persistAccountCredentialSnapshot(account); }, @@ -384,6 +386,19 @@ export const qqbotPlugin: ChannelPlugin = { lastError: error.message, }); }, + onDisconnected: ({ reason, fatal }) => { + log?.info( + `[qqbot:${account.accountId}] Gateway disconnected${reason ? `: ${reason}` : ""}`, + ); + // Keep the raw lifecycle snapshot truthful so readiness and the shared + // health monitor see the failed transport. QQBot's fatal flag only + // suppresses its immediate reconnect policy. + ctx.setStatus({ + ...ctx.getStatus(), + connected: false, + ...(fatal && reason ? { lastError: reason } : {}), + }); + }, }); }, logoutAccount: async ({ accountId, cfg }) => { diff --git a/extensions/qqbot/src/engine/gateway/gateway-connection.test.ts b/extensions/qqbot/src/engine/gateway/gateway-connection.test.ts new file mode 100644 index 00000000000..55388c3c8c6 --- /dev/null +++ b/extensions/qqbot/src/engine/gateway/gateway-connection.test.ts @@ -0,0 +1,256 @@ +// Qqbot tests cover gateway connection close/disconnect status behavior. +import { EventEmitter } from "node:events"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { EngineAdapters } from "../adapter/index.js"; +import { MAX_RECONNECT_ATTEMPTS } from "./constants.js"; +import { GatewayConnection } from "./gateway-connection.js"; +import type { GatewayAccount, GatewayPluginRuntime } from "./types.js"; + +const createQQWSClientMock = vi.hoisted(() => vi.fn()); + +vi.mock("./ws-client.js", () => ({ + createQQWSClient: createQQWSClientMock, +})); + +vi.mock("../messaging/sender.js", () => ({ + getAccessToken: vi.fn(async () => "test-token"), + getGatewayUrl: vi.fn(async () => "wss://mock-gateway"), + getPluginUserAgent: vi.fn(() => "test-agent"), + startBackgroundTokenRefresh: vi.fn(), + stopBackgroundTokenRefresh: vi.fn(), + clearTokenCache: vi.fn(), +})); + +vi.mock("../session/session-store.js", () => ({ + loadSession: vi.fn(() => undefined), + saveSession: vi.fn(), + clearSession: vi.fn(), +})); + +vi.mock("../session/known-users.js", () => ({ + recordKnownUser: vi.fn(), + flushKnownUsers: vi.fn(), +})); + +vi.mock("../ref/store.js", () => ({ + flushRefIndex: vi.fn(), +})); + +vi.mock("../commands/slash-command-handler.js", () => ({ + trySlashCommand: vi.fn(async () => "enqueue"), +})); + +class FakeWebSocket extends EventEmitter { + readyState = 3; // CLOSED — keeps cleanup() from re-entering close() + close = vi.fn(); + send = vi.fn(); +} + +function makeAccount(): GatewayAccount { + return { + accountId: "test-account", + appId: "test-app", + clientSecret: "test-secret", + markdownSupport: false, + config: {}, + }; +} + +async function startConnection(params: { onDisconnected?: (info: unknown) => void }) { + const ws = new FakeWebSocket(); + createQQWSClientMock.mockResolvedValue(ws); + const controller = new AbortController(); + const connection = new GatewayConnection({ + account: makeAccount(), + abortSignal: controller.signal, + cfg: {}, + runtime: {} as GatewayPluginRuntime, + adapters: {} as EngineAdapters, + handleMessage: async () => {}, + onDisconnected: params.onDisconnected, + }); + const started = connection.start(); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalled(); + }); + return { ws, controller, started }; +} + +describe("GatewayConnection disconnect status", () => { + beforeEach(() => { + vi.useFakeTimers(); + createQQWSClientMock.mockReset(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.clearAllMocks(); + }); + + it("reports a fatal disconnect when the close code says the bot is banned", async () => { + const onDisconnected = vi.fn(); + const { ws, controller, started } = await startConnection({ onDisconnected }); + + ws.emit("close", 4915, Buffer.from("")); + + expect(onDisconnected).toHaveBeenCalledWith({ reason: "banned", fatal: true }); + controller.abort(); + await started; + }); + + it("reports a non-fatal disconnect on a transient close before reconnecting", async () => { + const onDisconnected = vi.fn(); + const { ws, controller, started } = await startConnection({ onDisconnected }); + + ws.emit("close", 1006, Buffer.from("")); + + expect(onDisconnected).toHaveBeenCalledWith({ reason: "close code 1006", fatal: false }); + controller.abort(); + await started; + }); + + it("reports a fatal disconnect when reconnect attempts are exhausted", async () => { + const onDisconnected = vi.fn(); + const sockets = Array.from({ length: MAX_RECONNECT_ATTEMPTS + 1 }, () => new FakeWebSocket()); + let socketIndex = 0; + createQQWSClientMock.mockImplementation(async () => sockets[socketIndex++]); + const controller = new AbortController(); + const connection = new GatewayConnection({ + account: makeAccount(), + abortSignal: controller.signal, + cfg: {}, + runtime: {} as GatewayPluginRuntime, + adapters: {} as EngineAdapters, + handleMessage: async () => {}, + onDisconnected, + }); + const started = connection.start(); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalledTimes(1); + }); + + for (let attempt = 0; attempt < MAX_RECONNECT_ATTEMPTS; attempt++) { + sockets[attempt].emit("close", 1006, Buffer.from("")); + await vi.runOnlyPendingTimersAsync(); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalledTimes(attempt + 2); + }); + } + sockets[MAX_RECONNECT_ATTEMPTS].emit("close", 1006, Buffer.from("")); + + expect(onDisconnected).toHaveBeenCalledWith({ + reason: "reconnect attempts exhausted", + fatal: true, + }); + controller.abort(); + await started; + }); + + it("ignores a stale close from a superseded socket after a server-driven reconnect", async () => { + const onDisconnected = vi.fn(); + const staleWs = new FakeWebSocket(); + const replacementWs = new FakeWebSocket(); + createQQWSClientMock.mockResolvedValueOnce(staleWs).mockResolvedValueOnce(replacementWs); + const controller = new AbortController(); + const connection = new GatewayConnection({ + account: makeAccount(), + abortSignal: controller.signal, + cfg: {}, + runtime: {} as GatewayPluginRuntime, + adapters: {} as EngineAdapters, + handleMessage: async () => {}, + onDisconnected, + }); + const started = connection.start(); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalledTimes(1); + }); + + // Server asks for a reconnect: the old socket is torn down and a + // replacement is scheduled, then becomes live. + staleWs.emit("open"); + staleWs.emit("message", JSON.stringify({ op: 7 })); + expect(onDisconnected).toHaveBeenCalledWith({ + reason: "server requested reconnect", + fatal: false, + }); + await vi.advanceTimersByTimeAsync(1_100); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalledTimes(2); + }); + replacementWs.emit("open"); + + // The superseded socket's close arrives late; it must not regress + // the live replacement's status. + staleWs.emit("close", 1000, Buffer.from("")); + + expect(onDisconnected).toHaveBeenCalledTimes(1); + controller.abort(); + await started; + }); + + it("ignores a stale close while a server-driven reconnect is pending", async () => { + const onDisconnected = vi.fn(); + const staleWs = new FakeWebSocket(); + const replacementWs = new FakeWebSocket(); + createQQWSClientMock.mockResolvedValueOnce(staleWs).mockResolvedValueOnce(replacementWs); + const controller = new AbortController(); + const connection = new GatewayConnection({ + account: makeAccount(), + abortSignal: controller.signal, + cfg: {}, + runtime: {} as GatewayPluginRuntime, + adapters: {} as EngineAdapters, + handleMessage: async () => {}, + onDisconnected, + }); + const started = connection.start(); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalledTimes(1); + }); + + staleWs.emit("open"); + staleWs.emit("message", JSON.stringify({ op: 7 })); + expect(onDisconnected).toHaveBeenCalledWith({ + reason: "server requested reconnect", + fatal: false, + }); + staleWs.emit("close", 1006, Buffer.from("")); + + expect(onDisconnected).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1_100); + await vi.waitFor(() => { + expect(createQQWSClientMock).toHaveBeenCalledTimes(2); + }); + + controller.abort(); + await started; + }); + + it("reports a disconnect when the server invalidates the session", async () => { + const onDisconnected = vi.fn(); + const { ws, controller, started } = await startConnection({ onDisconnected }); + + ws.emit("open"); + ws.emit("message", JSON.stringify({ op: 9, d: false })); + + expect(onDisconnected).toHaveBeenCalledWith({ + reason: "session invalidated", + fatal: false, + }); + + controller.abort(); + await started; + }); + + it("does not report a disconnect for the close caused by an intentional abort", async () => { + const onDisconnected = vi.fn(); + const { ws, controller, started } = await startConnection({ onDisconnected }); + + controller.abort(); + ws.emit("close", 1000, Buffer.from("")); + + expect(onDisconnected).not.toHaveBeenCalled(); + await started; + }); +}); diff --git a/extensions/qqbot/src/engine/gateway/gateway-connection.ts b/extensions/qqbot/src/engine/gateway/gateway-connection.ts index 8e607541f1f..2a688382940 100644 --- a/extensions/qqbot/src/engine/gateway/gateway-connection.ts +++ b/extensions/qqbot/src/engine/gateway/gateway-connection.ts @@ -35,6 +35,7 @@ interface GatewayConnectionContext { onReady?: (data: unknown) => void; onResumed?: (data: unknown) => void; onError?: (error: Error) => void; + onDisconnected?: (info: { reason?: string; fatal?: boolean }) => void; handleMessage: (event: QueuedMessage) => Promise; onInteraction?: (event: InteractionEvent) => void; } @@ -132,6 +133,11 @@ export class GatewayConnection { const { account: _account, log } = this.ctx; if (this.isAborted || this.reconnect.isExhausted()) { log?.error(`Max reconnect attempts reached or aborted`); + // Exhaustion is a permanent give-up: report it as fatal so the + // channel status does not keep claiming a live connection. + if (!this.isAborted) { + this.ctx.onDisconnected?.({ reason: "reconnect attempts exhausted", fatal: true }); + } return; } if (this.reconnectTimer) { @@ -248,12 +254,20 @@ export class GatewayConnection { break; case GatewayOp.RECONNECT: + this.ctx.onDisconnected?.({ + reason: "server requested reconnect", + fatal: false, + }); this.cleanup(); this.scheduleReconnect(); break; case GatewayOp.INVALID_SESSION: { const canResume = d as boolean; + this.ctx.onDisconnected?.({ + reason: canResume ? "session resume rejected" : "session invalidated", + fatal: false, + }); if (!canResume) { this.sessionId = null; this.lastSeq = null; @@ -273,6 +287,12 @@ export class GatewayConnection { // ---- WebSocket: close ---- ws.on("close", (code, reason) => { log?.info(`WebSocket closed: ${code} ${reason.toString()}`); + // cleanup() clears currentWs before a server-driven reconnect. Ignore + // the old socket's delayed close both during that gap and after the + // replacement is live, or it can reschedule reconnect handling. + if (this.currentWs !== ws) { + return; + } this.isConnecting = false; this.handleClose(code); }); @@ -347,6 +367,13 @@ export class GatewayConnection { this.cleanup(); + // Publish the disconnect so channel status stops claiming a live + // connection; a fatal close (bot banned / offline) never reconnects. + // Abort-driven closes are an intentional stop, not a status change. + if (!this.isAborted) { + this.ctx.onDisconnected?.({ reason: action.reason, fatal: action.fatal }); + } + if (action.fatal) { return; } diff --git a/extensions/qqbot/src/engine/gateway/gateway.ts b/extensions/qqbot/src/engine/gateway/gateway.ts index 0750577ed82..5b99086f7fd 100644 --- a/extensions/qqbot/src/engine/gateway/gateway.ts +++ b/extensions/qqbot/src/engine/gateway/gateway.ts @@ -260,6 +260,7 @@ export async function startGateway(ctx: CoreGatewayContext): Promise { onReady: ctx.onReady, onResumed: ctx.onResumed, onError: ctx.onError, + onDisconnected: ctx.onDisconnected, onInteraction: handleInteraction, handleMessage, }); diff --git a/extensions/qqbot/src/engine/gateway/types.ts b/extensions/qqbot/src/engine/gateway/types.ts index 74faba9a1bc..612a72a4d83 100644 --- a/extensions/qqbot/src/engine/gateway/types.ts +++ b/extensions/qqbot/src/engine/gateway/types.ts @@ -221,6 +221,12 @@ export interface CoreGatewayContext { */ onResumed?: (data: unknown) => void; onError?: (error: Error) => void; + /** + * Invoked when the gateway websocket closes or permanently stops + * (fatal close code / reconnect attempts exhausted). Without this the + * channel status keeps reporting the last `connected: true` snapshot. + */ + onDisconnected?: (info: { reason?: string; fatal?: boolean }) => void; log?: EngineLogger; /** PluginRuntime injected by the framework — same object in both versions. */ runtime: GatewayPluginRuntime;