fix: keep channel typing active during long inference (#25886, thanks @stakeswky)

Co-authored-by: stakeswky <stakeswky@users.noreply.github.com>

CHANGELOG.md

@@ -50,6 +50,7 @@ Docs: https://docs.openclaw.ai
 - Cron/Heartbeat delivery: stop inheriting cached session `lastThreadId` for heartbeat-mode target resolution unless a thread/topic is explicitly requested, so announce-mode cron and heartbeat deliveries stay on top-level destinations instead of leaking into active conversation threads. (#25730) Thanks @markshields-tl.
 - Heartbeat defaults/prompts: switch the implicit heartbeat delivery target from `last` to `none` (opt-in for external delivery), and use internal-only cron/exec heartbeat prompt wording when delivery is disabled so background checks do not nudge user-facing relay behavior. (#25871, #24638, #25851)
 - Auto-reply/Heartbeat queueing: drop heartbeat runs when a session already has an active run instead of enqueueing a stale followup, preventing duplicate heartbeat response branches after queue drain. (#25610, #25606) Thanks @mcaxtr.
+- Channels/Typing keepalive: refresh channel typing callbacks on a keepalive interval during long replies and clear keepalive timers on idle/cleanup across core + extension dispatcher callsites so typing indicators do not expire mid-inference. (#25886, #25882) Thanks @stakeswky.
 - Security/Sandbox media: restrict sandbox media tmp-path allowances to OpenClaw-managed tmp roots instead of broad host `os.tmpdir()` trust, and add outbound/channel guardrails (tmp-path lint + media-root smoke tests) to prevent regressions in local media attachment reads. Thanks @tdjackey for reporting.
 - Security/Sandbox media: reject hard-linked OpenClaw tmp media aliases (including symlink-to-hardlink chains) during sandbox media path resolution to prevent out-of-sandbox inode alias reads. (#25820) Thanks @bmendonca3.
 - Config/Plugins: treat stale removed `google-antigravity-auth` plugin references as compatibility warnings (not hard validation errors) across `plugins.entries`, `plugins.allow`, `plugins.deny`, and `plugins.slots.memory`, so startup no longer fails after antigravity removal. (#25538, #25862) Thanks @chilu18.

extensions/matrix/src/matrix/monitor/handler.ts

@@ -654,6 +654,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
           },
           onReplyStart: typingCallbacks.onReplyStart,
           onIdle: typingCallbacks.onIdle,
+          onCleanup: typingCallbacks.onCleanup,
         });
 
       const { queuedFinal, counts } = await core.channel.reply.dispatchReplyFromConfig({

extensions/mattermost/src/mattermost/monitor.ts

@@ -805,6 +805,8 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {}
           runtime.error?.(`mattermost ${info.kind} reply failed: ${String(err)}`);
         },
         onReplyStart: typingCallbacks.onReplyStart,
+        onIdle: typingCallbacks.onIdle,
+        onCleanup: typingCallbacks.onCleanup,
       });
 
     await core.channel.reply.dispatchReplyFromConfig({

extensions/msteams/src/reply-dispatcher.ts

@@ -122,6 +122,8 @@ export function createMSTeamsReplyDispatcher(params: {
         });
       },
       onReplyStart: typingCallbacks.onReplyStart,
+      onIdle: typingCallbacks.onIdle,
+      onCleanup: typingCallbacks.onCleanup,
     });
 
   return {

src/channels/channel-helpers.test.ts

@@ -190,4 +190,48 @@ describe("createTypingCallbacks", () => {
     expect(stop).toHaveBeenCalledTimes(1);
     expect(onStopError).toHaveBeenCalledTimes(1);
   });
+
+  it("sends typing keepalive pings until idle cleanup", async () => {
+    vi.useFakeTimers();
+    try {
+      const start = vi.fn().mockResolvedValue(undefined);
+      const stop = vi.fn().mockResolvedValue(undefined);
+      const onStartError = vi.fn();
+      const callbacks = createTypingCallbacks({ start, stop, onStartError });
+
+      await callbacks.onReplyStart();
+      expect(start).toHaveBeenCalledTimes(1);
+
+      await vi.advanceTimersByTimeAsync(2_999);
+      expect(start).toHaveBeenCalledTimes(1);
+
+      await vi.advanceTimersByTimeAsync(1);
+      expect(start).toHaveBeenCalledTimes(2);
+
+      await vi.advanceTimersByTimeAsync(3_000);
+      expect(start).toHaveBeenCalledTimes(3);
+
+      callbacks.onIdle?.();
+      await flushMicrotasks();
+      expect(stop).toHaveBeenCalledTimes(1);
+
+      await vi.advanceTimersByTimeAsync(9_000);
+      expect(start).toHaveBeenCalledTimes(3);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("deduplicates stop across idle and cleanup", async () => {
+    const start = vi.fn().mockResolvedValue(undefined);
+    const stop = vi.fn().mockResolvedValue(undefined);
+    const onStartError = vi.fn();
+    const callbacks = createTypingCallbacks({ start, stop, onStartError });
+
+    callbacks.onIdle?.();
+    callbacks.onCleanup?.();
+    await flushMicrotasks();
+
+    expect(stop).toHaveBeenCalledTimes(1);
+  });
 });

src/channels/typing.ts

@@ -10,9 +10,15 @@ export function createTypingCallbacks(params: {
   stop?: () => Promise<void>;
   onStartError: (err: unknown) => void;
   onStopError?: (err: unknown) => void;
+  keepaliveIntervalMs?: number;
 }): TypingCallbacks {
   const stop = params.stop;
-  const onReplyStart = async () => {
+  const keepaliveIntervalMs = params.keepaliveIntervalMs ?? 3_000;
+  let keepaliveTimer: ReturnType<typeof setInterval> | undefined;
+  let keepaliveStartInFlight = false;
+  let stopSent = false;
+
+  const fireStart = async () => {
     try {
       await params.start();
     } catch (err) {
@@ -20,11 +26,41 @@ export function createTypingCallbacks(params: {
     }
   };
 
-  const fireStop = stop
-    ? () => {
-        void stop().catch((err) => (params.onStopError ?? params.onStartError)(err));
+  const clearKeepalive = () => {
+    if (!keepaliveTimer) {
+      return;
+    }
+    clearInterval(keepaliveTimer);
+    keepaliveTimer = undefined;
+    keepaliveStartInFlight = false;
+  };
+
+  const onReplyStart = async () => {
+    stopSent = false;
+    clearKeepalive();
+    await fireStart();
+    if (keepaliveIntervalMs <= 0) {
+      return;
+    }
+    keepaliveTimer = setInterval(() => {
+      if (keepaliveStartInFlight) {
+        return;
       }
-    : undefined;
+      keepaliveStartInFlight = true;
+      void fireStart().finally(() => {
+        keepaliveStartInFlight = false;
+      });
+    }, keepaliveIntervalMs);
+  };
+
+  const fireStop = () => {
+    clearKeepalive();
+    if (!stop || stopSent) {
+      return;
+    }
+    stopSent = true;
+    void stop().catch((err) => (params.onStopError ?? params.onStartError)(err));
+  };
 
   return { onReplyStart, onIdle: fireStop, onCleanup: fireStop };
 }

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

@@ -669,6 +669,8 @@ export async function processDiscordMessage(ctx: DiscordMessagePreflightContext)
       await typingCallbacks.onReplyStart();
       await statusReactions.setThinking();
     },
+    onIdle: typingCallbacks.onIdle,
+    onCleanup: typingCallbacks.onCleanup,
   });
 
   let dispatchResult: Awaited<ReturnType<typeof dispatchInboundMessage>> | null = null;

src/signal/monitor/event-handler.ts

@@ -238,6 +238,8 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) {
         deps.runtime.error?.(danger(`signal ${info.kind} reply failed: ${String(err)}`));
       },
       onReplyStart: typingCallbacks.onReplyStart,
+      onIdle: typingCallbacks.onIdle,
+      onCleanup: typingCallbacks.onCleanup,
     });
 
     const { queuedFinal } = await dispatchInboundMessage({

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

@@ -306,6 +306,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
     },
     onReplyStart: typingCallbacks.onReplyStart,
     onIdle: typingCallbacks.onIdle,
+    onCleanup: typingCallbacks.onCleanup,
   });
 
   const draftStream = createSlackDraftStream({

src/telegram/bot-message-dispatch.ts

@@ -418,6 +418,18 @@ export const dispatchTelegramMessage = async ({
     void statusReactionController.setThinking();
   }
 
+  const typingCallbacks = createTypingCallbacks({
+    start: sendTyping,
+    onStartError: (err) => {
+      logTypingFailure({
+        log: logVerbose,
+        channel: "telegram",
+        target: String(chatId),
+        error: err,
+      });
+    },
+  });
+
   try {
     ({ queuedFinal } = await dispatchReplyWithBufferedBlockDispatcher({
       ctx: ctxPayload,
@@ -528,17 +540,9 @@ export const dispatchTelegramMessage = async ({
           deliveryState.markNonSilentFailure();
           runtime.error?.(danger(`telegram ${info.kind} reply failed: ${String(err)}`));
         },
-        onReplyStart: createTypingCallbacks({
-          start: sendTyping,
-          onStartError: (err) => {
-            logTypingFailure({
-              log: logVerbose,
-              channel: "telegram",
-              target: String(chatId),
-              error: err,
-            });
-          },
-        }).onReplyStart,
+        onReplyStart: typingCallbacks.onReplyStart,
+        onIdle: typingCallbacks.onIdle,
+        onCleanup: typingCallbacks.onCleanup,
       },
       replyOptions: {
         skillFilter,