refactor: centralize ack reaction removal

extensions/bluebubbles/src/monitor.test.ts

@@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import type { IncomingMessage, ServerResponse } from "node:http";
 import { EventEmitter } from "node:events";
 
-import { shouldAckReaction } from "clawdbot/plugin-sdk";
+import { removeAckReactionAfterReply, shouldAckReaction } from "clawdbot/plugin-sdk";
 import type { ClawdbotConfig, PluginRuntime } from "clawdbot/plugin-sdk";
 import {
   handleBlueBubblesWebhookRequest,
@@ -138,6 +138,7 @@ function createMockRuntime(): PluginRuntime {
       },
       reactions: {
         shouldAckReaction,
+        removeAckReactionAfterReply,
       },
       groups: {
         resolveGroupPolicy: mockResolveGroupPolicy as unknown as PluginRuntime["channel"]["groups"]["resolveGroupPolicy"],

extensions/bluebubbles/src/monitor.ts

@@ -1750,29 +1750,26 @@ async function processMessage(
       },
     });
   } finally {
-    if (
+    if (sentMessage && chatGuidForActions && ackMessageId) {
-      removeAckAfterReply &&
+      core.channel.reactions.removeAckReactionAfterReply({
-      sentMessage &&
+        removeAfterReply: removeAckAfterReply,
-      ackReactionPromise &&
+        ackReactionPromise,
-      ackReactionValue &&
+        ackReactionValue: ackReactionValue ?? null,
-      chatGuidForActions &&
+        remove: () =>
-      ackMessageId
+          sendBlueBubblesReaction({
-    ) {
+            chatGuid: chatGuidForActions,
-      void ackReactionPromise.then((didAck) => {
+            messageGuid: ackMessageId,
-        if (!didAck) return;
+            emoji: ackReactionValue ?? "",
-        sendBlueBubblesReaction({
+            remove: true,
-          chatGuid: chatGuidForActions,
+            opts: { cfg: config, accountId: account.accountId },
-          messageGuid: ackMessageId,
+          }),
-          emoji: ackReactionValue,
+        onError: (err) => {
-          remove: true,
-          opts: { cfg: config, accountId: account.accountId },
-        }).catch((err) => {
           logVerbose(
             core,
             runtime,
             `ack reaction removal failed chatGuid=${chatGuidForActions} msg=${ackMessageId}: ${String(err)}`,
           );
-        });
+        },
       });
     }
     if (chatGuidForActions && baseUrl && password && !sentMessage) {

src/channels/ack-reactions.test.ts

@@ -1,6 +1,10 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 
-import { shouldAckReaction, shouldAckReactionForWhatsApp } from "./ack-reactions.js";
+import {
+  removeAckReactionAfterReply,
+  shouldAckReaction,
+  shouldAckReactionForWhatsApp,
+} from "./ack-reactions.js";
 
 describe("shouldAckReaction", () => {
   it("honors direct and group-all scopes", () => {
@@ -222,3 +226,44 @@ describe("shouldAckReactionForWhatsApp", () => {
     ).toBe(false);
   });
 });
+
+describe("removeAckReactionAfterReply", () => {
+  it("removes only when ack succeeded", async () => {
+    const remove = vi.fn().mockResolvedValue(undefined);
+    const onError = vi.fn();
+    removeAckReactionAfterReply({
+      removeAfterReply: true,
+      ackReactionPromise: Promise.resolve(true),
+      ackReactionValue: "👀",
+      remove,
+      onError,
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(remove).toHaveBeenCalledTimes(1);
+    expect(onError).not.toHaveBeenCalled();
+  });
+
+  it("skips removal when ack did not happen", async () => {
+    const remove = vi.fn().mockResolvedValue(undefined);
+    removeAckReactionAfterReply({
+      removeAfterReply: true,
+      ackReactionPromise: Promise.resolve(false),
+      ackReactionValue: "👀",
+      remove,
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(remove).not.toHaveBeenCalled();
+  });
+
+  it("skips when not configured", async () => {
+    const remove = vi.fn().mockResolvedValue(undefined);
+    removeAckReactionAfterReply({
+      removeAfterReply: false,
+      ackReactionPromise: Promise.resolve(true),
+      ackReactionValue: "👀",
+      remove,
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(remove).not.toHaveBeenCalled();
+  });
+});

src/channels/ack-reactions.ts

@@ -53,3 +53,19 @@ export function shouldAckReactionForWhatsApp(params: {
     shouldBypassMention: params.groupActivated,
   });
 }
+
+export function removeAckReactionAfterReply(params: {
+  removeAfterReply: boolean;
+  ackReactionPromise: Promise<boolean> | null;
+  ackReactionValue: string | null;
+  remove: () => Promise<void>;
+  onError?: (err: unknown) => void;
+}) {
+  if (!params.removeAfterReply) return;
+  if (!params.ackReactionPromise) return;
+  if (!params.ackReactionValue) return;
+  void params.ackReactionPromise.then((didAck) => {
+    if (!didAck) return;
+    params.remove().catch((err) => params.onError?.(err));
+  });
+}

src/discord/monitor/message-handler.process.ts

@@ -8,7 +8,10 @@ import {
   extractShortModelName,
   type ResponsePrefixContext,
 } from "../../auto-reply/reply/response-prefix-template.js";
-import { shouldAckReaction as shouldAckReactionGate } from "../../channels/ack-reactions.js";
+import {
+  removeAckReactionAfterReply,
+  shouldAckReaction as shouldAckReactionGate,
+} from "../../channels/ack-reactions.js";
 import {
   formatInboundEnvelope,
   formatThreadStarterEnvelope,
@@ -394,19 +397,18 @@ export async function processDiscordMessage(ctx: DiscordMessagePreflightContext)
       `discord: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${replyTarget}`,
     );
   }
-  if (removeAckAfterReply && ackReactionPromise && ackReaction) {
+  removeAckReactionAfterReply({
-    const ackReactionValue = ackReaction;
+    removeAfterReply: removeAckAfterReply,
-    void ackReactionPromise.then((didAck) => {
+    ackReactionPromise,
-      if (!didAck) return;
+    ackReactionValue: ackReaction,
-      removeReactionDiscord(message.channelId, message.id, ackReactionValue, {
+    remove: () =>
-        rest: client.rest,
+      removeReactionDiscord(message.channelId, message.id, ackReaction, { rest: client.rest }),
-      }).catch((err) => {
+    onError: (err) => {
-        logVerbose(
+      logVerbose(
-          `discord: failed to remove ack reaction from ${message.channelId}/${message.id}: ${String(err)}`,
+        `discord: failed to remove ack reaction from ${message.channelId}/${message.id}: ${String(err)}`,
-        );
+      );
-      });
+    },
-    });
+  });
-  }
   if (isGuildMessage && historyLimit > 0) {
     clearHistoryEntries({
       historyMap: guildHistories,

src/plugin-sdk/index.ts

@@ -122,7 +122,11 @@ export type {
   AckReactionScope,
   WhatsAppAckReactionMode,
 } from "../channels/ack-reactions.js";
-export { shouldAckReaction, shouldAckReactionForWhatsApp } from "../channels/ack-reactions.js";
+export {
+  removeAckReactionAfterReply,
+  shouldAckReaction,
+  shouldAckReactionForWhatsApp,
+} from "../channels/ack-reactions.js";
 export { resolveChannelMediaMaxBytes } from "../channels/plugins/media-limits.js";
 export type { NormalizedLocation } from "../channels/location.js";
 export { formatLocationText, toLocationContext } from "../channels/location.js";

src/plugins/runtime/index.ts

@@ -25,7 +25,7 @@ import { resolveEffectiveMessagesConfig, resolveHumanDelayConfig } from "../../a
 import { createMemoryGetTool, createMemorySearchTool } from "../../agents/tools/memory-tool.js";
 import { handleSlackAction } from "../../agents/tools/slack-actions.js";
 import { handleWhatsAppAction } from "../../agents/tools/whatsapp-actions.js";
-import { shouldAckReaction } from "../../channels/ack-reactions.js";
+import { removeAckReactionAfterReply, shouldAckReaction } from "../../channels/ack-reactions.js";
 import { resolveCommandAuthorizedFromAuthorizers } from "../../channels/command-gating.js";
 import { discordMessageActions } from "../../channels/plugins/actions/discord.js";
 import { telegramMessageActions } from "../../channels/plugins/actions/telegram.js";
@@ -201,6 +201,7 @@ export function createPluginRuntime(): PluginRuntime {
       },
       reactions: {
         shouldAckReaction,
+        removeAckReactionAfterReply,
       },
       groups: {
         resolveGroupPolicy: resolveChannelGroupPolicy,

src/plugins/runtime/types.ts

@@ -20,6 +20,8 @@ type BuildMentionRegexes = typeof import("../../auto-reply/reply/mentions.js").b
 type MatchesMentionPatterns =
   typeof import("../../auto-reply/reply/mentions.js").matchesMentionPatterns;
 type ShouldAckReaction = typeof import("../../channels/ack-reactions.js").shouldAckReaction;
+type RemoveAckReactionAfterReply =
+  typeof import("../../channels/ack-reactions.js").removeAckReactionAfterReply;
 type ResolveChannelGroupPolicy =
   typeof import("../../config/group-policy.js").resolveChannelGroupPolicy;
 type ResolveChannelGroupRequireMention =
@@ -214,6 +216,7 @@ export type PluginRuntime = {
     };
     reactions: {
       shouldAckReaction: ShouldAckReaction;
+      removeAckReactionAfterReply: RemoveAckReactionAfterReply;
     };
     groups: {
       resolveGroupPolicy: ResolveChannelGroupPolicy;

src/slack/monitor/message-handler/dispatch.ts

@@ -9,6 +9,7 @@ import {
 } from "../../../auto-reply/reply/response-prefix-template.js";
 import { dispatchInboundMessage } from "../../../auto-reply/dispatch.js";
 import { clearHistoryEntries } from "../../../auto-reply/reply/history.js";
+import { removeAckReactionAfterReply } from "../../../channels/ack-reactions.js";
 import { createReplyDispatcherWithTyping } from "../../../auto-reply/reply/reply-dispatcher.js";
 import { resolveStorePath, updateLastRoute } from "../../../config/sessions.js";
 import { danger, logVerbose, shouldLogVerbose } from "../../../globals.js";
@@ -152,21 +153,26 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
     );
   }
 
-  if (ctx.removeAckAfterReply && prepared.ackReactionPromise && prepared.ackReactionMessageTs) {
+  removeAckReactionAfterReply({
-    const messageTs = prepared.ackReactionMessageTs;
+    removeAfterReply: ctx.removeAckAfterReply,
-    const ackValue = prepared.ackReactionValue;
+    ackReactionPromise: prepared.ackReactionPromise,
-    void prepared.ackReactionPromise.then((didAck) => {
+    ackReactionValue: prepared.ackReactionValue,
-      if (!didAck) return;
+    remove: () =>
-      removeSlackReaction(message.channel, messageTs, ackValue, {
+      removeSlackReaction(
-        token: ctx.botToken,
+        message.channel,
-        client: ctx.app.client,
+        prepared.ackReactionMessageTs ?? "",
-      }).catch((err) => {
+        prepared.ackReactionValue,
-        logVerbose(
+        {
-          `slack: failed to remove ack reaction from ${message.channel}/${message.ts}: ${String(err)}`,
+          token: ctx.botToken,
-        );
+          client: ctx.app.client,
-      });
+        },
-    });
+      ),
-  }
+    onError: (err) => {
+      logVerbose(
+        `slack: failed to remove ack reaction from ${message.channel}/${message.ts}: ${String(err)}`,
+      );
+    },
+  });
 
   if (prepared.isRoomish && ctx.historyLimit > 0) {
     clearHistoryEntries({

src/telegram/bot-message-dispatch.ts

@@ -7,6 +7,7 @@ import {
 import { EmbeddedBlockChunker } from "../agents/pi-embedded-block-chunker.js";
 import { clearHistoryEntries } from "../auto-reply/reply/history.js";
 import { dispatchReplyWithBufferedBlockDispatcher } from "../auto-reply/reply/provider-dispatcher.js";
+import { removeAckReactionAfterReply } from "../channels/ack-reactions.js";
 import { danger, logVerbose } from "../globals.js";
 import { resolveMarkdownTableMode } from "../config/markdown-tables.js";
 import { deliverReplies } from "./bot/delivery.js";
@@ -184,16 +185,18 @@ export const dispatchTelegramMessage = async ({
     }
     return;
   }
-  if (removeAckAfterReply && ackReactionPromise && msg.message_id && reactionApi) {
+  removeAckReactionAfterReply({
-    void ackReactionPromise.then((didAck) => {
+    removeAfterReply: removeAckAfterReply,
-      if (!didAck) return;
+    ackReactionPromise,
-      reactionApi(chatId, msg.message_id, []).catch((err) => {
+    ackReactionValue: ackReactionPromise ? "ack" : null,
-        logVerbose(
+    remove: () => reactionApi?.(chatId, msg.message_id ?? 0, []) ?? Promise.resolve(),
-          `telegram: failed to remove ack reaction from ${chatId}/${msg.message_id}: ${String(err)}`,
+    onError: (err) => {
-        );
+      if (!msg.message_id) return;
-      });
+      logVerbose(
-    });
+        `telegram: failed to remove ack reaction from ${chatId}/${msg.message_id}: ${String(err)}`,
-  }
+      );
+    },
+  });
   if (isGroup && historyKey && historyLimit > 0) {
     clearHistoryEntries({ historyMap: groupHistories, historyKey });
   }