From 2a55c7c59a0a8a11c8830232c2d2d8e90944c07d Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Fri, 10 Jul 2026 13:21:32 +0530 Subject: [PATCH] fix(telegram): complete spooled updates at turn adoption, rescope ingress watchdog (cherry picked from commit 77c84de5109a42d5a53b5a0fa4c1d9b87ae12bbd) --- extensions/telegram/AGENTS.md | 9 + .../telegram/src/bot-message-dispatch.ts | 4 + extensions/telegram/src/bot-message.test.ts | 26 ++- extensions/telegram/src/bot-message.ts | 131 +++++++---- .../telegram/src/polling-session.test.ts | 214 +++++++++++++++++- extensions/telegram/src/polling-session.ts | 26 ++- 6 files changed, 353 insertions(+), 57 deletions(-) diff --git a/extensions/telegram/AGENTS.md b/extensions/telegram/AGENTS.md index ad91963bb257..97bfcda18630 100644 --- a/extensions/telegram/AGENTS.md +++ b/extensions/telegram/AGENTS.md @@ -25,6 +25,15 @@ Verified against Telegram Bot API 10.1, July 1 2026. - Never swallow inbound processing errors. A transient store error on a spooled replay must record a `failed-retryable` processing result; a swallowed throw acks the update as completed and deletes the message. +- Spool completes at turn adoption, not settle. Once the recovery-relevant + session/run state is durably persisted (`restartRecoveryDeliveryContext` + + run id), the spooled row tombstones via `complete()` and the per-chat lane + frees. Run health after that is owned by run lifecycle / main-session + restart recovery — not by the ingress spool. Pre-adoption timeout + (`ISOLATED_INGRESS_ADOPTION_STALL_MS`, default 5 minutes, overridable via + `OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS`) is the only ingress + guillotine; it dead-letters with `handler-timeout` when claim→adoption + stalls. Healthy long turns must not be killed by the spool watchdog. - No per-message full-store writes. Hot-path SQLite writes are per-entry. Rewriting a cache on every send or read stalls the event loop, and that stall masquerades as a polling stall (the sent-message-cache regression). diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index 6bdacea1de44..a2da555a97fc 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/extensions/telegram/src/bot-message-dispatch.ts @@ -246,6 +246,8 @@ type DispatchTelegramMessageParams = { opts: Pick; retryDispatchErrors?: boolean; suppressFailureFallback?: boolean; + /** Fires after recovery-relevant session/run state is durably persisted. */ + onTurnAdopted?: () => void | Promise; }; type TelegramDispatchResult = { kind: "completed" } | { kind: "failed-retryable"; error: unknown }; @@ -785,6 +787,7 @@ export const dispatchTelegramMessage = async ({ opts, retryDispatchErrors = false, suppressFailureFallback = false, + onTurnAdopted, }: DispatchTelegramMessageParams): Promise => { const dispatchStartedAt = Date.now(); const dispatchContext = resolveDispatchTelegramContext({ context }); @@ -2613,6 +2616,7 @@ export const dispatchTelegramMessage = async ({ skillFilter, disableBlockStreaming, abortSignal: replyAbortController.signal, + ...(onTurnAdopted ? { onTurnAdopted } : {}), sourceReplyDeliveryMode: isRoomEvent ? "message_tool_only" : undefined, queuedDeliveryCorrelations: isRoomEvent ? [{ begin: beginDeliveryCorrelation }] diff --git a/extensions/telegram/src/bot-message.test.ts b/extensions/telegram/src/bot-message.test.ts index 6f4c974d3121..2cb5e78841f3 100644 --- a/extensions/telegram/src/bot-message.test.ts +++ b/extensions/telegram/src/bot-message.test.ts @@ -31,13 +31,13 @@ vi.mock("./bot-message-dispatch.js", () => ({ let createTelegramMessageProcessor: typeof import("./bot-message.js").createTelegramMessageProcessor; let formatTelegramInboundLogLine: typeof import("./bot-message.js").formatTelegramInboundLogLine; let runWithTelegramUpdateProcessingFrame: typeof import("./bot-processing-outcome.js").runWithTelegramUpdateProcessingFrame; -let withTelegramSpooledReplayUpdate: typeof import("./bot-processing-outcome.js").withTelegramSpooledReplayUpdate; +let runWithTelegramSpooledReplayUpdate: typeof import("./bot-processing-outcome.js").runWithTelegramSpooledReplayUpdate; describe("telegram bot message processor", () => { beforeAll(async () => { ({ createTelegramMessageProcessor, formatTelegramInboundLogLine } = await import("./bot-message.js")); - ({ runWithTelegramUpdateProcessingFrame, withTelegramSpooledReplayUpdate } = + ({ runWithTelegramUpdateProcessingFrame, runWithTelegramSpooledReplayUpdate } = await import("./bot-processing-outcome.js")); }); @@ -289,11 +289,17 @@ describe("telegram bot message processor", () => { sendMessage, ); const update = { update_id: 123456 }; - const result = await withTelegramSpooledReplayUpdate(update, async () => + // Spooled agent turns detach at adoption: processMessage returns once the + // deferred participant is registered; the real result is on deferred.task. + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => processSampleMessage(processMessage, undefined, { update }), ); - - expect(result).toEqual({ kind: "failed-retryable", error: dispatchError }); + expect(replay.value).toEqual({ kind: "completed" }); + expect(replay.deferredWork).toBeDefined(); + await expect(replay.deferredWork!.task).resolves.toEqual({ + kind: "failed-retryable", + error: dispatchError, + }); expect(sendMessage).not.toHaveBeenCalled(); expect(runtimeError).toHaveBeenCalledWith( "telegram message processing failed: Error: dispatch exploded", @@ -359,11 +365,15 @@ describe("telegram bot message processor", () => { runtime: { error: runtimeError }, } as unknown as Parameters[0]); const update = { update_id: 123457 }; - const result = await withTelegramSpooledReplayUpdate(update, async () => + const replay = await runWithTelegramSpooledReplayUpdate(update, async () => processSampleMessage(processMessage, undefined, { update }), ); - - expect(result).toEqual({ kind: "failed-retryable", error: dispatchError }); + expect(replay.value).toEqual({ kind: "completed" }); + expect(replay.deferredWork).toBeDefined(); + await expect(replay.deferredWork!.task).resolves.toEqual({ + kind: "failed-retryable", + error: dispatchError, + }); expect(sendMessage).not.toHaveBeenCalled(); expect(dispatchTelegramMessage).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/extensions/telegram/src/bot-message.ts b/extensions/telegram/src/bot-message.ts index a854f5e7a440..f22f9a112ad8 100644 --- a/extensions/telegram/src/bot-message.ts +++ b/extensions/telegram/src/bot-message.ts @@ -18,6 +18,8 @@ import type { TelegramMessageContextOptions } from "./bot-message-context.types. import type { TelegramPromptContextEntry } from "./bot-message-context.types.js"; import { dispatchTelegramMessage } from "./bot-message-dispatch.js"; import { + createTelegramSpooledReplayDeferredParticipant, + getTelegramSpooledReplayDeferredParticipant, isTelegramSpooledReplayUpdate, recordTelegramMessageProcessingResult, type TelegramMessageProcessingResult, @@ -202,55 +204,102 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep await lifecycle?.onDispatchStart?.(); const spooledReplay = options?.spooledReplay === true || isTelegramSpooledReplayUpdate(primaryCtx.update); - try { - const dispatchResult = await dispatchTelegramMessage({ - context, - bot, - cfg, - runtime, - replyToMode, - streamMode, - textLimit, - telegramCfg, - telegramDeps, - opts, - retryDispatchErrors: spooledReplay, - suppressFailureFallback: spooledReplay, - }); - if (dispatchResult?.kind === "failed-retryable") { + const runDispatch = async (params: { + onTurnAdopted?: () => void | Promise; + }): Promise => { + try { + const dispatchResult = await dispatchTelegramMessage({ + context, + bot, + cfg, + runtime, + replyToMode, + streamMode, + textLimit, + telegramCfg, + telegramDeps, + opts, + retryDispatchErrors: spooledReplay, + suppressFailureFallback: spooledReplay, + onTurnAdopted: params.onTurnAdopted, + }); + if (dispatchResult?.kind === "failed-retryable") { + const result: TelegramMessageProcessingResult = { + kind: "failed-retryable", + error: dispatchResult.error, + }; + recordCurrentUpdateProcessingResult(result); + return result; + } + if (ingressDebugEnabled && ingressReceivedAtMs) { + logVerbose( + `telegram ingress: chatId=${context.chatId} dispatchCompleteMs=${Date.now() - ingressReceivedAtMs}` + + (options?.ingressBuffer ? ` buffer=${options.ingressBuffer}` : ""), + ); + } + const result: TelegramMessageProcessingResult = { kind: "completed" }; + recordCurrentUpdateProcessingResult(result); + return result; + } catch (err) { + runtime.error?.(danger(`telegram message processing failed: ${String(err)}`)); + if (!spooledReplay) { + try { + await bot.api.sendMessage( + context.chatId, + "Something went wrong while processing your request. Please try again.", + buildTelegramThreadParams(context.threadSpec), + ); + } catch {} + } const result: TelegramMessageProcessingResult = { kind: "failed-retryable", - error: dispatchResult.error, + error: err, }; recordCurrentUpdateProcessingResult(result); return result; } - if (ingressDebugEnabled && ingressReceivedAtMs) { - logVerbose( - `telegram ingress: chatId=${context.chatId} dispatchCompleteMs=${Date.now() - ingressReceivedAtMs}` + - (options?.ingressBuffer ? ` buffer=${options.ingressBuffer}` : ""), + }; + + // Spooled ingress: complete the spool row at turn adoption (recovery state + // persisted), not settle. The deferred participant hands ownership back to + // the spool drain so the per-chat lane frees while the agent turn continues. + if (spooledReplay) { + const existingParticipant = getTelegramSpooledReplayDeferredParticipant(); + const participant = + existingParticipant ?? + createTelegramSpooledReplayDeferredParticipant( + `agent-turn:${context.chatId}:${context.ctxPayload.MessageSid ?? Date.now()}`, ); + if (participant) { + let adopted = false; + const settleIfNeeded = (result: TelegramMessageProcessingResult) => { + if (adopted) { + return; + } + participant.settle(result); + }; + const run = async () => { + const result = await runDispatch({ + onTurnAdopted: async () => { + if (adopted) { + return; + } + adopted = true; + participant.settle({ kind: "completed" }); + }, + }); + settleIfNeeded(result); + return result; + }; + if (existingParticipant) { + return await run(); + } + void run(); + const detached: TelegramMessageProcessingResult = { kind: "completed" }; + return detached; } - const result: TelegramMessageProcessingResult = { kind: "completed" }; - recordCurrentUpdateProcessingResult(result); - return result; - } catch (err) { - runtime.error?.(danger(`telegram message processing failed: ${String(err)}`)); - if (!spooledReplay) { - try { - await bot.api.sendMessage( - context.chatId, - "Something went wrong while processing your request. Please try again.", - buildTelegramThreadParams(context.threadSpec), - ); - } catch {} - } - const result: TelegramMessageProcessingResult = { - kind: "failed-retryable", - error: err, - }; - recordCurrentUpdateProcessingResult(result); - return result; } + + return await runDispatch({}); }; }; diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index f3caecf76592..919151a84918 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -2086,8 +2086,220 @@ describe("TelegramPollingSession", () => { await vi.waitFor(async () => expect(await failedUpdateIds(tempDir)).toEqual([42])); expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]); - expectLogIncludes(log, "buffered processing timed out behind update 42"); + expectLogIncludes(log, "pre-adoption timed out behind update 42"); expectLogExcludes(log, "spooled update 42 failed; keeping for retry"); + expect(await failedUpdateReasons(tempDir)).toEqual([{ id: 42, reason: "handler-timeout" }]); + abort.abort(); + stopWorker(); + await runPromise; + }); + }); + + it("completes spooled row at adoption while a long turn is still settling (healthy long turn)", async () => { + await withTempSpool(async (tempDir) => { + const abort = new AbortController(); + const log = vi.fn(); + const participants: TelegramSpooledReplayDeferredParticipant[] = []; + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "healthy long turn")]); + + const { runPromise, stopWorker } = startIsolatedIngressSession({ + abort, + spoolDir: tempDir, + log, + drainIntervalMs: 10, + spooledUpdateHandlerTimeoutMs: 80, + handleUpdate: async (update) => { + const participant = createTelegramSpooledReplayDeferredParticipant( + `test-adopt:${update.update_id}`, + ); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + participants.push(participant); + // Return immediately (deferred registered). Adoption settles the + // spool row; the agent turn would continue under run lifecycle. + queueMicrotask(() => { + participant.settle({ kind: "completed" }); + }); + }, + }); + + await vi.waitFor(() => expect(participants).toHaveLength(1)); + await vi.waitFor(async () => + expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]), + ); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + expect(await failedUpdateIds(tempDir)).toEqual([]); + + // Past the handler/adoption timeout after adoption: no dead-letter. + await new Promise((resolve) => setTimeout(resolve, 150)); + expect(await failedUpdateIds(tempDir)).toEqual([]); + expectLogExcludes(log, "timed out"); + expectLogExcludes(log, "handler-timeout"); + + abort.abort(); + stopWorker(); + await runPromise; + }); + }); + + it("replays claimed spooled updates after a crash before adoption", async () => { + await withTempSpool(async (tempDir) => { + const abort = new AbortController(); + const participants: TelegramSpooledReplayDeferredParticipant[] = []; + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "pre-adoption crash")]); + + const { runPromise, stopWorker } = startIsolatedIngressSession({ + abort, + spoolDir: tempDir, + drainIntervalMs: 10, + handleUpdate: async (update) => { + const participant = createTelegramSpooledReplayDeferredParticipant( + `test-pre-crash:${update.update_id}`, + ); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + participants.push(participant); + // Never adopt: process dies with claim held. + }, + }); + + await vi.waitFor(() => expect(participants).toHaveLength(1)); + await vi.waitFor(async () => + expect( + (await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).map((c) => c.updateId), + ).toEqual([42]), + ); + + abort.abort(); + stopWorker(); + await runPromise; + + // Stale-claim recovery after crash: row is still claimed → replayable. + const recovered = await recoverStaleTelegramSpooledUpdateClaims({ + spoolDir: tempDir, + staleMs: 0, + }); + expect(recovered).toBeGreaterThanOrEqual(1); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([42]); + expect(await failedUpdateIds(tempDir)).toEqual([]); + }); + }); + + it("does not replay spooled updates after crash post-adoption (row already tombstoned)", async () => { + await withTempSpool(async (tempDir) => { + const abort = new AbortController(); + const participants: TelegramSpooledReplayDeferredParticipant[] = []; + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "post-adoption crash")]); + + const { runPromise, stopWorker } = startIsolatedIngressSession({ + abort, + spoolDir: tempDir, + drainIntervalMs: 10, + handleUpdate: async (update) => { + const participant = createTelegramSpooledReplayDeferredParticipant( + `test-post-crash:${update.update_id}`, + ); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + participants.push(participant); + participant.settle({ kind: "completed" }); + // Turn would continue under run lifecycle; process crash after this is fine. + }, + }); + + await vi.waitFor(() => expect(participants).toHaveLength(1)); + await vi.waitFor(async () => + expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]), + ); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + + abort.abort(); + stopWorker(); + await runPromise; + + const recovered = await recoverStaleTelegramSpooledUpdateClaims({ + spoolDir: tempDir, + staleMs: 0, + }); + expect(recovered).toBe(0); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + expect(await failedUpdateIds(tempDir)).toEqual([]); + }); + }); + + it("records failed-retryable when dispatch throws before adoption", async () => { + await withTempSpool(async (tempDir) => { + const abort = new AbortController(); + const log = vi.fn(); + const events: string[] = []; + await writeSpooledTestUpdates(tempDir, [topicUpdate(42, 10, "pre-adoption failure")]); + + const { runPromise, stopWorker } = startIsolatedIngressSession({ + abort, + spoolDir: tempDir, + log, + drainIntervalMs: 500, + handleUpdate: async (update) => { + events.push(`throw:${update.update_id}`); + throw new Error("session resolve failed before adoption"); + }, + }); + + await vi.waitFor(() => expect(events).toEqual(["throw:42"])); + await vi.waitFor(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([42])); + expect(await failedUpdateIds(tempDir)).toEqual([]); + expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]); + expectLogIncludes(log, "spooled update 42 failed; keeping for retry"); + expectLogExcludes(log, "handler-timeout"); + + abort.abort(); + stopWorker(); + await runPromise; + }); + }); + + it("drains a second same-lane update after the first turn is adopted", async () => { + await withTempSpool(async (tempDir) => { + const abort = new AbortController(); + const events: string[] = []; + const participants: TelegramSpooledReplayDeferredParticipant[] = []; + await writeSpooledTestUpdates(tempDir, [ + topicUpdate(42, 10, "first long turn"), + topicUpdate(43, 10, "second turn same lane"), + ]); + + const { runPromise, stopWorker } = startIsolatedIngressSession({ + abort, + spoolDir: tempDir, + drainIntervalMs: 10, + handleUpdate: async (update) => { + events.push(`dispatch:${update.update_id}`); + if (update.update_id === 42) { + const participant = createTelegramSpooledReplayDeferredParticipant( + `test-lane:${update.update_id}`, + ); + if (!participant) { + throw new Error("expected spooled replay participant"); + } + participants.push(participant); + // Adopt immediately so the lane frees while a long turn would + // continue under run lifecycle (not retested here). + participant.settle({ kind: "completed" }); + } + }, + }); + + await vi.waitFor(() => expect(events).toContain("dispatch:42")); + await vi.waitFor(async () => + expect(await listTelegramSpooledUpdateClaims({ spoolDir: tempDir })).toEqual([]), + ); + // Lane free after adoption: second update reaches kernel dispatch. + await vi.waitFor(() => expect(events).toEqual(["dispatch:42", "dispatch:43"])); + expect(await pendingUpdateIds(tempDir, "all")).toEqual([]); + abort.abort(); stopWorker(); await runPromise; diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index fe632188b337..c756b8e0fbb0 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -132,7 +132,10 @@ const TELEGRAM_DELIVERY_DRAIN_INTERVAL_MS = 5_000; const MAX_POLL_STALL_THRESHOLD_MS = 600_000; const POLL_WATCHDOG_INTERVAL_MS = 30_000; const POLL_STOP_GRACE_MS = 15_000; +// Status-only backlog note threshold (unrelated to adoption timeout). const ISOLATED_INGRESS_BACKLOG_STALL_MS = 25 * 60_000; +// claim→adoption only; once adopted, run lifecycle owns the turn. +const ISOLATED_INGRESS_ADOPTION_STALL_MS = 5 * 60_000; const TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS = 5_000; const TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV = "OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS"; const TELEGRAM_SPOOLED_DRAIN_START_LIMIT = 100; @@ -309,7 +312,7 @@ function resolveSpooledUpdateHandlerTimeoutMs(params: { return timeoutMs; } } - return ISOLATED_INGRESS_BACKLOG_STALL_MS; + return ISOLATED_INGRESS_ADOPTION_STALL_MS; } function buildSpooledUpdateHandlerKey(params: { spoolDir: string; laneKey: string }): string { @@ -723,7 +726,9 @@ export class TelegramPollingSession { }; state.timer = setTimeout(() => { const age = formatDurationPrecise(this.#spooledUpdateHandlerTimeoutMs); - state.timedOutMessage = `Telegram isolated polling spool buffered processing timed out behind update ${params.update.updateId} on lane ${params.laneKey} after ${age}; marking the update failed, aborting active reply work, and keeping the claim out of retry while the buffered task settles.`; + // Pre-adoption only: once the deferred participant settles at adoption, + // this timer is cleared. A fire means ingress never adopted the turn. + state.timedOutMessage = `Telegram isolated polling spool pre-adoption timed out behind update ${params.update.updateId} on lane ${params.laneKey} after ${age}; marking the update failed (handler-timeout) and keeping the claim out of retry.`; state.stopClaimRefresh(); params.deferredWork.settle({ kind: "failed-retryable", @@ -742,7 +747,7 @@ export class TelegramPollingSession { async #failTimedOutDeferredSpooledUpdate(state: DeferredSpooledUpdateClaimState): Promise { const message = state.timedOutMessage ?? - `Telegram isolated polling spool buffered processing timed out behind update ${state.updateId} on lane ${state.laneKey}; marking the update failed.`; + `Telegram isolated polling spool pre-adoption timed out behind update ${state.updateId} on lane ${state.laneKey}; marking the update failed.`; try { const failed = await failTelegramSpooledUpdateClaim({ update: state.update, @@ -751,18 +756,19 @@ export class TelegramPollingSession { }); if (!failed) { this.opts.log( - `[telegram][diag] timed out buffered spooled update ${state.updateId} no longer had a processing marker to fail.`, + `[telegram][diag] timed out pre-adoption spooled update ${state.updateId} no longer had a processing marker to fail.`, ); this.#status.notePollingError(message); return; } } catch (err) { this.opts.log( - `[telegram][diag] timed out buffered spooled update ${state.updateId} could not be marked failed: ${formatErrorMessage(err)}`, + `[telegram][diag] timed out pre-adoption spooled update ${state.updateId} could not be marked failed: ${formatErrorMessage(err)}`, ); this.#status.notePollingError(message); return; } + // Pre-adoption only: if a reply fence opened before adoption, release it. const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ accountId: this.opts.accountId, sequentialKey: state.laneKey, @@ -770,7 +776,7 @@ export class TelegramPollingSession { const abortedReplyWork = supersedeTelegramReplyFenceLane(scopedReplyFenceLaneKey); if (!abortedReplyWork) { this.opts.log( - `[telegram][diag] timed out buffered spooled update ${state.updateId} had no active reply fence on lane ${state.laneKey}.`, + `[telegram][diag] timed out pre-adoption spooled update ${state.updateId} had no active reply fence on lane ${state.laneKey}.`, ); } this.opts.log(`[telegram] ${message}`); @@ -1048,7 +1054,9 @@ export class TelegramPollingSession { const age = formatDurationPrecise(timedOutHandler.ageMs); activeHandler.timedOutAt = Date.now(); activeHandler.stopClaimRefresh(); - const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; marking the update failed, aborting active reply work, and restarting isolated ingress so later updates can drain.`; + // Pre-adoption stall: the active handler should return once deferred work + // is registered. A timeout here means ingress never reached adoption. + const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; marking the update failed (handler-timeout / pre-adoption) and restarting isolated ingress so later updates can drain.`; activeHandler.timeoutMessage = message; try { const failed = await failTelegramSpooledUpdateClaim({ @@ -1070,6 +1078,9 @@ export class TelegramPollingSession { this.#status.notePollingError(message); return { handlerKey: handler.handlerKey, restart: false }; } + // Best-effort: supersede any reply fence already opened during pre-adoption + // setup so a wedged handleUpdate can return. After adoption the spool no + // longer owns the turn, so this path should not see a settled agent run. const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ accountId: this.opts.accountId, sequentialKey: handler.laneKey, @@ -1688,6 +1699,7 @@ export const testing = { spooledRetryMaxAttempts: TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS, spooledRetryDeadLetterMinAgeMs: TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS, isolatedIngressBacklogStallMs: ISOLATED_INGRESS_BACKLOG_STALL_MS, + isolatedIngressAdoptionStallMs: ISOLATED_INGRESS_ADOPTION_STALL_MS, spooledClaimRefreshIntervalMs: TELEGRAM_SPOOLED_CLAIM_REFRESH_INTERVAL_MS, resolveSpooledUpdateHandlerAbortGraceMs: (valueMs: unknown): number => resolvePositiveTimerTimeoutMs(valueMs, TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS),