fix(telegram): let participant settlement own spooled update disposition (#149953)

## What Problem This Solves

A Telegram message that OpenClaw had already accepted into its durable ingress queue could vanish during a Gateway restart. If the stop signal arrived while the reply pipeline was still preparing context, shutdown aborted the turn after five seconds, no reply was sent, and the queued row was marked completed. After restart nothing replayed it, while the Gateway reported healthy.

## Why This Change Was Made

Two mechanisms decided the durable disposition of one spooled update: the per-update processing frame (which `bot-message.ts` wrote as `completed` as soon as the dispatch returned, including when the turn had been superseded by shutdown) and the spooled-replay participant (which settled `skipped` or `failed-retryable` after the abort). The ingress drain read the frame first and tombstoned the row before the participant's settlement was seen.

This change makes the participant the only owner. The decision "what happens to the durable row of a spooled Telegram update" is made by exactly one mechanism: the participant branch in `extensions/telegram/src/telegram-ingress-drain.ts`. The frame outcome is consulted only for handlers that never created a participant. `bot-message.ts` no longer writes frame outcomes at all; every writer that can coexist with a participant also settles that participant on the same path. The other writer of this state was the frame write in `bot-message.ts`, and it is removed; this state now has one writer.

## User Impact

A message interrupted by a restart before the turn was adopted returns to pending and is replayed after the Gateway comes back, so the user gets their reply. Turns already adopted keep their current behavior (not replayed, because that would re-run tools). No config, schema, protocol, or Doctor change: nothing persisted changes shape.

## Evidence

Red on `b884e39bb2` through the production path: a real bot plus the transport ingress monitor, `bot.handleUpdate`, message processing, and a downstream turn that returns only after the monitor's stop aborts it before adoption. The durable row was tombstoned; after this change it stays pending with no attempt counted:

```
node scripts/run-vitest.mjs extensions/telegram/src/telegram-ingress-coalescing.e2e.test.ts -t 'before adoption'
# base: exit 1, expected [] to match object [{ attempts: 0, id: "0000000000000901" }]
# this head: exit 0
```

A second regression at the drain seam (frame records `completed`, participant settles `skipped` after abort) is red on the base and green here with the same shape. Owner suites on this head: all `extensions/telegram/src` test files pass except two `telegram-ingress-coalescing.e2e` cases that fail identically on the unmodified base (same command, exit 1 on `b884e39bb28`). Type-aware lint, production and test types, protocol outputs, format, ratchets, and dead-export checks pass. Live Telegram Test Server run on the built candidate (real QA user, real bot, isolated Gateway, mock model): a photo message was accepted and spooled; its `getFile` response was held at the Bot API proxy so the turn was still hydrating media when the Gateway received SIGTERM. The Gateway logged `spooled update … failed; keeping for retry`, exited cleanly, restarted 3 s later, replayed the spooled update, and the user saw the bot's reply 12.8 s after sending. The prepare-context abort itself (the reported timing window) is covered by the drain regression above, since it cannot be hit deterministically on a live transport.

Proof boundary: the end-to-end regression keeps the production path intact from the transport monitor through message processing and reply dispatch, and substitutes only the reply pipeline's final dispatch with a stand-in that mirrors its pre-adoption abort contract (returns after the abort signal, never adopts). That contract is owned and tested by the reply pipeline's own abort tests, which this change does not touch. The context-preparation stage has no operator-controlled hold, so the live Test Server run proves the user flow on the real transport rather than that exact window. The maintainer accepted this proof set before landing.

## Compatibility

Additive removal only: `-13` production lines, no exported symbol, config key, schema, or protocol field changed. Feishu's ingress adapter has a different shape (no participant) and is unchanged.

## Consumers

Frame writers audited (event bindings, inbound processing, message pipeline, callback router, bot core): each site that can coexist with a participant settles it on the same path. Watermark tracking in `bot-core.ts` is unchanged.

## Invalidation

None; no cached or derived state introduced.

## Tests

One end-to-end regression through the real bot and transport monitor (shutdown abort before adoption keeps the update pending). Three drain tests: participant settlement owns the row after abort; participant adoption wins over a failed frame outcome; an adopted row stays tombstoned when inline completion returns after shutdown. Two handler-level tests that asserted the removed frame write now assert the participant's settlement or the absence of a retry marker.

Co-authored-by: Ayaan Zaidi <hi@obviy.us>

extensions/telegram/src/bot-core.ts

@@ -245,9 +245,9 @@ export function createTelegramBotCore(
     }
   });
 
-  // Durable transports start the answer after spool commit; classic polling and
-  // restart replay start it here. Both paths precede same-lane sequentialization
-  // so callback acknowledgements cannot wait for earlier handlers.
+  // Both transports start callback answers after spool commit. Reuse that
+  // answer or start a missing one before same-lane sequentialization so
+  // callback acknowledgements cannot wait for earlier handlers.
   bot.use(async (ctx, next) => {
     const callback = ctx.callbackQuery;
     if (callback) {

extensions/telegram/src/bot-message.ts

@@ -29,7 +29,6 @@ import {
   getTelegramSpooledReplayDeferredParticipant,
   getTelegramSpooledReplayLifecycle,
   isTelegramSpooledReplayUpdate,
-  recordTelegramMessageProcessingResult,
   type TelegramMessageProcessingResult,
 } from "./bot-processing-outcome.js";
 import type { TelegramBotOptions } from "./bot.types.js";
@@ -196,12 +195,6 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
     const ingressDebugEnabled =
       shouldLogVerbose() || process.env.OPENCLAW_DEBUG_TELEGRAM_INGRESS === "1";
     const ingressContextStartMs = ingressReceivedAtMs ? Date.now() : undefined;
-    const recordCurrentUpdateProcessingResult = (result: TelegramMessageProcessingResult) => {
-      if (options?.spooledReplay === true) {
-        return;
-      }
-      recordTelegramMessageProcessingResult(result);
-    };
     const context = await buildTelegramMessageContext({
       primaryCtx,
       allMedia,
@@ -238,7 +231,6 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
         );
       }
       const result: TelegramMessageProcessingResult = { kind: "skipped" };
-      recordCurrentUpdateProcessingResult(result);
       return result;
     }
     if (ingressDebugEnabled && ingressReceivedAtMs && ingressContextStartMs) {
@@ -296,7 +288,6 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
             kind: "failed-retryable",
             error: dispatchResult.error,
           };
-          recordCurrentUpdateProcessingResult(result);
           return result;
         }
         if (ingressDebugEnabled && ingressReceivedAtMs) {
@@ -306,7 +297,6 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
           );
         }
         const result: TelegramMessageProcessingResult = { kind: "completed" };
-        recordCurrentUpdateProcessingResult(result);
         return result;
       } catch (err) {
         runtime.error?.(danger(`telegram message processing failed: ${String(err)}`));
@@ -323,7 +313,6 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep
           kind: "failed-retryable",
           error: err,
         };
-        recordCurrentUpdateProcessingResult(result);
         return result;
       }
     };

extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts

@@ -57,8 +57,11 @@ const harness = await import("./bot.create-telegram-bot.test-harness.js");
 const { getLoadConfigMock, getOnHandler, replySpy, sendMessageSpy, telegramBotDepsForTest } =
   harness;
 const { createTelegramBotCore: createTelegramBotBase } = await import("./bot-core.js");
-const { runWithTelegramSpooledReplayUpdate, runWithTelegramUpdateProcessingFrame } =
-  await import("./bot-processing-outcome.js");
+const {
+  getTelegramSpooledReplayDeferredParticipant,
+  runWithTelegramSpooledReplayUpdate,
+  runWithTelegramUpdateProcessingFrame,
+} = await import("./bot-processing-outcome.js");
 const { MediaFetchError } = await import("./telegram-media.runtime.js");
 
 let createTelegramBot: (
@@ -606,10 +609,16 @@ describe("createTelegramBot channel_post media", () => {
       fileName: "document.pdf",
       update,
     });
-    const { result } = await runWithTelegramUpdateProcessingFrame(() =>
-      withTelegramSpooledReplayUpdate(update, () => handler(ctx)),
-    );
-    expect(result).toEqual(testCase.result);
+    // Durable ingress reads the participant's settlement when the handler
+    // created one; the frame result covers handlers that never reached it.
+    const { value: participantResult, result: frameResult } =
+      await runWithTelegramUpdateProcessingFrame(() =>
+        withTelegramSpooledReplayUpdate(update, async () => {
+          await handler(ctx);
+          return await getTelegramSpooledReplayDeferredParticipant()?.task;
+        }),
+      );
+    expect(participantResult ?? frameResult).toEqual(testCase.result);
     const expectedWarnings = testCase.warning ? 1 : 0;
     expect(sendMessageSpy).toHaveBeenCalledTimes(expectedWarnings);
     expect(replySpy).toHaveBeenCalledTimes(expectedWarnings);

extensions/telegram/src/bot.test.ts

@@ -4343,7 +4343,9 @@ describe("createTelegramBot", () => {
       }),
     );
 
-    expect(result).toEqual({ kind: "completed" });
+    // Live polling records no retry marker for this abort; the middleware
+    // owner completes the update once the current text has been dispatched.
+    expect(result?.kind).not.toBe("failed-retryable");
     expect(getFileSpy).toHaveBeenCalledWith("reply-photo-1", expect.any(AbortSignal));
     expect(replySpy).toHaveBeenCalledTimes(1);
     const payload = mockMsgContextArg(replySpy, 0, 0, "replySpy call");

extensions/telegram/src/telegram-ingress-coalescing.e2e.test.ts

