fix(plugin-sdk): make ingress cancellation fan-in safe

extensions/discord/src/monitor/message-handler.ingress-recovery.test.ts

@@ -0,0 +1,263 @@
+// Discord tests cover durable retry recovery through full handler replacement.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import type { APIMessage } from "discord-api-types/v10";
+import {
+  type ChannelIngressQueue,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+} from "openclaw/plugin-sdk/channel-outbound";
+import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { describe, expect, it, vi } from "vitest";
+import { createDiscordIngressMonitor } from "./ingress.js";
+import { createDiscordMessageHandler } from "./message-handler.js";
+import { createDiscordHandlerParams } from "./message-handler.test-helpers.js";
+
+type DiscordIngressPayload = {
+  version: 1;
+  receivedAt: number;
+  rawMessage: APIMessage;
+};
+
+type DiscordQueue = ChannelIngressQueue<DiscordIngressPayload>;
+
+function rawMessage(id: string, channelId = "lane-a"): APIMessage {
+  return {
+    id,
+    channel_id: channelId,
+    content: "hello",
+    author: {
+      id: "user-1",
+      username: "alice",
+      discriminator: "0",
+      avatar: null,
+    },
+    attachments: [],
+    embeds: [],
+    mentions: [],
+    mention_roles: [],
+    mention_everyone: false,
+    timestamp: new Date(0).toISOString(),
+    edited_timestamp: null,
+    components: [],
+    pinned: false,
+    type: 0,
+    tts: false,
+  } as unknown as APIMessage;
+}
+
+async function withQueue(run: (queue: DiscordQueue) => Promise<void>): Promise<void> {
+  const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-discord-recovery-"));
+  const stateDir = await fs.realpath(created);
+  const queue = createChannelIngressQueueForTests<DiscordIngressPayload>({
+    channelId: "discord",
+    accountId: "default",
+    stateDir,
+  });
+  try {
+    await run(queue);
+  } finally {
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
+  }
+}
+
+async function seedPendingFailure(params: {
+  queue: DiscordQueue;
+  id: string;
+  attempts: number;
+  laneKey?: string;
+}): Promise<void> {
+  await params.queue.enqueue(
+    params.id,
+    { version: 1, receivedAt: 1, rawMessage: rawMessage(params.id) },
+    { laneKey: params.laneKey ?? "channel:lane-a", receivedAt: 1 },
+  );
+  for (let attempt = 1; attempt <= params.attempts; attempt += 1) {
+    const claim = await params.queue.claim(params.id, { ownerId: `seed-${attempt}` });
+    if (!claim) {
+      throw new Error(`Expected ${params.id} to be claimable for seed attempt ${attempt}`);
+    }
+    await params.queue.release(claim, {
+      lastError: `prior genuine failure ${attempt}`,
+      releasedAt: 10 + attempt,
+    });
+  }
+}
+
+async function retryFacts(queue: DiscordQueue, id: string) {
+  const record = (await queue.listPending({ limit: "all" })).find((entry) => entry.id === id);
+  if (!record) {
+    throw new Error(`Expected pending Discord ingress row ${id}`);
+  }
+  return {
+    attempts: record.attempts,
+    lastAttemptAt: record.lastAttemptAt,
+    lastError: record.lastError,
+  };
+}
+
+function createHandler(params: {
+  queue: DiscordQueue;
+  preflight: (input: { data: { message?: { id?: string } } }) => Promise<null>;
+  debounceMs?: number;
+  beforeDispatch?: () => Promise<void>;
+}) {
+  const handlerParams = createDiscordHandlerParams();
+  handlerParams.cfg.messages = { inbound: { debounceMs: params.debounceMs ?? 0 } };
+  return createDiscordMessageHandler({
+    ...handlerParams,
+    client: {} as never,
+    testing: {
+      preflightDiscordMessage: params.preflight as never,
+      createIngressMonitor: (monitorParams) =>
+        createDiscordIngressMonitor({
+          ...monitorParams,
+          queue: params.queue,
+          dispatch: params.beforeDispatch
+            ? async (event, lifecycle) => {
+                await params.beforeDispatch?.();
+                return await monitorParams.dispatch(event, lifecycle);
+              }
+            : monitorParams.dispatch,
+        }),
+    },
+  });
+}
+
+describe("Discord durable ingress replacement recovery", () => {
+  it("terminally settles a preexisting exhausted poison row before its follower", async () => {
+    await withQueue(async (queue) => {
+      await seedPendingFailure({
+        queue,
+        id: "poison",
+        attempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+      });
+      await queue.enqueue(
+        "follower",
+        { version: 1, receivedAt: 2, rawMessage: rawMessage("follower") },
+        { laneKey: "channel:lane-a", receivedAt: 2 },
+      );
+      const dispatched: string[] = [];
+      const handler = createHandler({
+        queue,
+        preflight: vi.fn(async ({ data }) => {
+          const id = data.message?.id ?? "unknown";
+          dispatched.push(id);
+          if (id === "poison") {
+            throw new Error("recovered poison failure");
+          }
+          return null;
+        }),
+      });
+      try {
+        await vi.waitFor(async () => {
+          await expect(queue.enqueue("poison", {} as DiscordIngressPayload)).resolves.toMatchObject(
+            { kind: "failed", record: { reason: "retry-limit-exceeded" } },
+          );
+          await expect(
+            queue.enqueue("follower", {} as DiscordIngressPayload),
+          ).resolves.toMatchObject({ kind: "completed" });
+        });
+        expect(dispatched).toEqual(["poison", "follower"]);
+        expect(await queue.listPending({ limit: "all" })).toEqual([]);
+        expect(await queue.listClaims()).toEqual([]);
+      } finally {
+        await handler.deactivate();
+      }
+    });
+  });
+
+  it("preserves retry facts across every Discord cancellation route and replacement", async () => {
+    await withQueue(async (queue) => {
+      await seedPendingFailure({
+        queue,
+        id: "poison",
+        attempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS - 1,
+      });
+      await queue.enqueue(
+        "follower",
+        { version: 1, receivedAt: 2, rawMessage: rawMessage("follower") },
+        { laneKey: "channel:lane-a", receivedAt: 2 },
+      );
+      const expectedFacts = await retryFacts(queue, "poison");
+
+      const dispatchEntered = createDeferred<void>();
+      const releaseDispatch = createDeferred<void>();
+      const beforeDispatch = async () => {
+        dispatchEntered.resolve();
+        await releaseDispatch.promise;
+      };
+      const beforeDispatchPreflight = vi.fn(async () => null);
+      const beforeDispatchHandler = createHandler({
+        queue,
+        preflight: beforeDispatchPreflight,
+        beforeDispatch,
+      });
+      await dispatchEntered.promise;
+      const beforeDispatchStop = beforeDispatchHandler.deactivate();
+      await Promise.resolve();
+      releaseDispatch.resolve();
+      await beforeDispatchStop;
+      expect(beforeDispatchPreflight).not.toHaveBeenCalled();
+      expect(await retryFacts(queue, "poison")).toEqual(expectedFacts);
+
+      const bufferedPreflight = vi.fn(async () => null);
+      const bufferedHandler = createHandler({
+        queue,
+        preflight: bufferedPreflight,
+        debounceMs: 60_000,
+      });
+      await vi.waitFor(async () => expect(await queue.listClaims()).toHaveLength(1));
+      await bufferedHandler.deactivate();
+      expect(bufferedPreflight).not.toHaveBeenCalled();
+      expect(await retryFacts(queue, "poison")).toEqual(expectedFacts);
+
+      const preflightEntered = createDeferred<void>();
+      const releasePreflight = createDeferred<void>();
+      const activePreflight = vi.fn(async () => {
+        preflightEntered.resolve();
+        await releasePreflight.promise;
+        return null;
+      });
+      const activeHandler = createHandler({ queue, preflight: activePreflight });
+      await preflightEntered.promise;
+      const activeStop = activeHandler.deactivate();
+      await Promise.resolve();
+      releasePreflight.resolve();
+      await activeStop;
+      expect(activePreflight).toHaveBeenCalledTimes(1);
+      expect(await retryFacts(queue, "poison")).toEqual(expectedFacts);
+
+      const finalDispatches: string[] = [];
+      const replacement = createHandler({
+        queue,
+        preflight: vi.fn(async ({ data }) => {
+          const id = data.message?.id ?? "unknown";
+          finalDispatches.push(id);
+          if (id === "poison") {
+            throw new Error("final genuine failure");
+          }
+          return null;
+        }),
+      });
+      try {
+        await vi.waitFor(async () => {
+          await expect(queue.enqueue("poison", {} as DiscordIngressPayload)).resolves.toMatchObject(
+            { kind: "failed", record: { reason: "retry-limit-exceeded" } },
+          );
+          await expect(
+            queue.enqueue("follower", {} as DiscordIngressPayload),
+          ).resolves.toMatchObject({ kind: "completed" });
+        });
+        expect(finalDispatches).toEqual(["poison", "follower"]);
+      } finally {
+        await replacement.deactivate();
+      }
+    });
+  });
+});

