Fix WebSocket connection accumulation across gateway restarts

Three listener leaks caused handlers to pile up on each in-process reload:
- notification.added_to_channel was registered in StreamChatClientRuntime.start()
  but never removed; stop() now saves the reference and calls client.off() on it.
- ai_indicator.stop was registered in startAccount but handleAbort only removed
  message.new; extracted to handleAiStop so it can be cleaned up properly.
- Added module-level activeGatewayCleanup registry: startAccount force-invokes
  any stale cleanup for the same accountId before opening a new connection,
  self-healing the case where the framework reloads without calling stop() first.
- handleAbort is now idempotent via a `stopped` boolean guard.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: steookk

CLAUDE.md

@@ -152,3 +152,5 @@ Config supports a flat default account or named sub-accounts:
 - **`safeSendEvent` swallows errors.** Indicator events are best-effort; a failed `ai_indicator` update must not abort message delivery. Retries: 5 attempts, exponential backoff starting at 100 ms, only on 429/5xx.
 - **`seenThreads` is process-scoped.** The `Set<string>` tracking "first message in thread" lives at module level, so it persists across gateway reloads until the process restarts. This is intentional — it avoids re-sending parent context for active threads after a config reload.
 - **`onTextChunk` receives deltas despite the wire protocol using full text.** `onPartialReply` provides cumulative text; `channel.ts` extracts the delta before calling `onTextChunk`. Inside `StreamingHandler`, `onTextChunk` re-accumulates deltas into `accumulatedText` and passes that full string to `partialUpdateMessage`. The round-trip is: cumulative → delta → cumulative. The delta extraction exists because `StreamingHandler` was designed around the "streaming chunks" mental model — it owns the accumulation and the throttle counter, making that API feel natural. The redundancy is intentional for architectural clarity, not a bug.
+- **`activeGatewayCleanup` is a module-level registry for defence against connection accumulation.** `startAccount` stores a cleanup function per `accountId`. On the next `startAccount` call for the same account, any existing entry is force-invoked before creating the new connection. This self-heals the case where OpenClaw's in-process gateway reload calls `startAccount` without calling `stop()` first, which would otherwise leave orphaned WebSocket connections accumulating (each receiving every `message.new` event). `handleAbort` is idempotent via a `stopped` boolean guard, so it is safe to call from both the abort signal and `stop()`.
+- **All three SDK event listeners are removed on cleanup.** `handleAbort` explicitly calls `client.off()` for `message.new` and `ai_indicator.stop`; `chatRuntime.stop()` removes `notification.added_to_channel` (saved as `addedToChannelHandler` in `StreamChatClientRuntime`). Failing to remove any one of these would cause listener accumulation across restarts.

src/channel.ts

@@ -27,6 +27,11 @@ import {
 // Track which threads we've already seen (for first-in-thread detection)
 const seenThreads = new Set<string>();
 
+// Module-level registry of active gateway cleanup functions keyed by accountId.
+// Allows startAccount to force-stop a stale connection if the framework calls
+// startAccount again without having called stop() first (e.g. in-process reloads).
+const activeGatewayCleanup = new Map<string, () => void>();
+
 // ---------------------------------------------------------------------------
 // Reactions helper
 // ---------------------------------------------------------------------------
@@ -471,6 +476,16 @@ export const streamchatPlugin: StreamChatChannelPlugin = {
         );
       }
 
+      // Force-stop any stale runtime for this accountId that was never cleaned up
+      // (can happen when the framework does an in-process reload without calling stop()).
+      const staleCleanup = activeGatewayCleanup.get(accountId);
+      if (staleCleanup) {
+        log?.warn?.(
+          `[StreamChat] Stale connection detected for account "${accountId}" — forcing cleanup before restart`,
+        );
+        staleCleanup();
+      }
+
       const chatRuntime = new StreamChatClientRuntime(account, log);
       const runContexts = new RunContextMap();
       const streamingHandler = new StreamingHandler({
@@ -508,15 +523,10 @@ export const streamchatPlugin: StreamChatChannelPlugin = {
         });
       };
 
-      client.on("message.new", handleMessage);
-
       // Listen for force stop from client
-      client.on("ai_indicator.stop" as "user.watching.start", (event: Event) => {
-        // Find the run associated with this message
+      const handleAiStop = (event: Event) => {
         const messageId = (event as unknown as Record<string, unknown>).message_id as string | undefined;
         if (!messageId) return;
-
-        // Look through active streams to find the matching run
         const activeRun = runContexts.findByResponseMessageId(messageId);
         if (activeRun) {
           streamingHandler.onForceStop(activeRun.runId).catch((err) => {
@@ -525,11 +535,19 @@ export const streamchatPlugin: StreamChatChannelPlugin = {
             );
           });
         }
-      });
+      };
 
-      // Handle abort signal
+      client.on("message.new", handleMessage);
+      client.on("ai_indicator.stop" as "user.watching.start", handleAiStop);
+
+      // Handle abort signal / explicit stop — idempotent via `stopped` guard
+      let stopped = false;
       const handleAbort = () => {
+        if (stopped) return;
+        stopped = true;
         client.off("message.new", handleMessage);
+        client.off("ai_indicator.stop" as "user.watching.start", handleAiStop);
+        activeGatewayCleanup.delete(accountId);
         chatRuntime.stop().catch((err) => {
           log?.error?.(
             `[StreamChat] Disconnect error: ${String(err)}`,
@@ -542,6 +560,8 @@ export const streamchatPlugin: StreamChatChannelPlugin = {
         });
       };
 
+      activeGatewayCleanup.set(accountId, handleAbort);
+
       if (abortSignal) {
         abortSignal.addEventListener("abort", handleAbort, { once: true });
       }

src/stream-chat-runtime.ts

@@ -9,6 +9,7 @@ export class StreamChatClientRuntime {
   private account: ResolvedAccount;
   private log?: ChannelLogSink;
   private connected = false;
+  private addedToChannelHandler?: (event: Event) => void;
 
   constructor(account: ResolvedAccount, log?: ChannelLogSink) {
     this.account = account;
@@ -48,7 +49,7 @@ export class StreamChatClientRuntime {
     );
 
     // Auto-watch new channels the bot is added to
-    this.client.on("notification.added_to_channel", (event: Event) => {
+    this.addedToChannelHandler = (event: Event) => {
       if (event.channel) {
         const ch = this.client.channel(
           event.channel.type,
@@ -66,11 +67,16 @@ export class StreamChatClientRuntime {
           );
         });
       }
-    });
+    };
+    this.client.on("notification.added_to_channel", this.addedToChannelHandler);
   }
 
   async stop(): Promise<void> {
     if (this.connected) {
+      if (this.addedToChannelHandler) {
+        this.client.off("notification.added_to_channel", this.addedToChannelHandler);
+        this.addedToChannelHandler = undefined;
+      }
       this.log?.info?.(`[StreamChat] Disconnecting...`);
       await this.client.disconnectUser();
       this.connected = false;