@@ -23,7 +23,7 @@ import type { TelegramTransport } from "./fetch.js";
 import type { TelegramRuntime } from "./runtime.types.js";
 
 const downstreamTurns = vi.hoisted(() =>
-  vi.fn(async (_ctx: MsgContext) => ({
+  vi.fn(async (_ctx: MsgContext, _abortSignal?: AbortSignal) => ({
     queuedFinal: false,
     counts: { block: 0, final: 0, tool: 0 },
   })),
@@ -45,7 +45,7 @@ vi.mock("openclaw/plugin-sdk/channel-inbound", async (importOriginal) => {
     ...actual,
     runChannelInboundEvent: async (params: Parameters<typeof actual.runChannelInboundEvent>[0]) =>
       await runTelegramChannelInboundEventWithHarness(actual, params, async (dispatchParams) => {
-        return await downstreamTurns(dispatchParams.ctx);
+        return await downstreamTurns(dispatchParams.ctx, dispatchParams.replyOptions?.abortSignal);
       }),
   };
 });
@@ -442,6 +442,49 @@ describe("Telegram durable ingress coalescing", () => {
     await telegramTransport.close();
   });
 
+  it("keeps a text update pending when shutdown aborts the turn before adoption", async () => {
+    const update = textUpdate({
+      updateId: 901,
+      messageId: 1,
+      text: "interrupted by restart",
+    });
+    const eventId = telegramQueueEventId(update.update_id);
+    await writeTelegramSpooledUpdate({ spoolDir, update });
+    const queue = openTelegramIngressQueue(spoolDir);
+    downstreamTurns.mockImplementationOnce(async (_ctx, abortSignal) => {
+      if (!abortSignal) {
+        throw new Error("Expected the turn's abort signal");
+      }
+      await new Promise<void>((resolve) => {
+        abortSignal.addEventListener("abort", () => resolve(), { once: true });
+      });
+      return { queuedFinal: false, counts: { block: 0, final: 0, tool: 0 } };
+    });
+    const { monitor, telegramTransport } = await createMonitor();
+
+    monitor.start();
+    await awaitSingleDownstreamTurn();
+    await monitor.stop();
+    await telegramTransport.close();
+
+    await vi.waitFor(async () => {
+      expect(await queue.listClaims()).toEqual([]);
+      expect(await queue.listPending({ limit: "all" })).toMatchObject([
+        { id: eventId, attempts: 0 },
+      ]);
+    });
+    expect(
+      (
+        await queue.enqueue(eventId, {
+          version: 1,
+          updateId: update.update_id,
+          receivedAt: Date.now(),
+          update,
+        })
+      ).kind,
+    ).not.toBe("completed");
+  });
+
   it("releases a stale forwarded claim once when custom debounce dispatch fails", async () => {
     const update = forwardedTextUpdate({
       updateId: 701,

extensions/telegram/src/telegram-ingress-drain.test.ts

@@ -8,6 +8,8 @@ import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import { describe, expect, it, vi } from "vitest";
 import {
   createTelegramSpooledReplayDeferredParticipant,
+  recordTelegramMessageProcessingResult,
+  runWithTelegramUpdateProcessingFrame,
   type TelegramSpooledReplayDeferredParticipant,
 } from "./bot-processing-outcome.js";
 import { resolveTelegramForumFlag } from "./bot/helpers.js";
@@ -98,6 +100,14 @@ describe("resolveTelegramIngressNonRetryableFailure", () => {
   });
 });
 
+function deferred<T = void>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((res) => {
+    resolve = res;
+  });
+  return { promise, resolve };
+}
+
 describe("createTelegramIngressMonitor", () => {
   it("dead-letters a real blocked-recipient Telegram API error without retrying it", async () => {
     await withTempState(async (stateDir) => {
@@ -350,6 +360,128 @@ describe("createTelegramIngressMonitor", () => {
     },
   );
 
+  it("uses participant settlement to own the row despite frame completion after abort", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueueForTests<TelegramSpooledUpdatePayload>({
+        channelId: "telegram",
+        accountId: "default",
+        stateDir,
+      });
+      const payload = updatePayload(8);
+      const eventId = String(payload.updateId).padStart(16, "0");
+      const laneKey = telegramSpooledUpdateLaneKey(payload.update);
+      await queue.enqueue(eventId, payload, { laneKey });
+      const started = deferred();
+      const monitor = createTelegramIngressMonitor({
+        queue,
+        getConfig: () => cfg,
+        accountId: "default",
+        dispatch: async (_update, lifecycle) => {
+          const { result } = await runWithTelegramUpdateProcessingFrame(async () => {
+            const participant = createTelegramSpooledReplayDeferredParticipant("test:inline-abort");
+            expect(participant).not.toBeNull();
+            const aborted = new Promise<void>((resolve) => {
+              lifecycle.abortSignal.addEventListener("abort", () => resolve(), { once: true });
+            });
+            started.resolve();
+            await aborted;
+            recordTelegramMessageProcessingResult({ kind: "completed" });
+            participant?.settle({ kind: "skipped" });
+            await participant?.task;
+          });
+          return result;
+        },
+      });
+
+      monitor.start();
+      await started.promise;
+      await monitor.stop();
+
+      expect(await queue.listPending({ limit: "all" })).toMatchObject([
+        { id: eventId, attempts: 0 },
+      ]);
+      expect((await queue.listPending({ limit: "all" }))[0]?.lastError).toBeUndefined();
+      expect((await queue.enqueue(eventId, payload, { laneKey })).kind).not.toBe("completed");
+    });
+  });
+
+  it("uses participant adoption to own the row despite a failed frame outcome", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueueForTests<TelegramSpooledUpdatePayload>({
+        channelId: "telegram",
+        accountId: "default",
+        stateDir,
+      });
+      const payload = updatePayload(8);
+      const eventId = String(payload.updateId).padStart(16, "0");
+      const laneKey = telegramSpooledUpdateLaneKey(payload.update);
+      await queue.enqueue(eventId, payload, { laneKey });
+      const release = vi.spyOn(queue, "release");
+      const monitor = createTelegramIngressMonitor({
+        queue,
+        getConfig: () => cfg,
+        accountId: "default",
+        dispatch: async () => {
+          const { result } = await runWithTelegramUpdateProcessingFrame(async () => {
+            const participant = createTelegramSpooledReplayDeferredParticipant("test:adopted");
+            expect(participant).not.toBeNull();
+            recordTelegramMessageProcessingResult({
+              kind: "failed-retryable",
+              error: new Error("late frame failure"),
+            });
+            participant?.settle({ kind: "completed" });
+            await participant?.task;
+          });
+          return result;
+        },
+      });
+
+      monitor.start();
+      await monitor.waitForIdle();
+      await monitor.stop();
+
+      expect((await queue.enqueue(eventId, payload, { laneKey })).kind).toBe("completed");
+      expect(await queue.listPending({ limit: "all" })).toEqual([]);
+      expect(release).not.toHaveBeenCalled();
+    });
+  });
+
+  it("preserves an adopted tombstone when inline completion returns after shutdown", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createChannelIngressQueueForTests<TelegramSpooledUpdatePayload>({
+        channelId: "telegram",
+        accountId: "default",
+        stateDir,
+      });
+      const payload = updatePayload(8);
+      const eventId = String(payload.updateId).padStart(16, "0");
+      const laneKey = telegramSpooledUpdateLaneKey(payload.update);
+      await queue.enqueue(eventId, payload, { laneKey });
+      const adopted = deferred();
+      const finishDispatch = deferred();
+      const monitor = createTelegramIngressMonitor({
+        queue,
+        getConfig: () => cfg,
+        accountId: "default",
+        dispatch: async (_update, lifecycle) => {
+          await lifecycle.onAdopted();
+          adopted.resolve();
+          await finishDispatch.promise;
+          return { kind: "completed" };
+        },
+      });
+
+      monitor.start();
+      await adopted.promise;
+      const stopped = monitor.stop();
+      finishDispatch.resolve();
+      await stopped;
+
+      expect(await queue.listPending({ limit: "all" })).toEqual([]);
+      expect((await queue.enqueue(eventId, payload, { laneKey })).kind).toBe("completed");
+    });
+  });
+
   it("requeues an aborted deferred participant even when its late result is non-retryable", async () => {
     await withTempState(async (stateDir) => {
       const queue = createChannelIngressQueueForTests<TelegramSpooledUpdatePayload>({

extensions/telegram/src/telegram-ingress-drain.ts

@@ -373,18 +373,8 @@ export function createTelegramIngressMonitor(params: CreateTelegramIngressMonito
           },
           telegramLifecycle,
         );
-        const outcome = result.value;
-        if (outcome && typeof outcome === "object" && "kind" in outcome) {
-          if (outcome.kind === "failed-retryable") {
-            return { kind: "failed-retryable", error: outcome.error };
-          }
-          if (outcome.kind === "completed" || outcome.kind === "skipped") {
-            await lifecycle.onAdopted();
-            return { kind: "completed" };
-          }
-        }
-        // Every spooled participant gets deferredWork. Forward its terminal
-        // result without retaining Telegram's ingress serialization lane.
+        // A participant owns durable disposition; frame outcomes apply only
+        // to handlers that did not create one.
         const participant = result.deferredWork;
         if (participant) {
           let abortedWhilePending = participant.wasOwnerAbortedWhilePending();
@@ -452,14 +442,22 @@ export function createTelegramIngressMonitor(params: CreateTelegramIngressMonito
             });
           return { kind: "deferred" };
         }
-        if (!participant) {
-          // A dispatched update that records no outcome and defers no participant
-          // was consumed silently; completing here tombstones the spool row with
-          // attempts=0 and no trace, so keep a diagnostic trail for regressions.
-          params.onLog?.(
-            `telegram ingress: update ${resolveTelegramUpdateId(update) ?? "unknown"} completed without a recorded processing outcome`,
-          );
+        const outcome = result.value;
+        if (outcome && typeof outcome === "object" && "kind" in outcome) {
+          if (outcome.kind === "failed-retryable") {
+            return { kind: "failed-retryable", error: outcome.error };
+          }
+          if (outcome.kind === "completed" || outcome.kind === "skipped") {
+            await lifecycle.onAdopted();
+            return { kind: "completed" };
+          }
         }
+        // A dispatched update that records no outcome and defers no participant
+        // was consumed silently; completing here tombstones the spool row with
+        // attempts=0 and no trace, so keep a diagnostic trail for regressions.
+        params.onLog?.(
+          `telegram ingress: update ${resolveTelegramUpdateId(update) ?? "unknown"} completed without a recorded processing outcome`,
+        );
         await lifecycle.onAdopted();
         return { kind: "completed" };
       } catch (error) {