fix(telegram): complete spooled updates at turn adoption, rescope ingress watchdog

(cherry picked from commit 77c84de510)

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).

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

@@ -246,6 +246,8 @@ type DispatchTelegramMessageParams = {
   opts: Pick<TelegramBotOptions, "token" | "mediaMaxMb">;
   retryDispatchErrors?: boolean;
   suppressFailureFallback?: boolean;
+  /** Fires after recovery-relevant session/run state is durably persisted. */
+  onTurnAdopted?: () => void | Promise<void>;
 };
 
 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<TelegramDispatchResult> => {
   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 }]

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<typeof createTelegramMessageProcessor>[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({

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<void>;
+    }): Promise<TelegramMessageProcessingResult> => {
+      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({});
   };
 };

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;

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<void> {
     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),