From f03dc9785e48812ed2e7a5194f8370e59ea7d754 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 02:14:19 -0700 Subject: [PATCH] fix(telegram): keep typing alive during active steered tasks (#116721) --- extensions/telegram/src/bot-core.ts | 5 +- .../telegram/src/bot-message-dispatch-turn.ts | 5 ++ ...bot-message-dispatch.pipeline-init.test.ts | 13 +++++ extensions/telegram/src/chat-action-timing.ts | 3 ++ src/auto-reply/reply/agent-runner-run.ts | 9 ++++ .../agent-runner.runreplyagent.e2e.test.ts | 39 +++++++++++++++ src/auto-reply/reply/reply-run-typing.ts | 36 +++++++++++++ src/channels/typing.test.ts | 50 +++++++++++++++++++ src/channels/typing.ts | 4 +- 9 files changed, 160 insertions(+), 4 deletions(-) create mode 100644 extensions/telegram/src/chat-action-timing.ts create mode 100644 src/auto-reply/reply/reply-run-typing.ts diff --git a/extensions/telegram/src/bot-core.ts b/extensions/telegram/src/bot-core.ts index c0cf86f3c5f1..e3cb36cf0b71 100644 --- a/extensions/telegram/src/bot-core.ts +++ b/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({ diff --git a/extensions/telegram/src/bot-message-dispatch-turn.ts b/extensions/telegram/src/bot-message-dispatch-turn.ts index dba60b3a4af5..b2bc25178bd1 100644 --- a/extensions/telegram/src/bot-message-dispatch-turn.ts +++ b/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({ diff --git a/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts b/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts index 1b7d3267d197..24c4159c1b38 100644 --- a/extensions/telegram/src/bot-message-dispatch.pipeline-init.test.ts +++ b/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(); diff --git a/extensions/telegram/src/chat-action-timing.ts b/extensions/telegram/src/chat-action-timing.ts new file mode 100644 index 000000000000..e782c6481bb4 --- /dev/null +++ b/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; diff --git a/src/auto-reply/reply/agent-runner-run.ts b/src/auto-reply/reply/agent-runner-run.ts index 3e03df5ef454..39617776aacd 100644 --- a/src/auto-reply/reply/agent-runner-run.ts +++ b/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 = (value: T): T => { diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index cfe5fefaa52e..01a286de97f0 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/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( diff --git a/src/auto-reply/reply/reply-run-typing.ts b/src/auto-reply/reply/reply-run-typing.ts new file mode 100644 index 000000000000..f2ca73edb83f --- /dev/null +++ b/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(); + +/** 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 { + const typing = typingByReplyOperation.get(operation); + if (!typing || operation.result || (!options.startIfIdle && !typing.isActive())) { + return false; + } + await typing.startTypingLoop(); + typing.refreshTypingTtl(); + return true; +} diff --git a/src/channels/typing.test.ts b/src/channels/typing.test.ts index 7a0bd9424e93..58797915bfec 100644 --- a/src/channels/typing.test.ts +++ b/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({ diff --git a/src/channels/typing.ts b/src/channels/typing.ts index 784a98d9e533..91e0951baa62 100644 --- a/src/channels/typing.ts +++ b/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(); });