fix(discord): unblock ingress after retry exhaustion

docs/plugins/sdk-channel-outbound.md

@@ -47,13 +47,15 @@ the transport callback instead of dispatching an event that was not made
 durable. At claim time it decodes the versioned payload, re-runs `inspect`, and
 rejects an id or lane mismatch before delivery.
 
-`deliver` receives `onAdopted`, `onDeferred`, `onAdoptionFinalizing`,
-`onAbandoned`, and `abortSignal`. Returning without an explicit handoff marks a
-terminal no-dispatch event adopted. `admission` is always `exclusive`. A
-deferred handoff keeps the claim held, while shutdown or abort leaves unadopted
-work retryable. The monitor tracks delivery independently from claim settlement
-because adoption can tombstone a row before the channel's delivery promise
-returns.
+`deliver` receives `onAdopted`, `onDeferred`, `onAdoptionFinalizing`, `onFailed`,
+`onCancelled`, `onAbandoned`, and `abortSignal`. Use `onFailed` for delivery
+errors, `onCancelled` for explicit pre-adoption cancellation that must preserve
+retry accounting, and `onAbandoned` when a non-adopted turn should consume a
+retry attempt. Returning without an explicit handoff marks a terminal
+no-dispatch event adopted. `admission` is always `exclusive`. A deferred handoff
+keeps the claim held, while shutdown or abort leaves unadopted work retryable.
+The monitor tracks delivery independently from claim settlement because
+adoption can tombstone a row before the channel's delivery promise returns.
 
 Optional settings include custom append delays, a `drain` option block for
 advanced drain ordering/concurrency/retry policy, an external `abortSignal`, a

extensions/discord/src/monitor/ingress.test.ts

@@ -116,6 +116,31 @@ describe("Discord durable ingress", () => {
     });
   });
 
