fix(discord): avoid duplicate typing keepalive for tool replies (#84288)

dr00-eth · Andrew Meyer · web-flow · commit 94ab33036e90 · 2026-06-25T01:22:18.000+08:00
Co-authored-by: Andrew Meyer &lt;andrewmeyer@andrews-air.lan&gt;
diff --git a/extensions/discord/src/monitor/message-handler.process.test.ts b/extensions/discord/src/monitor/message-handler.process.test.ts
@@ -175,6 +175,7 @@ type DispatchInboundParams = {
     }) => Promise<void> | void;
     onReplyStart?: () => Promise<void> | void;
     sourceReplyDeliveryMode?: "automatic" | "message_tool_only";
+    typingKeepalive?: boolean;
     disableBlockStreaming?: boolean;
     suppressDefaultToolProgressMessages?: boolean;
     queuedDeliveryCorrelations?: Array<{ begin: () => () => void }>;
@@ -944,6 +945,7 @@ describe("processDiscordMessage ack reactions", () => {
     expect(replyTypingFeedback.onReplyStart).toHaveBeenCalledTimes(1);
     expect(replyTypingFeedback.onIdle).toHaveBeenCalledTimes(1);
     expect(replyTypingFeedback.onCleanup).toHaveBeenCalledTimes(1);
+    expect(getLastDispatchReplyOptions()?.typingKeepalive).toBe(false);
     expect(typingMocks.sendTyping).not.toHaveBeenCalled();
   });
 
@@ -984,6 +986,33 @@ describe("processDiscordMessage ack reactions", () => {
     }
   });
 
