fix: retry debounced ingress failures before adoption

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

@@ -21,7 +21,12 @@ import { runTelegramChannelInboundEventWithHarness } from "./bot.test-helpers.js
 import type { TelegramTransport } from "./fetch.js";
 import type { TelegramRuntime } from "./runtime.types.js";
 
-const downstreamTurns = vi.hoisted(() => vi.fn());
+const downstreamTurns = vi.hoisted(() =>
+  vi.fn(async (_ctx: MsgContext) => ({
+    queuedFinal: false,
+    counts: { block: 0, final: 0, tool: 0 },
+  })),
+);
 
 vi.mock("./fetch.js", () => ({
   resolveTelegramApiBase: (apiRoot?: string) => apiRoot ?? "https://api.telegram.org",
@@ -39,8 +44,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) => {
-        downstreamTurns(dispatchParams.ctx);
-        return { queuedFinal: false, counts: { block: 0, final: 0, tool: 0 } };
+        return await downstreamTurns(dispatchParams.ctx);
       }),
   };
 });
@@ -255,7 +259,9 @@ describe("Telegram durable ingress coalescing", () => {
     process.env.OPENCLAW_STATE_DIR = stateDir;
     spoolDir = path.join(stateDir, "telegram", "ingress-spool-default");
     activeResources = [];
-    downstreamTurns.mockClear();
+    downstreamTurns
+      .mockReset()
+      .mockResolvedValue({ queuedFinal: false, counts: { block: 0, final: 0, tool: 0 } });
     resetInboundDedupe();
     resetPluginStateStoreForTests({ closeDatabase: false });
     resetTelegramAccountThrottlersForTest();
@@ -531,4 +537,40 @@ describe("Telegram durable ingress coalescing", () => {
     await monitor.stop();
     await telegramTransport.close();
   });
+
+  it("releases a stale forwarded claim once when custom debounce dispatch fails", async () => {
+    const update = forwardedTextUpdate({
+      updateId: 701,
+      messageId: 1,
+      text: "recovered forward",
+    });
+    const eventId = telegramQueueEventId(update.update_id);
+    const sessionError = new Error("Session changed while starting work. Retry.");
+    await writeTelegramSpooledUpdate({ spoolDir, update });
+    const queue = openTelegramIngressQueue(spoolDir);
+    expect(await queue.claim(eventId, { ownerId: "999:1:dead-owner" })).not.toBeNull();
+    downstreamTurns.mockRejectedValueOnce(sessionError);
+    const runtimeError = vi.fn();
+    const { monitor, telegramTransport } = await createMonitor({
+      adoptionStallTimeoutMs: 5_000,
+      onRuntimeError: runtimeError,
+    });
+
+    monitor.start();
+    await vi.waitFor(
+      async () => {
+        expect(await queue.listClaims()).toEqual([]);
+        expect(await queue.listFailed?.({ limit: "all" })).toEqual([]);
+        expect(await queue.listPending({ limit: "all" })).toMatchObject([
+          { id: eventId, attempts: 2, lastError: sessionError.message },
+        ]);
+      },
+      { timeout: 2_000, interval: 5 },
+    );
+    expect(downstreamTurns).toHaveBeenCalledOnce();
+    expect(runtimeError).toHaveBeenCalledOnce();
+
+    await monitor.stop();
+    await telegramTransport.close();
+  });
 });

src/auto-reply/inbound-debounce.ts