extensions/feishu/src/monitor.message-handler.ingress.test.ts

@@ -1,11 +1,20 @@
-import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers";
-import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime";
 // Feishu ingress tests cover debounce ownership and constituent claim settlement.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce";
+import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound";
+import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { ClawdbotConfig, PluginRuntime, RuntimeEnv } from "../runtime-api.js";
 import * as dedup from "./dedup.js";
 import type { FeishuMessageEvent } from "./event-types.js";
-import type { FeishuIngressLifecycle } from "./feishu-ingress.js";
+import { createFeishuDurableIngress, type FeishuIngressLifecycle } from "./feishu-ingress.js";
 import { createFeishuMessageReceiveHandler } from "./monitor.message-handler.js";
 
 type MessageReceiveHandlerContext = Parameters<typeof createFeishuMessageReceiveHandler>[0];
@@ -370,4 +379,144 @@ describe("Feishu durable ingress debounce lifecycle", () => {
     expect(harness.handleMessage).toHaveBeenCalledTimes(1);
     expect(second.calls.adopted).not.toHaveBeenCalled();
   });
+
+  it("preserves abandon retry accounting, backoff, threshold, and restart behavior", async () => {
+    vi.useFakeTimers();
+    const now = Date.UTC(2026, 0, 2);
+    vi.setSystemTime(now);
+    const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-abandon-"));
+    const stateDir = await fs.realpath(created);
+    type Queue = NonNullable<Parameters<typeof createFeishuDurableIngress>[0]["queue"]>;
+    type Payload = Parameters<Queue["enqueue"]>[1];
+    const queue = createChannelIngressQueueForTests<Payload>({
+      channelId: "feishu",
+      accountId: "default",
+      stateDir,
+    });
+    const event = {
+      ...createTextEvent("evt-abandon-retry", "om-abandon-retry", "retry me"),
+      event_type: "im.message.receive_v1",
+    };
+    const handleMessage = vi.fn(async () => {
+      throw new Error("Feishu dispatch failed before adoption");
+    });
+    vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockImplementation(async () => ({
+      kind: "claimed",
+      handle: createClaim(`retry-${handleMessage.mock.calls.length}`),
+    }));
+
+    const createIntegratedIngress = () => {
+      const channelRuntime = {
+        commands: { isControlCommandMessage: () => false },
+        debounce: {
+          resolveInboundDebounceMs: () => 0,
+          createInboundDebouncer,
+        },
+      } as unknown as PluginRuntime["channel"];
+      const handler = createFeishuMessageReceiveHandler({
+        cfg: {} as ClawdbotConfig,
+        channelRuntime,
+        accountId: "default",
+        runtime: createNonExitingRuntimeEnv(),
+        chatHistories: new Map(),
+        handleMessage,
+        resolveDebounceText: () => "retry me",
+        hasProcessedMessage: vi.fn(async () => false),
+        getBotOpenId: () => "ou-bot",
+        resolveIngressLifecycle: (data) => ingress.resolveLifecycle(data),
+      });
+      const ingress = createFeishuDurableIngress({
+        accountId: "default",
+        queue,
+        dispatcher: { invoke: async (data: unknown) => await handler(data as never) } as never,
+        runtime: { error: vi.fn(), log: vi.fn() },
+        pollIntervalMs: 500,
+      });
+      return ingress;
+    };
+    const pendingAttempt = async (attempts: number) => {
+      let observed: Awaited<ReturnType<typeof queue.listPending>>[number] | undefined;
+      await vi.waitFor(async () => {
+        const pending = await queue.listPending({ limit: "all" });
+        expect(pending).toEqual([
+          expect.objectContaining({
+            id: "evt-abandon-retry",
+            attempts,
+            lastAttemptAt: expect.any(Number),
+            lastError: "turn-abandoned",
+          }),
+        ]);
+        observed = pending[0];
+      });
+      const lastAttemptAt = observed?.lastAttemptAt;
+      if (lastAttemptAt === undefined) {
+        throw new Error(`Missing Feishu retry timestamp for attempt ${attempts}`);
+      }
+      return { ...observed, lastAttemptAt };
+    };
+
+    try {
+      const first = createIntegratedIngress();
+      first.start();
+      await first.invokeWebhook(event);
+      const firstAttempt = await pendingAttempt(1);
+      expect(handleMessage).toHaveBeenCalledTimes(1);
+      await first.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 999);
+      const blocked = createIntegratedIngress();
+      blocked.start();
+      await blocked.invokeWebhook(event);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(handleMessage).toHaveBeenCalledTimes(1);
+      await blocked.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001);
+      const second = createIntegratedIngress();
+      second.start();
+      await second.invokeWebhook(event);
+      const secondAttempt = await pendingAttempt(2);
+      expect(handleMessage).toHaveBeenCalledTimes(2);
+      await second.stop();
+
+      for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
+        const claim = await queue.claim("evt-abandon-retry", { ownerId: `seed-${attempt}` });
+        if (!claim) {
+          throw new Error(`Expected Feishu seed claim ${attempt}`);
+        }
+        await queue.release(claim, {
+          lastError: "turn-abandoned",
+          releasedAt: secondAttempt.lastAttemptAt,
+        });
+      }
+
+      vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001);
+      const threshold = createIntegratedIngress();
+      threshold.start();
+      await threshold.invokeWebhook(event);
+      const thresholdAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS);
+      expect(handleMessage).toHaveBeenCalledTimes(3);
+      await threshold.stop();
+
+      vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001);
+      const beyond = createIntegratedIngress();
+      beyond.start();
+      await beyond.invokeWebhook(event);
+      const beyondAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1);
+      expect(handleMessage).toHaveBeenCalledTimes(4);
+      await beyond.stop();
+
+      vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000);
+      const blockedRestart = createIntegratedIngress();
+      blockedRestart.start();
+      await blockedRestart.invokeWebhook(event);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(handleMessage).toHaveBeenCalledTimes(4);
+      await blockedRestart.stop();
+    } finally {
+      closeOpenClawStateDatabaseForTest();
+      await fs.rm(stateDir, { recursive: true, force: true });
+      vi.useRealTimers();
+    }
+  });
 });

extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts

@@ -1,10 +1,20 @@
 // Mattermost tests cover monitor.inbound system event plugin behavior.
 import { once } from "node:events";
+import fs from "node:fs/promises";
 import { createServer } from "node:http";
+import os from "node:os";
+import path from "node:path";
 import { createChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound";
 import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce";
-import { createMessageReceiptFromOutboundResults } from "openclaw/plugin-sdk/channel-outbound";
+import {
+  createMessageReceiptFromOutboundResults,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+} from "openclaw/plugin-sdk/channel-outbound";
 import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { WebSocketServer } from "ws";
 import type { MattermostPost } from "./client.js";
@@ -96,6 +106,7 @@ const mockState = vi.hoisted(() => ({
   enqueueSystemEvent: vi.fn(),
   fetchMattermostMe: vi.fn(),
   getGlobalHookRunner: vi.fn(),
+  ingressQueue: undefined as unknown,
   progressDrafts: [] as Array<{ getSnapshot: () => { lines: readonly unknown[] } }>,
   registerMattermostMonitorSlashCommands: vi.fn(),
   registerPluginHttpRoute: vi.fn(),
@@ -180,27 +191,36 @@ vi.mock("./monitor-ingress.js", async (importOriginal) => {
     ...actual,
     createMattermostIngressMonitor: (
       options: Parameters<typeof actual.createMattermostIngressMonitor>[0],
-    ) => ({
-      receive: async (rawEvent: string) => {
-        const payload = JSON.parse(rawEvent) as MattermostEventPayload;
-        const post =
-          typeof payload.data?.post === "string"
-            ? (JSON.parse(payload.data.post) as MattermostPost)
-            : (payload.data?.post as MattermostPost | undefined);
-        if (payload.event !== "posted" || !post) {
-          return;
-        }
-        await options.dispatch(post, payload, {
-          abortSignal: new AbortController().signal,
-          onAdopted: async () => {},
-          onDeferred: () => {},
-          onAdoptionFinalizing: () => {},
-          onAbandoned: async () => {},
+    ) => {
+      if (mockState.ingressQueue) {
+        return actual.createMattermostIngressMonitor({
+          ...options,
+          queue: mockState.ingressQueue as NonNullable<typeof options.queue>,
+          pollIntervalMs: 60_000,
         });
-      },
-      stop: async () => {},
-      waitForIdle: async () => {},
-    }),
+      }
+      return {
+        receive: async (rawEvent: string) => {
+          const payload = JSON.parse(rawEvent) as MattermostEventPayload;
+          const post =
+            typeof payload.data?.post === "string"
+              ? (JSON.parse(payload.data.post) as MattermostPost)
+              : (payload.data?.post as MattermostPost | undefined);
+          if (payload.event !== "posted" || !post) {
+            return;
+          }
+          await options.dispatch(post, payload, {
+            abortSignal: new AbortController().signal,
+            onAdopted: async () => {},
+            onDeferred: () => {},
+            onAdoptionFinalizing: () => {},
+            onAbandoned: async () => {},
+          });
+        },
+        stop: async () => {},
+        waitForIdle: async () => {},
+      };
+    },
   };
 });
 
@@ -534,6 +554,7 @@ describe("mattermost inbound user posts", () => {
   beforeEach(() => {
     vi.clearAllMocks();
     mockState.abortController = undefined;
+    mockState.ingressQueue = undefined;
     mockState.progressDrafts.length = 0;
     mockState.getGlobalHookRunner.mockReturnValue(null);
     mockState.runtimeCore = createRuntimeCore(testConfig);
@@ -568,6 +589,146 @@ describe("mattermost inbound user posts", () => {
     });
   });
 
+  it("preserves abandon retry accounting, backoff, threshold, and restart behavior", async () => {
+    vi.useFakeTimers();
+    const now = Date.UTC(2026, 0, 2);
+    vi.setSystemTime(now);
+    const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mattermost-abandon-"));
+    const stateDir = await fs.realpath(created);
+    type Payload = { version: 1; receivedAt: number; rawEvent: string };
+    const queue = createChannelIngressQueueForTests<Payload>({
+      channelId: "mattermost",
+      accountId: "default",
+      stateDir,
+    });
+    mockState.ingressQueue = queue;
+    mockState.runtimeCore = createRuntimeCore(testConfig, undefined, {
+      inboundDebounceMs: 0,
+      createInboundDebouncer,
+    });
+    mockState.dispatchInboundMessage.mockRejectedValue(
+      new Error("Mattermost dispatch failed before adoption"),
+    );
+
+    const activeProviders: Array<{ stop: () => Promise<void> }> = [];
+    const startProvider = async () => {
+      const socket = new FakeWebSocket();
+      const abortController = new AbortController();
+      const monitor = monitorMattermostProvider({
+        config: testConfig,
+        runtime: testRuntime(),
+        abortSignal: abortController.signal,
+        webSocketFactory: () => socket,
+      });
+      for (let tick = 0; tick < 20 && socket.openListenerCount === 0; tick += 1) {
+        await Promise.resolve();
+      }
+      expect(socket.openListenerCount).toBeGreaterThan(0);
+      socket.emitOpen();
+      let stopped = false;
+      const provider = {
+        socket,
+        stop: async () => {
+          if (stopped) {
+            return;
+          }
+          stopped = true;
+          abortController.abort();
+          socket.emitClose(1000);
+          await monitor;
+        },
+      };
+      activeProviders.push(provider);
+      return provider;
+    };
+    const send = async (provider: Awaited<ReturnType<typeof startProvider>>) => {
+      await emitMattermostChannelPost(provider.socket, {
+        id: "post-abandon-retry",
+        message: "retry me",
+      });
+    };
+    const pendingAttempt = async (attempts: number) => {
+      let observed: Awaited<ReturnType<typeof queue.listPending>>[number] | undefined;
+      await vi.waitFor(async () => {
+        const pending = await queue.listPending({ limit: "all" });
+        expect(pending).toEqual([
+          expect.objectContaining({
+            id: "post-abandon-retry",
+            attempts,
+            lastAttemptAt: expect.any(Number),
+            lastError: "turn-abandoned",
+          }),
+        ]);
+        observed = pending[0];
+      });
+      const lastAttemptAt = observed?.lastAttemptAt;
+      if (lastAttemptAt === undefined) {
+        throw new Error(`Missing Mattermost retry timestamp for attempt ${attempts}`);
+      }
+      return { ...observed, lastAttemptAt };
+    };
+
+    try {
+      const first = await startProvider();
+      await send(first);
+      const firstAttempt = await pendingAttempt(1);
+      expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(1);
+      await first.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 999);
+      const blocked = await startProvider();
+      await send(blocked);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(1);
+      await blocked.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001);
+      const second = await startProvider();
+      await send(second);
+      const secondAttempt = await pendingAttempt(2);
+      expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(2);
+      await second.stop();
+
+      for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
+        const claim = await queue.claim("post-abandon-retry", { ownerId: `seed-${attempt}` });
+        if (!claim) {
+          throw new Error(`Expected Mattermost seed claim ${attempt}`);
+        }
+        await queue.release(claim, {
+          lastError: "turn-abandoned",
+          releasedAt: secondAttempt.lastAttemptAt,
+        });
+      }
+
+      vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001);
+      const threshold = await startProvider();
+      await send(threshold);
+      const thresholdAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS);
+      expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(3);
+      await threshold.stop();
+
+      vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001);
+      const beyond = await startProvider();
+      await send(beyond);
+      const beyondAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1);
+      expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(4);
+      await beyond.stop();
+
+      vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000);
+      const blockedRestart = await startProvider();
+      await send(blockedRestart);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(4);
+      await blockedRestart.stop();
+    } finally {
+      await Promise.allSettled(activeProviders.map(async (provider) => await provider.stop()));
+      mockState.ingressQueue = undefined;
+      closeOpenClawStateDatabaseForTest();
+      await fs.rm(stateDir, { recursive: true, force: true });
+      vi.useRealTimers();
+    }
+  });
+
   it("publishes recovering while API authentication retries, including 401", async () => {
     const abortController = new AbortController();
     const statusSink = vi.fn();

extensions/msteams/src/monitor-handler/message-handler.ingress-lifecycle.test.ts

@@ -1,7 +1,16 @@
 // Microsoft Teams tests cover durable claim ownership through inbound debounce.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
 import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce";
+import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig } from "../../runtime-api.js";
+import { createMSTeamsIngress } from "../msteams-ingress.js";
 import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js";
 import type { MSTeamsTurnContext } from "../sdk-types.js";
 import "./message-handler-mock-support.test-support.js";
@@ -156,4 +165,125 @@ describe("Microsoft Teams drain claim ownership", () => {
     expect(runtimeApiMockState.dispatchReplyWithBufferedBlockDispatcher).not.toHaveBeenCalled();
     expect(lifecycle.abandonedCount()).toBe(0);
   });
+
+  it("preserves abandon retry accounting, backoff, threshold, and restart behavior", async () => {
+    vi.useFakeTimers();
+    const now = Date.UTC(2026, 0, 2);
+    vi.setSystemTime(now);
+    const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-msteams-abandon-"));
+    const stateDir = await fs.realpath(created);
+    type Queue = NonNullable<Parameters<typeof createMSTeamsIngress>[0]["queue"]>;
+    type Payload = Parameters<Queue["enqueue"]>[1];
+    const queue = createChannelIngressQueueForTests<Payload>({
+      channelId: "msteams",
+      accountId: "test-app",
+      stateDir,
+    });
+    const incoming = directActivity("activity-abandon", "retry me");
+    await queue.enqueue(
+      "activity-abandon",
+      { version: 1, receivedAt: now - 2 * 24 * 60 * 60_000, rawActivity: JSON.stringify(incoming) },
+      { laneKey: "dm-conversation", receivedAt: now - 2 * 24 * 60 * 60_000 },
+    );
+    const dispatchMock = runtimeApiMockState.dispatchReplyWithBufferedBlockDispatcher;
+    const priorImplementation = dispatchMock.getMockImplementation();
+    dispatchMock.mockRejectedValue(new Error("Microsoft Teams dispatch failed before adoption"));
+
+    const createIntegratedIngress = () => {
+      const handler = createHandler({
+        channels: { msteams: { dmPolicy: "open", allowFrom: ["*"] } },
+      } as OpenClawConfig);
+      return createMSTeamsIngress({
+        accountId: "test-app",
+        queue,
+        runtime: { error: vi.fn(), log: vi.fn() },
+        dispatch: async (activity, lifecycle) => await handler(context(activity), lifecycle),
+      });
+    };
+    const expectPendingAttempt = async (attempts: number) => {
+      let observed: Awaited<ReturnType<typeof queue.listPending>>[number] | undefined;
+      await vi.waitFor(async () => {
+        const pending = await queue.listPending({ limit: "all" });
+        expect(pending).toEqual([
+          expect.objectContaining({
+            id: "activity-abandon",
+            attempts,
+            lastAttemptAt: expect.any(Number),
+            lastError: "turn-abandoned",
+          }),
+        ]);
+        observed = pending[0];
+      });
+      const lastAttemptAt = observed?.lastAttemptAt;
+      if (lastAttemptAt === undefined) {
+        throw new Error(`Missing Microsoft Teams retry timestamp for attempt ${attempts}`);
+      }
+      return { ...observed, lastAttemptAt };
+    };
+
+    try {
+      const first = createIntegratedIngress();
+      first.start();
+      const firstAttempt = await expectPendingAttempt(1);
+      expect(dispatchMock).toHaveBeenCalledTimes(1);
+      await first.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 999);
+      const second = createIntegratedIngress();
+      second.start();
+      await second.accept(incoming);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(dispatchMock).toHaveBeenCalledTimes(1);
+      await second.stop();
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001);
+      const afterBackoff = createIntegratedIngress();
+      afterBackoff.start();
+      await afterBackoff.accept(incoming);
+      const secondAttempt = await expectPendingAttempt(2);
+      expect(dispatchMock).toHaveBeenCalledTimes(2);
+      await afterBackoff.stop();
+
+      for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
+        const claim = await queue.claim("activity-abandon", { ownerId: `seed-${attempt}` });
+        if (!claim) {
+          throw new Error(`Expected Microsoft Teams seed claim ${attempt}`);
+        }
+        await queue.release(claim, {
+          lastError: "turn-abandoned",
+          releasedAt: secondAttempt.lastAttemptAt,
+        });
+      }
+      vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001);
+      const threshold = createIntegratedIngress();
+      threshold.start();
+      await threshold.accept(incoming);
+      const thresholdAttempt = await expectPendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS);
+      expect(dispatchMock).toHaveBeenCalledTimes(3);
+      await threshold.stop();
+
+      vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001);
+      const beyond = createIntegratedIngress();
+      beyond.start();
+      await beyond.accept(incoming);
+      const beyondAttempt = await expectPendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1);
+      expect(dispatchMock).toHaveBeenCalledTimes(4);
+      await beyond.stop();
+
+      vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000);
+      const blockedRestart = createIntegratedIngress();
+      blockedRestart.start();
+      await blockedRestart.accept(incoming);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(dispatchMock).toHaveBeenCalledTimes(4);
+      await blockedRestart.stop();
+    } finally {
+      dispatchMock.mockReset();
+      if (priorImplementation) {
+        dispatchMock.mockImplementation(priorImplementation);
+      }
+      closeOpenClawStateDatabaseForTest();
+      await fs.rm(stateDir, { recursive: true, force: true });
+      vi.useRealTimers();
+    }
+  });
 });