+  it("keeps one typing refresh loop for default message-tool replies", async () => {
+    vi.useFakeTimers();
+    try {
+      dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
+        await params?.replyOptions?.onReplyStart?.();
+        await vi.advanceTimersByTimeAsync(3_500);
+        return createNoQueuedDispatchResult();
+      });
+      const ctx = await createBaseContext({
+        shouldRequireMention: false,
+        effectiveWasMentioned: false,
+        cfg: {
+          messages: { groupChat: { visibleReplies: "message_tool" } },
+          session: { store: "/tmp/openclaw-discord-process-test-sessions.json" },
+        },
+        route: BASE_CHANNEL_ROUTE,
+      });
+
+      await runProcessDiscordMessage(ctx);
+
+      expect(getLastDispatchReplyOptions()?.typingKeepalive).toBe(false);
+      expect(typingMocks.sendTyping).toHaveBeenCalledTimes(2);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("debounces intermediate phase reactions and jumps to done for short runs", async () => {
     dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
       await params?.replyOptions?.onReasoningStream?.();
@@ -1532,6 +1561,7 @@ describe("processDiscordMessage session routing", () => {
 
     expectRecordFields(requireRecord(getLastDispatchReplyOptions(), "dispatch reply options"), {
       sourceReplyDeliveryMode: "message_tool_only",
+      typingKeepalive: false,
       disableBlockStreaming: true,
     });
     expect(createDiscordDraftStream).not.toHaveBeenCalled();
diff --git a/extensions/discord/src/monitor/message-handler.process.ts b/extensions/discord/src/monitor/message-handler.process.ts
@@ -251,6 +251,14 @@ async function processDiscordMessageInner(
     },
   });
   const sourceRepliesAreToolOnly = sourceReplyDeliveryMode === "message_tool_only";
+  const configuredTypingMode = cfg.session?.typingMode ?? cfg.agents?.defaults?.typingMode;
+  const configuredTypingInterval =
+    cfg.agents?.defaults?.typingIntervalSeconds ?? cfg.session?.typingIntervalSeconds;
+  const shouldDisableCoreTypingKeepalive =
+    Boolean(replyTypingFeedback) ||
+    (sourceRepliesAreToolOnly &&
+      configuredTypingMode === undefined &&
+      configuredTypingInterval === undefined);
   const ackReaction = resolveAckReaction(cfg, route.agentId, {
     channel: "discord",
     accountId,
@@ -460,6 +468,7 @@ async function processDiscordMessageInner(
       channelId: typingChannelId,
       rest: feedbackRest,
       log: logVerbose,
+      keepaliveIntervalMs: shouldDisableCoreTypingKeepalive ? undefined : 0,
     });
   if (replyTypingFeedback) {
     // A carried prestart only covers queue wait time; dispatch needs a fresh
@@ -955,6 +964,7 @@ async function processDiscordMessageInner(
         abortSignal,
         skillFilter: channelConfig?.skills,
         sourceReplyDeliveryMode,
+        typingKeepalive: shouldDisableCoreTypingKeepalive ? false : undefined,
         queuedDeliveryCorrelations: isRoomEvent ? [{ begin: beginDeliveryCorrelation }] : undefined,
         suppressTyping: isRoomEvent ? true : undefined,
         allowProgressCallbacksWhenSourceDeliverySuppressed:
diff --git a/extensions/discord/src/monitor/message-handler.queue.test.ts b/extensions/discord/src/monitor/message-handler.queue.test.ts
@@ -222,6 +222,34 @@ describe("createDiscordMessageHandler queue behavior", () => {
     );
   });
 
+  it("keeps the configured typing cadence for prestarted feedback", async () => {
+    preflightDiscordMessageMock.mockReset();
+    processDiscordMessageMock.mockReset();
+    preflightDiscordMessageMock.mockImplementation(async () =>
+      createAcceptedDmPreflightContext({
+        cfg: {
+          ...createPreflightContext().cfg,
+          agents: { defaults: { typingIntervalSeconds: 7 } },
+          session: { typingIntervalSeconds: 5 },
+        },
+      }),
+    );
+    processDiscordMessageMock.mockResolvedValue(undefined);
+    const replyTypingFeedback = createReplyTypingFeedbackMock("dm-1");
+    const createReplyTypingFeedback = vi.fn(() => replyTypingFeedback);
+
+    const handler = createDiscordMessageHandler({
+      ...createDiscordHandlerParams(),
+      testing: { createReplyTypingFeedback },
+    });
+    await handler(createMessageData("m-typing-cadence", "dm-1") as never, {} as never);
+    await flushQueueWork();
+
+    expect(createReplyTypingFeedback).toHaveBeenCalledWith(
+      expect.objectContaining({ keepaliveIntervalMs: 7_000 }),
+    );
+  });
+
   it("keeps accepted DM dispatch running when accepted typing feedback fails", async () => {
     preflightDiscordMessageMock.mockReset();
     processDiscordMessageMock.mockReset();
diff --git a/extensions/discord/src/monitor/message-handler.ts b/extensions/discord/src/monitor/message-handler.ts
@@ -3,6 +3,7 @@ import {
   createChannelInboundDebouncer,
   shouldDebounceTextInbound,
 } from "openclaw/plugin-sdk/channel-inbound";
+import { finiteSecondsToTimerSafeMilliseconds } from "openclaw/plugin-sdk/number-runtime";
 import { danger, logVerbose } from "openclaw/plugin-sdk/runtime-env";
 import { resolveOpenProviderRuntimeGroupPolicy } from "openclaw/plugin-sdk/runtime-group-policy";
 import type { Client } from "../internal/discord.js";
@@ -102,6 +103,9 @@ function startAcceptedTypingFeedback(params: {
       accountId: ctx.accountId,
       channelId: ctx.messageChannelId,
       log: logVerbose,
+      keepaliveIntervalMs: finiteSecondsToTimerSafeMilliseconds(
+        ctx.cfg.agents?.defaults?.typingIntervalSeconds ?? ctx.cfg.session?.typingIntervalSeconds,
+      ),
     });
   const cleanup = replyTypingFeedback.onCleanup;
   replyTypingFeedback.onCleanup = () => {
diff --git a/extensions/discord/src/monitor/reply-typing-feedback.ts b/extensions/discord/src/monitor/reply-typing-feedback.ts
@@ -24,6 +24,7 @@ export function createDiscordReplyTypingFeedback(params: {
   rest?: RequestClient;
   log: (message: string) => void;
   maxDurationMs?: number;
+  keepaliveIntervalMs?: number;
 }): DiscordReplyTypingFeedback {
   let channelId = params.channelId;
   const rest =
@@ -44,6 +45,7 @@ export function createDiscordReplyTypingFeedback(params: {
           error: err,
         });
       },
+      keepaliveIntervalMs: params.keepaliveIntervalMs,
       maxDurationMs: params.maxDurationMs ?? DISCORD_REPLY_TYPING_MAX_DURATION_MS,
     });
   const updateChannelId = (nextChannelId: string) => {
diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts
@@ -1,8 +1,8 @@
+import type { FastMode } from "@openclaw/normalization-core/string-coerce";
 /** Public option types for reply generation callbacks, streaming, and delivery policy. */
 import type { ImageContent } from "../llm/types.js";
 import type { PromptImageOrderEntry } from "../media/prompt-image-order.js";
 import type { UserTurnTranscriptRecorder } from "../sessions/user-turn-transcript.types.js";
-import type { FastMode } from "@openclaw/normalization-core/string-coerce";
 import type { ReplyPayload } from "./reply-payload.js";
 import type { TypingController } from "./reply/typing.js";
 
@@ -73,6 +73,8 @@ export type GetReplyOptions = {
   /** Called when the typing controller cleans up (e.g., run ended with NO_REPLY). */
   onTypingCleanup?: () => void;
   onTypingController?: (typing: TypingController) => void;
+  /** If false, send only the initial typing signal without periodic keepalive refreshes. */
+  typingKeepalive?: boolean;
   isHeartbeat?: boolean;
   /** Policy-level typing control for run classes (user/system/internal/heartbeat). */
   typingPolicy?: TypingPolicy;
diff --git a/src/auto-reply/reply/get-reply.ts b/src/auto-reply/reply/get-reply.ts
@@ -389,6 +389,7 @@ export async function getReplyFromConfig(
       onReplyStart: opts?.onReplyStart,
       onCleanup: opts?.onTypingCleanup,
       typingIntervalSeconds,
+      keepalive: opts?.typingKeepalive ?? true,
       silentToken: SILENT_REPLY_TOKEN,
       log: defaultRuntime.log,
     });
diff --git a/src/auto-reply/reply/reply-utils.test.ts b/src/auto-reply/reply/reply-utils.test.ts
@@ -483,6 +483,27 @@ describe("typing controller", () => {
     await vi.advanceTimersByTimeAsync(5_000);
     expect(onReplyStart).toHaveBeenCalledTimes(1);
   });
+
+  it("can send the first typing signal without periodic keepalive refreshes", async () => {
+    vi.useFakeTimers();
+    const onReplyStart = vi.fn();
+    const typing = createTypingController({
+      onReplyStart,
+      typingIntervalSeconds: 1,
+      typingTtlMs: 30_000,
+      keepalive: false,
+    });
+
+    await typing.startTypingLoop();
+    expect(onReplyStart).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(5_000);
+    expect(onReplyStart).toHaveBeenCalledTimes(1);
+
+    await typing.startTypingLoop();
+    await vi.advanceTimersByTimeAsync(5_000);
+    expect(onReplyStart).toHaveBeenCalledTimes(1);
+  });
 });
 
 describe("resolveTypingMode", () => {
@@ -530,6 +551,17 @@ describe("resolveTypingMode", () => {
         },
         expected: "message",
       },
+      {
+        name: "configured instant typing mode wins over message-tool-only default",
+        input: {
+          configured: "instant" as const,
+          isGroupChat: true,
+          wasMentioned: false,
+          isHeartbeat: false,
+          sourceReplyDeliveryMode: "message_tool_only" as const,
+        },
+        expected: "instant",
+      },
       {
         name: "default mentioned group chat",
         input: {
diff --git a/src/auto-reply/reply/typing.ts b/src/auto-reply/reply/typing.ts
@@ -39,10 +39,17 @@ export function createTypingController(params: {
   onCleanup?: () => void;
   typingIntervalSeconds?: number;
   typingTtlMs?: number;
+  keepalive?: boolean;
   silentToken?: string;
   log?: (message: string) => void;
 }): TypingController {
-  const { onReplyStart, onCleanup, silentToken = SILENT_REPLY_TOKEN, log } = params;
+  const {
+    onReplyStart,
+    onCleanup,
+    keepalive = true,
+    silentToken = SILENT_REPLY_TOKEN,
+    log,
+  } = params;
   if (!onReplyStart && !onCleanup) {
     return {
       onReplyStart: async () => {},
@@ -202,6 +209,10 @@ export function createTypingController(params: {
     if (!onReplyStart) {
       return;
     }
+    if (!keepalive) {
+      await ensureStart();
+      return;
+    }
     if (typingLoop.isRunning()) {
       return;
     }
