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;