extensions/signal/src/monitor/event-handler.reply-session-conflict.test.ts

@@ -1,6 +1,15 @@
-import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 // Signal tests cover retry behavior for reply session initialization conflicts.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound";
+import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { beforeEach, describe, expect, it, vi } from "vitest";
+import { startSignalIngressMonitor } from "../signal-ingress.js";
 import type { SignalEventHandlerDeps } from "./event-handler.types.js";
 
 const [
@@ -326,6 +335,127 @@ describe("signal reply session init conflict retry", () => {
     }
   });
 
+  it("preserves durable abandon accounting through backoff, threshold, and restart", async () => {
+    vi.useFakeTimers();
+    const now = Date.UTC(2026, 0, 2);
+    vi.setSystemTime(now);
+    const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-signal-abandon-"));
+    const stateDir = await fs.realpath(created);
+    type Queue = NonNullable<Parameters<typeof startSignalIngressMonitor>[0]["queue"]>;
+    type Payload = Parameters<Queue["enqueue"]>[1];
+    const queue = createChannelIngressQueueForTests<Payload>({
+      channelId: "signal",
+      accountId: "default",
+      stateDir,
+    });
+    const timestamp = 1_700_000_000_777;
+    const event = createSignalReceiveEvent({
+      timestamp,
+      dataMessage: { timestamp, message: "retry through durable ingress", attachments: [] },
+    });
+    const eventId = JSON.stringify(["number:+15550001111", timestamp]);
+    dispatchInboundMessageMock.mockRejectedValue(CONFLICT_ERROR);
+
+    const createIntegratedMonitor = async () => {
+      const tracked = createTrackedTaskHarness();
+      const handler = createSignalEventHandler(
+        createBaseSignalEventHandlerDeps({
+          cfg: { messages: { inbound: { debounceMs: 10 } } },
+          runTrackedTask: tracked.runTrackedTask,
+        }),
+      );
+      const monitor = await startSignalIngressMonitor({
+        accountId: "default",
+        queue,
+        dispatch: async (incoming, lifecycle) => await handler(incoming, lifecycle),
+        runtime: { error: vi.fn(), log: vi.fn() },
+      });
+      return { monitor, tracked };
+    };
+    const finishOuterAttempt = async (tracked: ReturnType<typeof createTrackedTaskHarness>) => {
+      await vi.advanceTimersByTimeAsync(10);
+      expect(tracked.tasks).toHaveLength(1);
+      await vi.advanceTimersByTimeAsync(7_000);
+      await Promise.all(tracked.tasks);
+    };
+    const pendingAttempt = async (attempts: number) => {
+      const pending = await queue.listPending({ limit: "all" });
+      expect(pending).toEqual([
+        expect.objectContaining({
+          id: eventId,
+          attempts,
+          lastAttemptAt: expect.any(Number),
+          lastError: "turn-abandoned",
+        }),
+      ]);
+      const record = pending[0];
+      const lastAttemptAt = record?.lastAttemptAt;
+      if (lastAttemptAt === undefined) {
+        throw new Error(`Missing Signal retry timestamp for attempt ${attempts}`);
+      }
+      return { ...record, lastAttemptAt };
+    };
+
+    try {
+      const first = await createIntegratedMonitor();
+      await first.monitor.receive(event);
+      await finishOuterAttempt(first.tracked);
+      const firstAttempt = await pendingAttempt(1);
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(4);
+      await first.monitor.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 999);
+      const blocked = await createIntegratedMonitor();
+      await vi.advanceTimersByTimeAsync(10);
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(4);
+      expect(blocked.tracked.tasks).toHaveLength(0);
+      await blocked.monitor.stop();
+
+      vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001);
+      const second = await createIntegratedMonitor();
+      await finishOuterAttempt(second.tracked);
+      const secondAttempt = await pendingAttempt(2);
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(8);
+      await second.monitor.stop();
+
+      for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
+        const claim = await queue.claim(eventId, { ownerId: `seed-${attempt}` });
+        if (!claim) {
+          throw new Error(`Expected Signal seed claim ${attempt}`);
+        }
+        await queue.release(claim, {
+          lastError: "turn-abandoned",
+          releasedAt: secondAttempt.lastAttemptAt,
+        });
+      }
+
+      vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001);
+      const threshold = await createIntegratedMonitor();
+      await finishOuterAttempt(threshold.tracked);
+      const thresholdAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS);
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(12);
+      await threshold.monitor.stop();
+
+      vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001);
+      const beyond = await createIntegratedMonitor();
+      await finishOuterAttempt(beyond.tracked);
+      const beyondAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1);
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(16);
+      await beyond.monitor.stop();
+
+      vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000);
+      const blockedRestart = await createIntegratedMonitor();
+      await vi.advanceTimersByTimeAsync(10);
+      expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(16);
+      expect(blockedRestart.tracked.tasks).toHaveLength(0);
+      await blockedRestart.monitor.stop();
+    } finally {
+      closeOpenClawStateDatabaseForTest();
+      await fs.rm(stateDir, { recursive: true, force: true });
+      vi.useRealTimers();
+    }
+  });
+
   it("does not retry non-conflict flush failures", async () => {
     dispatchInboundMessageMock.mockRejectedValue(new Error("some other dispatch failure"));
 

src/channels/message/ingress-queue.test.ts

@@ -119,32 +119,66 @@ describe("channel ingress queue", () => {
   it("keeps channel and account queue identities unambiguous", async () => {
     await withTempState(async (stateDir) => {
       const first = createChannelIngressQueue<{ text: string }>({
-        channelId: "a",
-        accountId: "b:c",
+        channelId: "discord",
+        accountId: "account-a",
         stateDir,
       });
       const second = createChannelIngressQueue<{ text: string }>({
-        channelId: "a:b",
-        accountId: "c",
+        channelId: "discord",
+        accountId: "account-b",
         stateDir,
       });
 
-      expect(await first.enqueue("same-id", { text: "first" })).toMatchObject({
+      expect(
+        await first.enqueue("same-id", { text: "first" }, { laneKey: "channel:same-lane" }),
+      ).toMatchObject({
         kind: "accepted",
       });
-      expect(await second.enqueue("same-id", { text: "second" })).toMatchObject({
+      expect(
+        await second.enqueue("same-id", { text: "second" }, { laneKey: "channel:same-lane" }),
+      ).toMatchObject({
         kind: "accepted",
       });
 
-      await first.complete("same-id");
+      const firstClaim = await first.claim("same-id", { ownerId: "first-worker" });
+      expect(firstClaim).not.toBeNull();
+      if (!firstClaim) {
+        return;
+      }
+      await first.fail(firstClaim, { reason: "poison", failedAt: 20 });
 
       expect(await first.enqueue("same-id", { text: "first duplicate" })).toMatchObject({
-        kind: "completed",
+        kind: "failed",
       });
       expect(await second.enqueue("same-id", { text: "second duplicate" })).toMatchObject({
         kind: "pending",
         record: { payload: { text: "second" } },
       });
+
+      if (!first.resubmit) {
+        return;
+      }
+      await expect(first.resubmit("same-id", { resubmittedAt: 30 })).resolves.toMatchObject({
+        kind: "resubmitted",
+        record: { attempts: 0, laneKey: "channel:same-lane", payload: { text: "first" } },
+      });
+      const resubmittedClaim = await first.claim("same-id", { ownerId: "replacement" });
+      const secondClaim = await second.claim("same-id", { ownerId: "second-worker" });
+      expect(resubmittedClaim).not.toBeNull();
+      expect(secondClaim).not.toBeNull();
+      if (!resubmittedClaim || !secondClaim) {
+        return;
+      }
+      await first.fail(resubmittedClaim, { reason: "poison-again", failedAt: 40 });
+      await second.complete(secondClaim, { completedAt: 40 });
+
+      expect(await first.prune({ failedTtlMs: 1, now: 42 })).toBe(1);
+      expect(await first.enqueue("same-id", { text: "fresh after prune" })).toMatchObject({
+        kind: "accepted",
+      });
+      expect(await second.enqueue("same-id", { text: "completed duplicate" })).toMatchObject({
+        kind: "completed",
+      });
     });
   });
 

src/plugin-sdk/channel-ingress-runtime.test.ts

@@ -76,6 +76,27 @@ describe("plugin-sdk/channel-ingress-runtime", () => {
     expect(fanInChannelIngressLifecycles([]).lifecycle).toBeUndefined();
   });
 
+  it("does not expose cancellation when a source cannot cancellation-settle", async () => {
+    const adopted = vi.fn(async () => {});
+    const cancelled = vi.fn(async () => {});
+    const createLifecycle = (onCancelled?: () => Promise<void>) => ({
+      abortSignal: new AbortController().signal,
+      onAdopted: adopted,
+      onDeferred: vi.fn(),
+      onAdoptionFinalizing: vi.fn(),
+      onFailed: vi.fn(async () => {}),
+      onAbandoned: vi.fn(async () => {}),
+      ...(onCancelled ? { onCancelled } : {}),
+    });
+    const combined = fanInChannelIngressLifecycles([createLifecycle(cancelled), createLifecycle()]);
+
+    expect(combined.lifecycle).not.toHaveProperty("onCancelled");
+    await combined.settle();
+
+    expect(adopted).toHaveBeenCalledTimes(2);
+    expect(cancelled).not.toHaveBeenCalled();
+  });
+
   it("can abandon claims after terminal settlement adoption fails", async () => {
     const abandoned = vi.fn(async () => {});
     const combined = fanInChannelIngressLifecycles([

src/plugin-sdk/channel-ingress-runtime.ts

@@ -151,6 +151,18 @@ export function fanInChannelIngressLifecycles(
   const failAll = async (error: unknown) => {
     await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onFailed?.(error)));
   };
+  const cancellationHandlers = lifecycles.flatMap((lifecycle) =>
+    lifecycle.onCancelled ? [lifecycle.onCancelled] : [],
+  );
+  // Omit aggregate cancellation unless every durable source supports it. Callers
+  // can then use settle/abandon without an acknowledged-but-unsettled claim.
+  const cancelAll =
+    cancellationHandlers.length === lifecycles.length
+      ? async () => {
+          handedOff = true;
+          await Promise.all(cancellationHandlers.map(async (cancel) => await cancel()));
+        }
+      : undefined;
   return {
     lifecycle: {
       abortSignal:
@@ -176,10 +188,7 @@ export function fanInChannelIngressLifecycles(
         handedOff = true;
         await failAll(error);
       },
-      onCancelled: async () => {
-        handedOff = true;
-        await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onCancelled?.()));
-      },
+      ...(cancelAll ? { onCancelled: cancelAll } : {}),
       onAbandoned: async () => {
         handedOff = true;
         await abandonAll();