@@ -108,7 +108,11 @@ function createInboundDebounceFlush(params: {
     onAdoptionFinalizing: () => source?.onAdoptionFinalizing?.(),
     onFailed: source?.onFailed
       ? async (error) => {
-          await source.onFailed?.(error);
+          try {
+            await source.onFailed?.(error);
+          } finally {
+            markAdmitted();
+          }
         }
       : undefined,
     onAbandoned: async () => {
@@ -121,9 +125,15 @@ function createInboundDebounceFlush(params: {
   } catch (error) {
     completion = Promise.reject(toErrorObject(error, "Inbound debounce dispatch failed"));
   }
-  // A skipped or failed dispatch may never call a lifecycle hook; its terminal
-  // completion must still release the keyed chain.
-  void completion.then(markAdmitted, markAdmitted);
+  // A failed dispatch must settle its source claim before releasing the keyed
+  // lane; an already-admitted turn owns its later completion failure.
+  void completion
+    .then(markAdmitted, async (error: unknown) => {
+      if (!admitted) {
+        await lifecycle.onFailed?.(error);
+      }
+    })
+    .then(markAdmitted, markAdmitted);
   return { admission, completion };
 }
 
@@ -187,7 +197,7 @@ export function createInboundDebouncer<T>(params: InboundDebounceCreateParams<T>
     activeCompletions.add(completion);
     const cleanup = () => activeCompletions.delete(completion);
     void completion.then(cleanup, cleanup);
-    await Promise.race([admission, completion]);
+    await admission;
   };
 
   const cancelItems = (items: T[]) => {

src/auto-reply/inbound.test.ts

@@ -908,6 +908,43 @@ describe("createInboundDebouncer", () => {
     expect(completed).toEqual(["2", "1"]);
   });
 
+  it("hands pre-admission completion failures to the source lifecycle once", async () => {
+    const sessionError = new Error("Session changed while starting work. Retry.");
+    const onFailed = vi.fn(async () => {});
+    const onError = vi.fn();
+    let attempt = 0;
+    const debouncer = createInboundDebouncer<{ key: string; id: string }>({
+      debounceMs: 0,
+      buildKey: (item) => item.key,
+      onFlush: (_items, createFlush) =>
+        createFlush({
+          lifecycle: { onFailed },
+          dispatch: async (lifecycle) => {
+            attempt += 1;
+            if (attempt === 1) {
+              throw sessionError;
+            }
+            await lifecycle.onAdopted();
+            throw new Error("post-adoption failure");
+          },
+        }),
+      onError,
+    });
+
+    await expect(debouncer.enqueue({ key: "a", id: "failed-before-admission" })).resolves.toBe(
+      undefined,
+    );
+    await expect(debouncer.enqueue({ key: "a", id: "failed-after-admission" })).resolves.toBe(
+      undefined,
+    );
+    await debouncer.drain();
+
+    expect(onFailed).toHaveBeenCalledOnce();
+    expect(onFailed).toHaveBeenCalledWith(sessionError);
+    expect(onError).toHaveBeenCalledTimes(2);
+    expect(onError.mock.calls[0]?.[0]).toBe(sessionError);
+  });
+
   it("drains same-key flushes queued before their completion is tracked", async () => {
     const started: string[] = [];
     let releaseFirst!: () => void;

src/channels/message/ingress-drain.debounce-failure.test.ts

@@ -0,0 +1,87 @@
+// Shared debounce-to-drain composition regression for pre-admission failures.
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createInboundDebouncer } from "../../auto-reply/inbound-debounce.js";
+import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js";
+import { createChannelIngressDrain, DEFAULT_INGRESS_ADOPTION_STALL_MS } from "./ingress-drain.js";
+import {
+  createTestIngressQueue,
+  type IngressDrainTestPayload as Payload,
+  withTempState,
+} from "./ingress-drain.test-helpers.js";
+
+type ChannelIngressDispatchLifecycle = Parameters<
+  Parameters<typeof createChannelIngressDrain>[0]["dispatchClaimedEvent"]
+>[1];
+
+describe("channel ingress drain debounce failures", () => {
+  afterEach(() => {
+    vi.useRealTimers();
+    closeOpenClawStateDatabaseForTest();
+  });
+
+  it("retries a pre-admission failure without waiting for the watchdog", async () => {
+    await withTempState(async (stateDir) => {
+      let clock = 10_000;
+      const queue = createTestIngressQueue(stateDir, { now: () => clock });
+      await queue.enqueue(
+        "debounced-retry",
+        { text: "retry me" },
+        {
+          laneKey: "shared",
+          receivedAt: clock,
+        },
+      );
+      const sessionError = new Error("Session changed while starting work. Retry.");
+      const reportedErrors: unknown[] = [];
+      let attempt = 0;
+      const debouncer = createInboundDebouncer<{ lifecycle: ChannelIngressDispatchLifecycle }>({
+        debounceMs: 0,
+        buildKey: () => "shared",
+        onFlush: (entries, createFlush) =>
+          createFlush({
+            lifecycle: entries[0]?.lifecycle,
+            dispatch: async (lifecycle) => {
+              attempt += 1;
+              if (attempt === 1) {
+                throw sessionError;
+              }
+              await lifecycle.onAdopted();
+            },
+          }),
+        onError: (error) => reportedErrors.push(error),
+      });
+      const drain = createChannelIngressDrain<Payload>({
+        queue,
+        now: () => clock,
+        adoptionStallTimeoutMs: DEFAULT_INGRESS_ADOPTION_STALL_MS,
+        retryPolicy: { baseMs: 1_000, maxMs: 1_000 },
+        dispatchClaimedEvent: async (_event, lifecycle) => {
+          await debouncer.enqueue({ lifecycle });
+          return { kind: "deferred" };
+        },
+      });
+
+      expect(await drain.drainOnce()).toEqual({ started: 1 });
+      await drain.waitForIdle();
+      expect(await queue.listPending({ limit: "all" })).toMatchObject([
+        { id: "debounced-retry", attempts: 1, lastError: sessionError.message },
+      ]);
+      expect(await queue.listFailed?.({ limit: "all" })).toEqual([]);
+
+      clock += 1_000;
+      expect(await drain.drainOnce()).toEqual({ started: 1 });
+      await drain.waitForIdle();
+      await debouncer.drain();
+
+      expect(attempt).toBe(2);
+      expect(reportedErrors).toEqual([sessionError]);
+      expect(await queue.listPending({ limit: "all" })).toEqual([]);
+      expect(await queue.listClaims()).toEqual([]);
+      expect(await queue.listFailed?.({ limit: "all" })).toEqual([]);
+      expect(await queue.enqueue("debounced-retry", { text: "retry me" })).toMatchObject({
+        kind: "completed",
+      });
+      drain.dispose();
+    });
+  });
+});