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

obviyus · obviyus · commit 77c84de5109a · 2026-07-10T14:32:11.000+05:30
diff --git 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
@@ -248,6 +248,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 };
@@ -787,6 +789,7 @@ export const dispatchTelegramMessage = async ({
   opts,
   retryDispatchErrors = false,
   suppressFailureFallback = false,
+  onTurnAdopted,
 }: DispatchTelegramMessageParams): Promise<TelegramDispatchResult> => {
   const dispatchStartedAt = Date.now();
   const dispatchContext = resolveDispatchTelegramContext({ context });
@@ -2615,6 +2618,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
@@ -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"));
   });
 
@@ -334,11 +334,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",
@@ -404,11 +410,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({
diff --git a/extensions/telegram/src/bot-message.ts b/extensions/telegram/src/bot-message.ts
@@ -19,6 +19,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,
@@ -243,55 +245,103 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
     await turnContext.onDispatchStart?.();
     const spooledReplay =
       options?.spooledReplay === true || isTelegramSpooledReplayUpdate(primaryCtx.update);
-    try {
-      const dispatchResult = await dispatchTelegramMessage({
-        context,
-        bot,
-        cfg: context.cfg,
-        runtime,
-        replyToMode: turnSettings.replyToMode,
-        streamMode: turnSettings.streamMode,
-        textLimit: turnSettings.textLimit,
-        telegramCfg: turnTelegramCfg,
-        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: context.cfg,
+          runtime,
+          replyToMode: turnSettings.replyToMode,
+          streamMode: turnSettings.streamMode,
+          textLimit: turnSettings.textLimit,
+          telegramCfg: turnTelegramCfg,
+          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
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
