fix(telegram): keep typing alive during active steered tasks (#116721)

extensions/telegram/src/bot-core.ts

@@ -46,6 +46,7 @@ import { apiThrottler, Bot, sequentialize, type ApiClientOptions } from "./bot.r
 import type { TelegramBotOptions } from "./bot.types.js";
 import { buildTelegramGroupPeerId } from "./bot/helpers.js";
 import { setTelegramCallbackQueryAnswerPromise } from "./callback-query-answer-state.js";
+import { TELEGRAM_CHAT_ACTION_INTERVAL_MS } from "./chat-action-timing.js";
 import {
   asTelegramClientFetch,
   createTelegramClientFetch,
@@ -77,8 +78,6 @@ const DEFAULT_TELEGRAM_BOT_RUNTIME: TelegramBotRuntime = {
   sequentialize,
   apiThrottler,
 };
-const TELEGRAM_TYPING_COALESCE_MS = 4_000;
-
 export function createTelegramBotCore(
   opts: TelegramBotOptions & { telegramDeps: TelegramBotDeps },
 ): TelegramBotInstance {
@@ -354,7 +353,7 @@ export function createTelegramBotCore(
     sendChatActionFn: (chatId, action, threadParams) =>
       bot.api.sendChatAction(chatId, action, threadParams),
     logger: (message) => logVerbose(`telegram: ${message}`),
-    minIntervalMs: TELEGRAM_TYPING_COALESCE_MS,
+    minIntervalMs: TELEGRAM_CHAT_ACTION_INTERVAL_MS,
   });
 
   const processMessage = createTelegramMessageProcessor({

extensions/telegram/src/bot-message-dispatch-turn.ts

@@ -19,6 +19,7 @@ import type { TelegramProgressController } from "./bot-message-dispatch-progress
 import type { TelegramReplyDelivery } from "./bot-message-dispatch-reply.js";
 import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js";
 import type { TelegramStreamMode } from "./bot/types.js";
+import { TELEGRAM_CHAT_ACTION_INTERVAL_MS } from "./chat-action-timing.js";
 import { beginTelegramInboundEventDeliveryCorrelation } from "./inbound-event-delivery.js";
 
 const TELEGRAM_MAX_CONSECUTIVE_TYPING_FAILURES = 5;
@@ -70,6 +71,10 @@ export async function runTelegramDispatchTurn(params: {
       accountId: context.route.accountId,
       typing: {
         start: context.sendTyping,
+        keepaliveIntervalMs: TELEGRAM_CHAT_ACTION_INTERVAL_MS,
+        // ReplyOperation owns terminal cleanup; a per-inbound TTL would kill
+        // feedback while the same long-running task is still active.
+        maxDurationMs: 0,
         maxConsecutiveFailures: TELEGRAM_MAX_CONSECUTIVE_TYPING_FAILURES,
         onStartError: (err) => {
           logTypingFailure({

extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts

@@ -11,6 +11,19 @@ import type { TelegramMessageContext } from "./bot-message-dispatch.test-harness
 import { notifyTelegramInboundEventOutboundSuccess } from "./inbound-event-delivery.js";
 
 describeTelegramDispatch("dispatchTelegramMessage pipeline-init", () => {
+  it("keeps Telegram typing below its client expiry without a per-message cutoff", async () => {
+    await dispatchWithContext({ context: createContext() });
+
+    expect(createChannelMessageReplyPipeline).toHaveBeenCalledWith(
+      expect.objectContaining({
+        typing: expect.objectContaining({
+          keepaliveIntervalMs: 4_000,
+          maxDurationMs: 0,
+        }),
+      }),
+    );
+  });
+
   it("cleans delivery correlation when reply-pipeline initialization fails", async () => {
     const sessionKey = "agent:main:telegram:direct:pipeline-init-failure";
     const statusReactionController = createStatusReactionController();

extensions/telegram/src/chat-action-timing.ts

@@ -0,0 +1,3 @@
+// Telegram typing expires after five seconds; renew before that without
+// fighting the account-scoped sendChatAction coalescing window.
+export const TELEGRAM_CHAT_ACTION_INTERVAL_MS = 4_000;

src/auto-reply/reply/agent-runner-run.ts

@@ -57,6 +57,7 @@ import { enqueueFollowupRun, type FollowupRun, scheduleFollowupDrain } from "./q
 import { createReplyMediaContext } from "./reply-media-paths.js";
 import { resolveReplyOperationRunState } from "./reply-operation-run-state.js";
 import { type ReplyOperation, replyRunRegistry } from "./reply-run-registry.js";
+import { bindReplyOperationTyping, refreshReplyOperationTyping } from "./reply-run-typing.js";
 import { createReplyToModeFilterForChannel, resolveReplyToMode } from "./reply-threading.js";
 import { admitReplyTurn, resolveReplyTurnKind } from "./reply-turn-admission.js";
 import {
@@ -331,6 +332,13 @@ export async function runReplyAgent(
       if (followupRun.currentInboundAudio === true) {
         activeReplyOperation?.markAcceptedSteeredInboundAudio();
       }
+      if (activeReplyOperation) {
+        // Steering joins the existing task; its dispatch-local controller is
+        // disposable, while the task-owned controller must keep its lifetime.
+        await refreshReplyOperationTyping(activeReplyOperation, {
+          startIfIdle: typingSignals.shouldStartImmediately,
+        });
+      }
       await touchActiveSessionEntry();
       typing.cleanup();
       return undefined;
@@ -557,6 +565,7 @@ export async function runReplyAgent(
       }
     }
   }
+  bindReplyOperationTyping(replyOperation, typing);
   let runFollowupTurn = queuedRunFollowupTurn;
   let shouldDrainQueuedFollowupsAfterClear = false;
   const returnWithQueuedFollowupDrain = <T>(value: T): T => {

src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts

@@ -42,6 +42,7 @@ import {
 } from "./reply-operation-run-state.js";
 import { createReplyOperation, type ReplyOperation } from "./reply-run-registry.js";
 import { testing as replyRunTesting } from "./reply-run-registry.test-support.js";
+import { bindReplyOperationTyping } from "./reply-run-typing.js";
 import { consumeReplyUsageState } from "./reply-usage-state.js";
 import { buildChannelSourceTurnId, setChannelSourceTurnId } from "./source-turn-id.js";
 import { createMockTypingController } from "./test-helpers.js";
@@ -437,6 +438,44 @@ function requireBuiltChannelSourceTurnId(
 }
 
 describe("runReplyAgent active steering", () => {
+  it("keeps the continuing Telegram task's typing alive after an accepted steer", async () => {
+    state.queueEmbeddedAgentMessageMock.mockReturnValueOnce(true);
+    const active = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session",
+      resetTriggered: false,
+    });
+    active.setPhase("running");
+    const taskTyping = createMockTypingController({ isActive: vi.fn(() => true) });
+    bindReplyOperationTyping(active, taskTyping);
+    const { run, typing } = createMinimalRun({
+      isActive: true,
+      isStreaming: true,
+      shouldSteer: true,
+      resolvedQueueMode: "steer",
+      sessionCtx: {
+        Provider: "telegram",
+        OriginatingChannel: "telegram",
+        OriginatingTo: "123",
+        NativeChannelId: "123",
+        MessageSid: "steer-telegram",
+      },
+      runOverrides: { agentId: "main", messageProvider: "telegram" },
+    });
+
+    await expect(run()).resolves.toBeUndefined();
+
+    expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
+    expect(taskTyping.startTypingLoop).toHaveBeenCalledOnce();
+    expect(taskTyping.refreshTypingTtl).toHaveBeenCalledOnce();
+    expect(taskTyping.cleanup).not.toHaveBeenCalled();
+    expect(typing.cleanup).toHaveBeenCalledOnce();
+
+    active.complete();
+
+    expect(taskTyping.cleanup).toHaveBeenCalledOnce();
+  });
+
   it("dispatches a declined steer once with its source-turn identity", async () => {
     const runState: ReplyOperationRunState = {};
     state.beforeAgentReplyHasHooksMock.mockImplementation(

src/auto-reply/reply/reply-run-typing.ts

@@ -0,0 +1,36 @@
+import { runAfterReplyOperationClear, type ReplyOperation } from "./reply-run-registry.js";
+import type { TypingController } from "./typing.js";
+
+const typingByReplyOperation = new WeakMap<ReplyOperation, TypingController>();
+
+/** Keep one feedback controller attached to the task that owns a reply run. */
+export function bindReplyOperationTyping(
+  operation: ReplyOperation,
+  typing: TypingController,
+): void {
+  if (typingByReplyOperation.has(operation)) {
+    return;
+  }
+  typingByReplyOperation.set(operation, typing);
+  runAfterReplyOperationClear(operation, () => {
+    if (typingByReplyOperation.get(operation) !== typing) {
+      return;
+    }
+    typingByReplyOperation.delete(operation);
+    typing.cleanup();
+  });
+}
+
+/** Refresh the continuing task's feedback after it adopts another inbound turn. */
+export async function refreshReplyOperationTyping(
+  operation: ReplyOperation,
+  options: { startIfIdle: boolean },
+): Promise<boolean> {
+  const typing = typingByReplyOperation.get(operation);
+  if (!typing || operation.result || (!options.startIfIdle && !typing.isActive())) {
+    return false;
+  }
+  await typing.startTypingLoop();
+  typing.refreshTypingTtl();
+  return true;
+}

src/channels/typing.test.ts

@@ -169,6 +169,56 @@ describe("createTypingCallbacks", () => {
     });
   });
 
+  it("preserves the existing keepalive cadence when an active reply starts again", async () => {
+    await withFakeTimers(async () => {
+      const { start, callbacks } = createTypingHarness({ keepaliveIntervalMs: 4_000 });
+
+      await callbacks.onReplyStart();
+      await vi.advanceTimersByTimeAsync(3_000);
+      await callbacks.onReplyStart();
+      expect(start).toHaveBeenCalledTimes(2);
+
+      await vi.advanceTimersByTimeAsync(1_000);
+
+      expect(start).toHaveBeenCalledTimes(3);
+    });
+  });
+
+  it("keeps coalesced typing alive beyond 60 seconds while the same task refreshes it", async () => {
+    await withFakeTimers(async () => {
+      vi.setSystemTime(0);
+      const acceptedStarts: number[] = [];
+      const { callbacks } = createTypingHarness({
+        keepaliveIntervalMs: 4_000,
+        maxDurationMs: 0,
+        start: async () => {
+          const now = Date.now();
+          const previous = acceptedStarts.at(-1);
+          if (previous !== undefined && now - previous < 4_000) {
+            return;
+          }
+          acceptedStarts.push(now);
+        },
+      });
+
+      await callbacks.onReplyStart();
+      for (let elapsedMs = 6_000; elapsedMs <= 132_000; elapsedMs += 6_000) {
+        await vi.advanceTimersByTimeAsync(6_000);
+        await callbacks.onReplyStart();
+      }
+
+      expect(acceptedStarts.at(-1)).toBeGreaterThan(120_000);
+      for (let index = 1; index < acceptedStarts.length; index += 1) {
+        expect(acceptedStarts[index]! - acceptedStarts[index - 1]!).toBeLessThanOrEqual(4_000);
+      }
+
+      callbacks.onIdle?.();
+      const countAtTaskCompletion = acceptedStarts.length;
+      await vi.advanceTimersByTimeAsync(12_000);
+      expect(acceptedStarts).toHaveLength(countAtTaskCompletion);
+    });
+  });
+
   it("stops keepalive after consecutive start failures", async () => {
     await withFakeTimers(async () => {
       const { start, onStartError, callbacks } = createTypingHarness({

src/channels/typing.ts

@@ -97,13 +97,15 @@ export function createTypingCallbacks(params: CreateTypingCallbacksParams): Typi
     }
     stopSent = false;
     startGuard.reset();
-    keepaliveLoop.stop();
     clearTtlTimer();
     const startPromise = fireStart();
     void startPromise.then(() => {
       if (closed || startGuard.isTripped()) {
         return;
       }
+      // Core can refresh an active reply independently of this channel loop.
+      // Restarting the interval here shifts its deadline and can outlive a
+      // provider's visible typing window between consecutive renewals.
       keepaliveLoop.start();
       startTtlTimer();
     });