fix: cancel Buzz turns during reconnect

extensions/buzz/src/buzz-bus.lifecycle.test.ts

@@ -108,7 +108,7 @@ vi.mock("nostr-tools", async (importOriginal) => {
   };
 });
 
-import { sendBuzzTextOneShot, startBuzzBus } from "./buzz-bus.js";
+import { sendBuzzTextOneShot, startBuzzBus, type BuzzBus } from "./buzz-bus.js";
 import {
   BUZZ_DIFF_MESSAGE_KIND,
   BUZZ_INBOUND_MESSAGE_KINDS,
@@ -493,6 +493,49 @@ describe("Buzz bus lifecycle", () => {
     expect(relayMocks.close).toHaveBeenCalledOnce();
   });
 
+  it("aborts active inbound dispatch before waiting for bus shutdown", async () => {
+    relayMocks.auth.mockResolvedValue("ok");
+    relayMocks.roomHistoryEvents = [
+      {
+        id: "d".repeat(64),
+        kind: 9,
+        pubkey: SENDER_PUBLIC_KEY,
+        created_at: 1_700_000_000,
+        content: "historical message",
+        sig: "e".repeat(128),
+        tags: [["h", CHANNEL_ID]],
+      },
+    ];
+    let dispatchSignal: AbortSignal | undefined;
+    const onMessage = vi.fn(
+      async (_message: BuzzInboundMessage, _bus: BuzzBus, signal: AbortSignal) => {
+        dispatchSignal = signal;
+        await new Promise<void>((resolve) => {
+          if (signal.aborted) {
+            resolve();
+            return;
+          }
+          signal.addEventListener("abort", () => resolve(), { once: true });
+        });
+      },
+    );
+    const bus = await startBuzzBus({
+      accountId: ACCOUNT_ID,
+      relayUrl: "wss://buzz.example.com",
+      privateKey: PRIVATE_KEY,
+      channelIds: [CHANNEL_ID],
+      onMessage,
+    });
+
+    await vi.waitFor(() => expect(onMessage).toHaveBeenCalledOnce());
+    expect(dispatchSignal?.aborted).toBe(false);
+
+    await bus.close();
+
+    expect(dispatchSignal?.aborted).toBe(true);
+    expect(relayMocks.close).toHaveBeenCalledOnce();
+  });
+
   it("leaves room subscription shutdown to the relay", async () => {
     relayMocks.auth.mockResolvedValue("ok");
     const bus = await startBuzzBus({

extensions/buzz/src/buzz-bus.ts

@@ -510,7 +510,7 @@ export async function startBuzzBus(options: {
   authTag?: string;
   channelIds: string[];
   since?: number;
-  onMessage: (message: BuzzInboundMessage, bus: BuzzBus) => Promise<void>;
+  onMessage: (message: BuzzInboundMessage, bus: BuzzBus, signal: AbortSignal) => Promise<void>;
   onMessageError?: (error: Error) => void;
   onFatalError?: (error: Error) => void;
   onDedupeError?: (error: Error) => void;
@@ -578,6 +578,7 @@ export async function startBuzzBus(options: {
       await directoryRelay?.refreshRooms(options.channelIds);
     },
     sendText: async ({ channelId, text, threadId, replyToId }) => {
+      signal.throwIfAborted();
       const event = buildBuzzTextEvent({ secretKey, channelId, text, threadId, replyToId });
       await relay.publish(event);
       return event.id;
@@ -597,10 +598,10 @@ export async function startBuzzBus(options: {
       await relay.send(JSON.stringify(["EVENT", event]));
     },
     close: async () => {
-      // A replacement bus must not overlap handlers admitted by this relay
-      // generation. Let those handlers finish while their reply socket is live.
-      await dispatchQueue.close();
       lifecycleAbort.abort(new Error("Buzz bus closed"));
+      // Agent turns receive this generation signal. Abort them before draining
+      // so reconnect and Gateway shutdown do not wait on stale work.
+      await dispatchQueue.close();
       stopPresenceHeartbeat();
       directoryRelay?.close();
       replayGuard.clearMemory();
@@ -630,7 +631,7 @@ export async function startBuzzBus(options: {
         // only by active workers, so replay cannot create unbounded in-flight state.
         const admission = dispatchQueue.enqueue(async () => {
           await replayGuard.processGuarded(event, async () => {
-            await options.onMessage(message, bus);
+            await options.onMessage(message, bus, signal);
           });
         });
         if (admission === "overflow") {
@@ -722,8 +723,8 @@ export async function startBuzzBus(options: {
   } catch (error) {
     // Every failed startup must release the socket before ownership returns to
     // the gateway-level reconnect loop.
-    await dispatchQueue.close();
     lifecycleAbort.abort(error);
+    await dispatchQueue.close();
     directoryRelay?.close();
     relay.close();
     throw error;

extensions/buzz/src/gateway.lifecycle.test.ts

@@ -10,7 +10,11 @@ const gatewayMocks = vi.hoisted(() => ({
   busSendTyping: vi.fn(async () => undefined),
   sendBuzzTextOneShot: vi.fn(async () => "standalone-event-id"),
   onMessage: undefined as
-    | ((message: import("./message-event.js").BuzzInboundMessage, bus: BuzzBus) => Promise<void>)
+    | ((
+        message: import("./message-event.js").BuzzInboundMessage,
+        bus: BuzzBus,
+        signal: AbortSignal,
+      ) => Promise<void>)
     | undefined,
   onMessageError: undefined as ((error: Error) => void) | undefined,
   onFatalError: undefined as ((error: Error) => void) | undefined,
@@ -83,6 +87,7 @@ describe("Buzz gateway lifecycle", () => {
         onMessage: (
           message: import("./message-event.js").BuzzInboundMessage,
           bus: BuzzBus,
+          signal: AbortSignal,
         ) => Promise<void>;
         onMessageError?: (error: Error) => void;
         onFatalError?: (error: Error) => void;

extensions/buzz/src/gateway.ts

@@ -97,12 +97,12 @@ export async function startBuzzGatewayAccount(ctx: ChannelGatewayContext<Resolve
         channelIds,
         since: sessionSince,
         signal: ctx.abortSignal,
-        onMessage: async (message, sessionBus) => {
+        onMessage: async (message, sessionBus, signal) => {
           // Subscription filters reduce traffic, but relay events remain untrusted.
           if (!isConfiguredBuzzChannel(configuredChannelIds, message.channelId)) {
             return;
           }
-          await handleBuzzInbound({ account, cfg: ctx.cfg, bus: sessionBus, message });
+          await handleBuzzInbound({ account, cfg: ctx.cfg, bus: sessionBus, message, signal });
         },
         onMessageError: (error) => {
           ctx.log?.error?.(`[${account.accountId}] Buzz message failed: ${error.message}`);

extensions/buzz/src/inbound.test.ts

@@ -55,6 +55,10 @@ function createMessage(overrides: Partial<BuzzInboundMessage> = {}): BuzzInbound
   };
 }
 
+function createSignal(): AbortSignal {
+  return new AbortController().signal;
+}
+
 function createBus(): BuzzBus {
   return {
     publicKey: BOT_PUBLIC_KEY,
@@ -88,15 +92,18 @@ describe("handleBuzzInbound", () => {
   it("accepts a native Nostr public-key mention", async () => {
     const runtime = createPluginRuntimeMock();
     setBuzzRuntime(runtime);
+    const signal = createSignal();
 
     await handleBuzzInbound({
       account: createAccount(),
       cfg: {} satisfies OpenClawConfig,
       bus: createBus(),
       message: createMessage({ mentionedPubkeys: [BOT_PUBLIC_KEY] }),
+      signal,
     });
 
     expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(1);
+    expect(firstDispatch(runtime).replyOptions?.abortSignal).toBe(signal);
     expect(firstDispatch(runtime).ctxPayload).toMatchObject({
       WasMentioned: true,
       SenderId: SENDER_PUBLIC_KEY,
@@ -158,6 +165,7 @@ describe("handleBuzzInbound", () => {
       cfg: {} satisfies OpenClawConfig,
       bus,
       message: createMessage(),
+      signal: createSignal(),
     });
 
     expect(firstDispatch(runtime).ctxPayload).toMatchObject({
@@ -178,6 +186,7 @@ describe("handleBuzzInbound", () => {
       cfg: {} satisfies OpenClawConfig,
       bus: createBus(),
       message: createMessage({ text: "@openclaw status" }),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(1);
@@ -193,6 +202,7 @@ describe("handleBuzzInbound", () => {
       cfg: {} satisfies OpenClawConfig,
       bus: createBus(),
       message: createMessage(),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled();
@@ -210,6 +220,7 @@ describe("handleBuzzInbound", () => {
       cfg: {} satisfies OpenClawConfig,
       bus: createBus(),
       message: createMessage({ mentionedPubkeys: [BOT_PUBLIC_KEY] }),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled();
@@ -231,6 +242,7 @@ describe("handleBuzzInbound", () => {
       message: createMessage({
         text: "/status",
       }),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(1);
@@ -251,6 +263,7 @@ describe("handleBuzzInbound", () => {
       cfg: {} satisfies OpenClawConfig,
       bus: createBus(),
       message: createMessage({ text: "/status" }),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled();
@@ -270,6 +283,7 @@ describe("handleBuzzInbound", () => {
         threadId: "event-root",
         mentionedPubkeys: [BOT_PUBLIC_KEY],
       }),
+      signal: createSignal(),
     });
 
     const dispatch = firstDispatch(runtime);
@@ -327,6 +341,7 @@ describe("handleBuzzInbound", () => {
           truncated: true,
         },
       }),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.commands.shouldComputeCommandAuthorized).not.toHaveBeenCalled();
@@ -365,6 +380,7 @@ describe("handleBuzzInbound", () => {
           truncated: false,
         },
       }),
+      signal: createSignal(),
     });
 
     expect(runtime.channel.mentions.matchesMentionPatterns).not.toHaveBeenCalled();
@@ -380,6 +396,7 @@ describe("handleBuzzInbound", () => {
       cfg: {} satisfies OpenClawConfig,
       bus: createBus(),
       message: createMessage({ mentionedPubkeys: [BOT_PUBLIC_KEY] }),
+      signal: createSignal(),
     });
 
     const dispatch = firstDispatch(runtime);

extensions/buzz/src/inbound.ts

@@ -22,9 +22,10 @@ export async function handleBuzzInbound(params: {
   cfg: OpenClawConfig;
   bus: BuzzBus;
   message: BuzzInboundMessage;
+  signal: AbortSignal;
 }) {
   const runtime = getBuzzRuntime();
-  const { account, cfg, bus, message } = params;
+  const { account, cfg, bus, message, signal } = params;
   const channelId = parseBuzzTarget(message.channelId);
   const target = buildBuzzTarget(channelId);
   const textForAgent = formatBuzzMessageForAgent(message);
@@ -161,6 +162,9 @@ export async function handleBuzzInbound(params: {
         throw error instanceof Error ? error : new Error(String(error));
       },
     },
+    replyOptions: {
+      abortSignal: signal,
+    },
     replyPipeline: {
       typing: {
         start: async () => {