+  it("rejects unstable message identity before durable allocation", async () => {
+    await withQueue(async (queue) => {
+      const dispatch = vi.fn();
+      const monitor = createDiscordIngressMonitor({
+        accountId: "default",
+        client: {} as never,
+        runtime: runtime(),
+        queue,
+        dispatch,
+      });
+      monitor.start();
+      try {
+        const missingMessageId = { ...createRawMessage("missing"), id: undefined };
+        const missingChannelId = { ...createRawMessage("missing"), channel_id: undefined };
+
+        await expect(monitor.accept(missingMessageId as never)).rejects.toThrow("snowflake");
+        await expect(monitor.accept(missingChannelId as never)).rejects.toThrow("channel_id");
+        expect(await queue.listPending({ limit: "all" })).toEqual([]);
+        expect(dispatch).not.toHaveBeenCalled();
+      } finally {
+        await monitor.stop();
+      }
+    });
+  });
+
   it("recovers a claimed row with a fresh drain and dispatches it exactly once", async () => {
     await withQueue(async (queue) => {
       const monitors: DiscordIngressMonitor[] = [];

extensions/discord/src/monitor/ingress.ts

@@ -3,6 +3,7 @@ import { GatewayDispatchEvents, type APIMessage } from "discord-api-types/v10";
 import {
   createChannelIngressError,
   createChannelIngressMonitor,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
   type ChannelIngressQueue,
   type ChannelIngressMonitorDeliveryResult,
   type ChannelIngressMonitorLifecycle,
@@ -147,6 +148,10 @@ export function createDiscordIngressMonitor(params: {
     },
     appendRetryDelaysMs: [0],
     drain: {
+      retryPolicy: {
+        maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+        deadLetterMinAgeMs: 0,
+      },
       resolveNonRetryableFailure: (error) => {
         if (error instanceof DiscordIngressPayloadError) {
           return { reason: "invalid-event", message: error.message };

extensions/discord/src/monitor/message-dispatcher.ts

@@ -139,7 +139,7 @@ export function createDiscordMessageDispatcher(
           }
           const abortSignal = last.abortSignal;
           if (abortSignal?.aborted) {
-            await admissionLifecycle.onAbandoned();
+            await ingress.lifecycle?.onCancelled?.();
             return;
           }
           try {
@@ -157,7 +157,7 @@ export function createDiscordMessageDispatcher(
                 turnAdoptionLifecycle: admissionLifecycle,
               });
               if (abortSignal?.aborted) {
-                await ingress.abandon(abortSignal.reason);
+                await ingress.lifecycle?.onCancelled?.();
                 return;
               }
               if (!ctx) {
@@ -211,7 +211,7 @@ export function createDiscordMessageDispatcher(
               turnAdoptionLifecycle: admissionLifecycle,
             });
             if (abortSignal?.aborted) {
-              await ingress.abandon(abortSignal.reason);
+              await ingress.lifecycle?.onCancelled?.();
               return;
             }
             if (!ctx) {
@@ -232,7 +232,10 @@ export function createDiscordMessageDispatcher(
             }
             messageRunQueue.enqueue(buildDiscordInboundJob(ctx, { ingressSettlement: ingress }));
           } catch (error) {
-            await admissionLifecycle.onAbandoned();
+            if (abortSignal?.aborted) {
+              await ingress.lifecycle?.onCancelled?.();
+              return;
+            }
             throw error;
           }
         },
@@ -244,7 +247,7 @@ export function createDiscordMessageDispatcher(
     onCancel: (entries) => {
       for (const entry of entries) {
         pendingDebounceEntries.delete(entry);
-        const settlement = Promise.resolve(entry.turnAdoptionLifecycle?.onAbandoned())
+        const settlement = Promise.resolve(entry.turnAdoptionLifecycle?.onCancelled?.())
           .catch((error: unknown) => {
             params.runtime.error(
               danger(`discord ingress cancellation settlement failed: ${String(error)}`),
@@ -271,6 +274,10 @@ export function createDiscordMessageDispatcher(
         const reason = dispatcherShutdown.signal.aborted
           ? (dispatcherShutdown.signal.reason ?? new Error("discord dispatcher shut down"))
           : (options?.abortSignal?.reason ?? new Error("discord dispatch aborted"));
+        if (options?.turnAdoptionLifecycle?.onCancelled) {
+          await options.turnAdoptionLifecycle.onCancelled();
+          return { kind: "deferred" };
+        }
         return { kind: "failed-retryable", error: reason };
       }
       // Filter bot-own messages before they enter the debounce queue.

extensions/discord/src/monitor/message-handler.queue.test.ts

@@ -1,9 +1,21 @@
 // Discord tests cover message handler.queue plugin behavior.
 import { getEventListeners } from "node:events";
+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 type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
 import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
+import {
+  closeOpenClawStateDatabaseForTest,
+  createChannelIngressQueueForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { beforeEach, describe, expect, it, vi } from "vitest";
-import type { DiscordIngressLifecycle } from "./ingress.js";
+import { createDiscordIngressMonitor, type DiscordIngressLifecycle } from "./ingress.js";
 import { createDiscordMessageHandler as createDurableDiscordMessageHandler } from "./message-handler.js";
 import {
   createDiscordMessageHandler,
@@ -35,6 +47,8 @@ function expectStatusPatch(setStatus: MockCallSource, expected: Record<string, u
 
 function createIngressLifecycle(): DiscordIngressLifecycle & {
   onAdopted: ReturnType<typeof vi.fn>;
+  onFailed: ReturnType<typeof vi.fn>;
+  onCancelled: ReturnType<typeof vi.fn>;
   onAbandoned: ReturnType<typeof vi.fn>;
 } {
   return {
@@ -42,10 +56,61 @@ function createIngressLifecycle(): DiscordIngressLifecycle & {
     onAdopted: vi.fn(async () => {}),
     onDeferred: vi.fn(),
     onAdoptionFinalizing: vi.fn(),
+    onFailed: vi.fn(async () => {}),
+    onCancelled: vi.fn(async () => {}),
     onAbandoned: vi.fn(async () => {}),
   };
 }
 
+type DiscordIngressPayload = {
+  version: 1;
+  receivedAt: number;
+  rawMessage: APIMessage;
+};
+
+async function withDiscordQueue<T>(
+  run: (queue: ChannelIngressQueue<DiscordIngressPayload>) => Promise<T>,
+): Promise<T> {
+  const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-discord-handler-"));
+  const stateDir = await fs.realpath(created);
+  const queue = createChannelIngressQueueForTests<DiscordIngressPayload>({
+    channelId: "discord",
+    accountId: "default",
+    stateDir,
+  });
+  try {
+    return await run(queue);
+  } finally {
+    closeOpenClawStateDatabaseForTest();
+    await fs.rm(stateDir, { recursive: true, force: true });
+  }
+}
+
+function createRawMessage(id: string, channelId = "ch-1"): 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().toISOString(),
+    edited_timestamp: null,
+    components: [],
+    pinned: false,
+    type: 0,
+    tts: false,
+  } as unknown as APIMessage;
+}
+
 async function flushQueueWork(): Promise<void> {
   for (let i = 0; i < 40; i += 1) {
     await Promise.resolve();
@@ -314,11 +379,31 @@ describe("createDiscordMessageHandler queue behavior", () => {
       turnAdoptionLifecycle: lifecycle,
     });
 
-    expect(result).toMatchObject({ kind: "failed-retryable" });
+    expect(result).toMatchObject({ kind: "deferred" });
+    expect(lifecycle.onCancelled).toHaveBeenCalledTimes(1);
     expect(lifecycle.onAdopted).not.toHaveBeenCalled();
   });
 
-  it("abandons a buffered ingress claim during deactivation", async () => {
+  it("reports a genuine pre-admission exception only through onFailed", async () => {
+    preflightDiscordMessageMock.mockReset();
+    processDiscordMessageMock.mockReset();
+    const failure = new Error("preflight failed");
+    preflightDiscordMessageMock.mockRejectedValue(failure);
+    const handler = createDiscordMessageHandler(createDiscordHandlerParams());
+    const lifecycle = createIngressLifecycle();
+
+    await expect(
+      handler(createTextMessageData("m-failed") as never, {} as never, {
+        turnAdoptionLifecycle: lifecycle,
+      }),
+    ).resolves.toEqual({ kind: "deferred" });
+
+    expect(lifecycle.onFailed).toHaveBeenCalledExactlyOnceWith(failure);
+    expect(lifecycle.onCancelled).not.toHaveBeenCalled();
+    expect(lifecycle.onAbandoned).not.toHaveBeenCalled();
+  });
+
+  it("cancels a buffered ingress claim during deactivation", async () => {
     preflightDiscordMessageMock.mockReset();
     processDiscordMessageMock.mockReset();
     const params = createDiscordHandlerParams();
@@ -332,11 +417,12 @@ describe("createDiscordMessageHandler queue behavior", () => {
     await handler.deactivate();
 
     expect(preflightDiscordMessageMock).not.toHaveBeenCalled();
-    expect(lifecycle.onAbandoned).toHaveBeenCalledTimes(1);
+    expect(lifecycle.onCancelled).toHaveBeenCalledTimes(1);
+    expect(lifecycle.onAbandoned).not.toHaveBeenCalled();
     expect(lifecycle.onAdopted).not.toHaveBeenCalled();
   });
 
-  it("waits for an active debounce flush and abandons it after shutdown", async () => {
+  it("waits for an active debounce flush and cancels it after shutdown", async () => {
     preflightDiscordMessageMock.mockReset();
     processDiscordMessageMock.mockReset();
     const preflightGate = createDeferred<void>();
@@ -360,7 +446,8 @@ describe("createDiscordMessageHandler queue behavior", () => {
 
     preflightGate.resolve();
     await Promise.all([handling, deactivation]);
-    expect(lifecycle.onAbandoned).toHaveBeenCalledTimes(1);
+    expect(lifecycle.onCancelled).toHaveBeenCalledTimes(1);
+    expect(lifecycle.onAbandoned).not.toHaveBeenCalled();
     expect(lifecycle.onAdopted).not.toHaveBeenCalled();
   });
 
@@ -394,6 +481,136 @@ describe("createDiscordMessageHandler queue behavior", () => {
     expect(stop).toHaveBeenCalledTimes(1);
   });
 
+  it("dead-letters an exhausted preflight failure and releases its Discord lane", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
+    try {
+      await withDiscordQueue(async (queue) => {
+        const attempted: string[] = [];
+        const preflight = vi.fn(async (params: { data: { message?: { id?: string } } }) => {
+          const id = params.data.message?.id ?? "unknown";
+          attempted.push(id);
+          if (id === "poison") {
+            throw new Error("deterministic preflight failure");
+          }
+          return null;
+        });
+        const params = createDiscordHandlerParams();
+        const handler = createDurableDiscordMessageHandler({
+          ...params,
+          client: {} as never,
+          testing: {
+            preflightDiscordMessage: preflight as never,
+            createIngressMonitor: (monitorParams) =>
+              createDiscordIngressMonitor({ ...monitorParams, queue }),
+          },
+        });
+        try {
+          await handler(createRawMessage("poison", "lane-a") as never, {} as never);
+          await handler(createRawMessage("follower", "lane-a") as never, {} as never);
+          await handler(createRawMessage("independent", "lane-b") as never, {} as never);
+
+          for (let attempt = 0; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
+            await vi.advanceTimersByTimeAsync(3 * 60_000);
+          }
+
+          await vi.waitFor(() => expect(attempted).toContain("follower"));
+          expect(attempted.indexOf("independent")).toBeGreaterThanOrEqual(0);
+          expect(attempted.indexOf("independent")).toBeLessThan(attempted.indexOf("follower"));
+          expect(attempted.filter((id) => id === "poison")).toHaveLength(
+            DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+          );
+          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" });
+          const runtimeErrors = mockCalls(params.runtime.error as unknown as MockCallSource).map(
+            ([message]) => String(message),
+          );
+          expect(runtimeErrors.some((message) => message.includes("reached retry limit"))).toBe(
+            true,
+          );
+          expect(runtimeErrors.join("\n")).not.toContain("hello");
+        } finally {
+          await handler.deactivate();
+        }
+      });
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("preserves retry facts when deactivation cancels a durable Discord claim", async () => {
+    await withDiscordQueue(async (queue) => {
+      const raw = createRawMessage("cancelled", "lane-a");
+      await queue.enqueue(
+        "cancelled",
+        { version: 1, receivedAt: 10, rawMessage: raw },
+        { laneKey: "channel:lane-a", receivedAt: 10 },
+      );
+      const failedClaim = await queue.claim("cancelled", { ownerId: "failed-owner" });
+      expect(failedClaim).not.toBeNull();
+      if (!failedClaim) {
+        return;
+      }
+      await queue.release(failedClaim, {
+        lastError: "previous genuine failure",
+        releasedAt: 20,
+      });
+      const before = (await queue.listPending())[0];
+      const firstPreflight = vi.fn(async () => null);
+      const firstParams = createDiscordHandlerParams();
+      firstParams.cfg.messages = { inbound: { debounceMs: 60_000 } };
+      const first = createDurableDiscordMessageHandler({
+        ...firstParams,
+        client: {} as never,
+        testing: {
+          preflightDiscordMessage: firstPreflight as never,
+          createIngressMonitor: (monitorParams) =>
+            createDiscordIngressMonitor({ ...monitorParams, queue }),
+        },
+      });
+
+      await vi.waitFor(async () => expect(await queue.listClaims()).toHaveLength(1));
+      await first.deactivate();
+
+      expect(firstPreflight).not.toHaveBeenCalled();
+      expect(await queue.listPending()).toEqual([
+        expect.objectContaining({
+          id: "cancelled",
+          attempts: before?.attempts,
+          lastAttemptAt: before?.lastAttemptAt,
+          lastError: before?.lastError,
+        }),
+      ]);
+
+      const replacementPreflight = vi.fn(async () => null);
+      const replacementParams = createDiscordHandlerParams();
+      const replacement = createDurableDiscordMessageHandler({
+        ...replacementParams,
+        client: {} as never,
+        testing: {
+          preflightDiscordMessage: replacementPreflight as never,
+          createIngressMonitor: (monitorParams) =>
+            createDiscordIngressMonitor({ ...monitorParams, queue }),
+        },
+      });
+      try {
+        await vi.waitFor(() => expect(replacementPreflight).toHaveBeenCalledTimes(1));
+        await expect(
+          queue.enqueue("cancelled", {} as DiscordIngressPayload),
+        ).resolves.toMatchObject({ kind: "completed" });
+      } finally {
+        await replacement.deactivate();
+      }
+    });
+  });
+
   it("does not abort concurrent runs with a Discord-owned channel timeout", async () => {
     vi.useFakeTimers();
     try {

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

@@ -170,6 +170,149 @@ describe("channel ingress drain", () => {
     });
   });
 
+  it("cancels unadopted work without changing its retry facts", async () => {
+    await withTempState(async (stateDir) => {
+      let clock = 100;
+      const queue = createTestIngressQueue(stateDir, { now: () => clock });
+      await queue.enqueue("evt-cancel", { text: "x" }, { laneKey: "l1", receivedAt: 1 });
+      const failedClaim = await queue.claim("evt-cancel", { ownerId: "failed-owner" });
+      expect(failedClaim).not.toBeNull();
+      if (!failedClaim) {
+        return;
+      }
+      await queue.release(failedClaim, { lastError: "previous failure", releasedAt: clock });
+      const before = (await queue.listPending())[0];
+      for (let cycle = 0; cycle < 3; cycle += 1) {
+        const lifecycles: ChannelIngressDispatchLifecycle[] = [];
+        clock += 1;
+        const drain = createChannelIngressDrain<Payload>({
+          queue,
+          now: () => clock,
+          retryPolicy: { baseMs: 0, maxMs: 0 },
+          dispatchClaimedEvent: async (_event, lifecycle) => {
+            lifecycles.push(lifecycle);
+            return { kind: "deferred" };
+          },
+        });
+
+        await drain.drainOnce();
+        await vi.waitFor(() => expect(lifecycles).toHaveLength(1));
+        await expectDefined(
+          expectDefined(lifecycles[0], "cancelled lifecycle").onCancelled,
+          "cancel callback",
+        )();
+        expect(await queue.listPending()).toEqual([
+          expect.objectContaining({
+            id: "evt-cancel",
+            attempts: before?.attempts,
+            lastAttemptAt: before?.lastAttemptAt,
+            lastError: before?.lastError,
+          }),
+        ]);
+        expect(await queue.listClaims()).toEqual([]);
+        drain.dispose();
+      }
+
+      const terminal = createChannelIngressDrain<Payload>({
+        queue,
+        now: () => clock,
+        retryPolicy: { maxAttempts: 2, deadLetterMinAgeMs: 0, baseMs: 0, maxMs: 0 },
+        dispatchClaimedEvent: async () => {
+          throw new Error("final genuine failure");
+        },
+      });
+      await terminal.drainOnce();
+      await terminal.waitForIdle();
+      expect(await queue.listFailed?.()).toEqual([
+        expect.objectContaining({
+          id: "evt-cancel",
+          attempts: 1,
+          reason: "retry-limit-exceeded",
+          message: "final genuine failure",
+        }),
+      ]);
+      terminal.dispose();
+    });
+  });
+
+  it("keeps the lane owned until a dead-letter write commits", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createTestIngressQueue(stateDir);
+      await queue.enqueue("poison", { text: "bad" }, { laneKey: "shared", receivedAt: 1 });
+      await queue.enqueue("follower", { text: "good" }, { laneKey: "shared", receivedAt: 2 });
+      const fail = queue.fail.bind(queue);
+      let failAttempts = 0;
+      queue.fail = async (...args) => {
+        failAttempts += 1;
+        if (failAttempts < 3) {
+          throw new Error(`transient fail write ${failAttempts}`);
+        }
+        return await fail(...args);
+      };
+      const dispatched: string[] = [];
+      const drain = createChannelIngressDrain<Payload>({
+        queue,
+        retryPolicy: { maxAttempts: 1, deadLetterMinAgeMs: 0 },
+        dispatchClaimedEvent: async (event, lifecycle) => {
+          dispatched.push(event.id);
+          if (event.id === "poison") {
+            throw new Error("poison delivery");
+          }
+          await lifecycle.onAdopted();
+        },
+      });
+
+      await drain.drainOnce();
+      const idle = drain.waitForIdle();
+      await vi.advanceTimersByTimeAsync(0);
+      expect(failAttempts).toBe(1);
+      expect(drain.activeLaneKeys()).toEqual(new Set(["shared"]));
+      expect(await drain.drainOnce()).toEqual({ started: 0 });
+      expect(dispatched).toEqual(["poison"]);
+
+      await vi.advanceTimersByTimeAsync(5_000);
+      await idle;
+      expect(failAttempts).toBe(3);
+      expect(await drain.drainOnce()).toEqual({ started: 1 });
+      await drain.waitForIdle();
+      expect(dispatched).toEqual(["poison", "follower"]);
+      drain.dispose();
+    });
+  });
+
+  it("keeps ownership when every dead-letter write fails", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createTestIngressQueue(stateDir);
+      await queue.enqueue("poison", { text: "bad" }, { laneKey: "shared", receivedAt: 1 });
+      await queue.enqueue("follower", { text: "good" }, { laneKey: "shared", receivedAt: 2 });
+      queue.fail = async () => {
+        throw new Error("persistent fail write");
+      };
+      const dispatched: string[] = [];
+      const drain = createChannelIngressDrain<Payload>({
+        queue,
+        retryPolicy: { maxAttempts: 1, deadLetterMinAgeMs: 0 },
+        dispatchClaimedEvent: async (event) => {
+          dispatched.push(event.id);
+          throw new Error("poison delivery");
+        },
+      });
+
+      await drain.drainOnce();
+      const idle = drain.waitForIdle();
+      for (let attempt = 0; attempt < 8; attempt += 1) {
+        await vi.advanceTimersByTimeAsync(180_000);
+      }
+      await idle;
+
+      expect(dispatched).toEqual(["poison"]);
+      expect(drain.activeLaneKeys()).toEqual(new Set(["shared"]));
+      expect((await queue.listClaims()).map((claim) => claim.id)).toEqual(["poison"]);
+      expect(await drain.drainOnce()).toEqual({ started: 0 });
+      drain.dispose();
+    });
+  });
+
   it("holds lanes by default and releases only opted-in deferred lanes", async () => {
     for (const occupancy of ["hold", "release"] as const) {
       await withTempState(async (stateDir) => {
@@ -697,6 +840,39 @@ describe("channel ingress drain", () => {
     });
   });
 
+  it("keeps retry-accounted abandonment pending beyond the failure threshold", async () => {
+    await withTempState(async (stateDir) => {
+      let clock = 1;
+      const queue = createTestIngressQueue(stateDir, { now: () => clock });
+      await queue.enqueue("abandoned", { text: "x" }, { laneKey: "l", receivedAt: 1 });
+
+      for (let attempt = 0; attempt < 3; attempt += 1) {
+        clock += 1;
+        const drain = createChannelIngressDrain<Payload>({
+          queue,
+          now: () => clock,
+          retryPolicy: { maxAttempts: 1, deadLetterMinAgeMs: 0, baseMs: 0, maxMs: 0 },
+          dispatchClaimedEvent: async (_event, lifecycle) => {
+            await lifecycle.onAbandoned();
+            return { kind: "deferred" };
+          },
+        });
+        await drain.drainOnce();
+        await drain.waitForIdle();
+        drain.dispose();
+      }
+
+      expect(await queue.listPending()).toEqual([
+        expect.objectContaining({
+          id: "abandoned",
+          attempts: 3,
+          lastError: "turn-abandoned",
+        }),
+      ]);
+      expect(await queue.listFailed?.()).toEqual([]);
+    });
+  });
+
   it("bindIngressLifecycleToReplyOptions returns only turnAdoptionLifecycle", async () => {
     const abort = new AbortController();
     const calls: string[] = [];
@@ -708,6 +884,9 @@ describe("channel ingress drain", () => {
       onFailed: () => {
         calls.push("failed");
       },
+      onCancelled: () => {
+        calls.push("cancelled");
+      },
       onAdopted: () => {
         calls.push("adopted");
       },
@@ -721,6 +900,7 @@ describe("channel ingress drain", () => {
     expect(bound.turnAdoptionLifecycle.abortSignal).toBe(abort.signal);
     expect(bound.turnAdoptionLifecycle.admission).toBe("exclusive");
     expect("onFailed" in bound.turnAdoptionLifecycle).toBe(false);
+    expect("onCancelled" in bound.turnAdoptionLifecycle).toBe(false);
     expect("onAdopted" in bound).toBe(false);
     expect(Object.keys(bound)).toEqual(["turnAdoptionLifecycle"]);
     bound.turnAdoptionLifecycle.onDeferred();
@@ -1150,6 +1330,7 @@ describe("channel ingress drain", () => {
       onDeferred: () => {},
       onAdoptionFinalizing: () => {},
       onFailed: () => {},
+      onCancelled: () => {},
       onAbandoned: () => {},
     });
     expect(bound.turnAdoptionLifecycle.admission).toBe("exclusive");

src/channels/message/ingress-drain.ts

@@ -69,6 +69,8 @@ type ChannelIngressDispatchLifecycle = {
   onAdoptionFinalizing: () => void;
   /** Deferred work terminally failed after dispatch returned. */
   onFailed?: (error: unknown) => void | Promise<void>;
+  /** Explicit cancellation before adoption; releases without consuming retry budget. */
+  onCancelled?: () => void | Promise<void>;
   /**
    * Deferred turn finished without ever owning the reply lane.
    * Drain releases the claim for retry.
@@ -347,13 +349,12 @@ export function createChannelIngressDrain<
 
   const releaseClaim = async (
     claim: ChannelIngressQueueClaim<TPayload, TMetadata>,
-    lastError?: string,
+    releaseOptions?: { lastError?: string; recordAttempt?: boolean },
   ) => {
     await commitClaimWriteWithRetry({
       claim,
       label: "release",
-      write: () =>
-        queue.release(claim, lastError === undefined ? {} : { lastError, releasedAt: now() }),
+      write: () => queue.release(claim, { ...releaseOptions, releasedAt: now() }),
       falseMeansReclaimed: false,
     });
   };
@@ -401,7 +402,7 @@ export function createChannelIngressDrain<
     }
     const displayId = claim.id.replace(/^0+(?=\d)/, "") || claim.id;
     log(`spooled update ${displayId} failed; keeping for retry: ${disposition.message}`);
-    await releaseClaim(claim, disposition.message);
+    await releaseClaim(claim, { lastError: disposition.message });
   };
 
   const createSettleOwner = (
@@ -468,6 +469,24 @@ export function createChannelIngressDrain<
     state.stallTimer.unref?.();
   };
 
+  const releaseUnadopted = async (
+    state: ActiveHandlerState<TPayload, TMetadata>,
+    releaseOptions: { lastError?: string; recordAttempt?: boolean },
+  ) => {
+    if (state.phase !== "deferred" && state.phase !== "dispatching") {
+      return;
+    }
+    if (state.guillotined || state.superseded) {
+      return;
+    }
+    clearStallTimer(state);
+    await state
+      .settleOnce(async () => {
+        await releaseClaim(state.claim, releaseOptions);
+      })
+      .catch(() => undefined);
+  };
+
   const createLifecycle = (
     state: ActiveHandlerState<TPayload, TMetadata>,
   ): ChannelIngressDispatchLifecycle => {
@@ -528,19 +547,13 @@ export function createChannelIngressDrain<
           await applyFailureDisposition(state.claim, error);
         });
       },
+      onCancelled: async () => {
+        // Cancellation means ownership ended before delivery, so preserve every
+        // prior retry fact while reopening the canonical row for replacement.
+        await releaseUnadopted(state, { recordAttempt: false });
+      },
       onAbandoned: async () => {
-        if (state.phase !== "deferred" && state.phase !== "dispatching") {
-          return;
-        }
-        if (state.guillotined || state.superseded) {
-          return;
-        }
-        clearStallTimer(state);
-        await state
-          .settleOnce(async () => {
-            await releaseClaim(state.claim, "turn-abandoned");
-          })
-          .catch(() => undefined);
+        await releaseUnadopted(state, { lastError: "turn-abandoned" });
       },
     };
   };

src/channels/message/ingress-monitor.ts

@@ -65,6 +65,7 @@ export type ChannelIngressMonitorLifecycle = {
   onDeferred: () => void;
   onAdoptionFinalizing: () => void;
   onFailed?: (error: unknown) => void | Promise<void>;
+  onCancelled?: () => void | Promise<void>;
   onAbandoned: () => void | Promise<void>;
 };
 
@@ -368,6 +369,16 @@ export function createChannelIngressMonitor<TRaw, TBody, TStoredPayload, TMetada
             settleDeferredClaim();
           }
         }
+        const settleDeferredLifecycle = async (settle: () => void | Promise<void>) => {
+          handedOff = true;
+          deferredHandoff = true;
+          try {
+            await settle();
+            requestDrain();
+          } finally {
+            settleDeferredClaim();
+          }
+        };
         const wrappedLifecycle: ChannelIngressMonitorLifecycle = {
           ...lifecycle,
           admission: "exclusive",
@@ -393,26 +404,9 @@ export function createChannelIngressMonitor<TRaw, TBody, TStoredPayload, TMetada
             deferredHandoff = true;
             lifecycle.onAdoptionFinalizing();
           },
-          onFailed: async (error) => {
-            handedOff = true;
-            deferredHandoff = true;
-            try {
-              await lifecycle.onFailed?.(error);
-              requestDrain();
-            } finally {
-              settleDeferredClaim();
-            }
-          },
-          onAbandoned: async () => {
-            handedOff = true;
-            deferredHandoff = true;
-            try {
-              await lifecycle.onAbandoned();
-              requestDrain();
-            } finally {
-              settleDeferredClaim();
-            }
-          },
+          onFailed: (error) => settleDeferredLifecycle(() => lifecycle.onFailed?.(error)),
+          onCancelled: () => settleDeferredLifecycle(() => lifecycle.onCancelled?.()),
+          onAbandoned: () => settleDeferredLifecycle(() => lifecycle.onAbandoned()),
         };
 
         // Adoption can complete before delivery returns; track both lifetimes so stop

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

@@ -33,10 +33,13 @@ describe("plugin-sdk/channel-ingress-runtime", () => {
       onDeferred: vi.fn(),
       onAdoptionFinalizing: vi.fn(),
       onFailed: vi.fn(async () => {}),
+      onCancelled: vi.fn(async () => {}),
       onAbandoned: vi.fn(async () => {}),
     });
     const first = createLifecycle();
     const second = createLifecycle();
+    const cancellation = fanInChannelIngressLifecycles([first, second]);
+    await cancellation.lifecycle?.onCancelled?.();
     const combined = fanInChannelIngressLifecycles([undefined, first, second]);
 
     combined.lifecycle?.onAdoptionFinalizing();
@@ -49,6 +52,8 @@ describe("plugin-sdk/channel-ingress-runtime", () => {
     expect(second.onAdopted).toHaveBeenCalledOnce();
     expect(first.onAbandoned).not.toHaveBeenCalled();
     expect(second.onAbandoned).not.toHaveBeenCalled();
+    expect(first.onCancelled).toHaveBeenCalledOnce();
+    expect(second.onCancelled).toHaveBeenCalledOnce();
   });
 
   it("settles or abandons claims that no reply lane adopted", async () => {

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

@@ -132,7 +132,11 @@ export function fanInChannelIngressLifecycles(
   const lifecycles = inputs.filter((lifecycle) => lifecycle !== undefined);
   const first = lifecycles[0];
   if (!first) {
-    return { lifecycle: undefined, settle: async () => {}, abandon: async () => {} };
+    return {
+      lifecycle: undefined,
+      settle: async () => {},
+      abandon: async () => {},
+    };
   }
 
   let handedOff = false;
@@ -147,7 +151,6 @@ export function fanInChannelIngressLifecycles(
   const failAll = async (error: unknown) => {
     await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onFailed?.(error)));
   };
-
   return {
     lifecycle: {
       abortSignal:
@@ -173,6 +176,10 @@ export function fanInChannelIngressLifecycles(
         handedOff = true;
         await failAll(error);
       },
+      onCancelled: async () => {
+        handedOff = true;
+        await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onCancelled?.()));
+      },
       onAbandoned: async () => {
         handedOff = true;
         await abandonAll();