diff --git a/docs/channels/telegram.md b/docs/channels/telegram.md
index ef432132c1d9..cb9cd0df50f8 100644
--- a/docs/channels/telegram.md
+++ b/docs/channels/telegram.md
@@ -739,7 +739,7 @@ curl "https://api.telegram.org/bot<bot_token>/getUpdates"
 
     Webhook mode validates request guards, the Telegram secret token, and the JSON body, then commits the update to its durable ingress queue before returning an empty `200`. Successful durable adoption includes `x-openclaw-delivery-accepted: durable`; health, routing, authentication, validation, and storage-error responses omit this header. Reverse proxies and host controllers can require the header to distinguish OpenClaw adoption from a generic empty `200` without inferring acceptance from response timing.
 
-    OpenClaw then processes the update asynchronously through the same per-chat/per-topic bot lanes used by long polling, so slow agent turns do not hold Telegram's delivery ACK.
+    After the durable write, OpenClaw claims and processes updates through the core channel-ingress drain (per-chat/per-topic lanes, complete at turn adoption, pre-adoption stall timeout). Slow agent turns do not hold Telegram's delivery ACK.
 
   </Accordion>
 
diff --git a/docs/plugins/sdk-channel-outbound.md b/docs/plugins/sdk-channel-outbound.md
index 83b3fd04c7b4..283ed1dc211b 100644
--- a/docs/plugins/sdk-channel-outbound.md
+++ b/docs/plugins/sdk-channel-outbound.md
@@ -12,10 +12,14 @@ Channel plugins expose outbound message behavior from
 `openclaw/plugin-sdk/channel-inbound` for receive/context/dispatch
 orchestration.
 
-Core owns queueing, durability, generic retry policy, hooks, receipts, and
-the shared `message` tool. The plugin owns native send/edit/delete calls,
-target normalization, platform threading, selected quotes, notification
-flags, account state, and platform-specific side effects.
+Core owns queueing, durability, the durable **ingress drain**
+(`createChannelIngressDrain` / `openChannelIngressDrain`), generic retry
+policy, turn-adoption lifecycle (`turnAdoptionLifecycle` /
+`bindIngressLifecycleToReplyOptions`), hooks, receipts, and the shared
+`message` tool. The plugin owns native send/edit/delete calls, target
+normalization, platform threading, selected quotes, notification flags,
+account state, accept-side enqueue, lane keys, non-retryable predicates,
+optional supersede authorization, and platform-specific side effects.
 
 ## Adapter
 
diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md
index 040f7236909b..137e3426c58c 100644
--- a/docs/plugins/sdk-runtime.md
+++ b/docs/plugins/sdk-runtime.md
@@ -739,8 +739,10 @@ two-party event loops that do not go through the shared inbound reply runner.
 
     `openChannelIngressQueue<TPayload>(...)` opens a persisted ingress queue scoped to the calling plugin, for buffering inbound events that need at-least-once processing across restarts. When stale-claim recovery uses `shouldRecover`, also provide `shouldRecoverCorrupt` if corrupt claimed payloads should be quarantined: its payload-independent claim identity lets the plugin preserve live owner and lane policy before the queue tombstones the row.
 
+    `openChannelIngressDrain(...)` opens the core channel-agnostic worker over that queue (or creates a queue when none is supplied). The drain owns stale-claim recovery, per-lane claim serialization, complete-at-adoption or complete-on-dispatch-return, retry/dead-letter disposition, optional pre-adoption supersede, and claim→adoption stall timeout. Wire claim ownership into reply generation with `turnAdoptionLifecycle` (via `bindIngressLifecycleToReplyOptions` from `plugin-sdk/channel-outbound`). Channel plugins keep accept-side enqueue, lane derivation, non-retryable classification, and any supersede authorization policy.
+
     <Warning>
-    `openBlobStore`, `openKeyedStore`, `openSyncKeyedStore`, and `openChannelIngressQueue` are available only to bundled plugins and trusted official plugin installations in this release.
+    `openBlobStore`, `openKeyedStore`, `openSyncKeyedStore`, `openChannelIngressQueue`, and `openChannelIngressDrain` are available only to bundled plugins and trusted official plugin installations in this release.
     </Warning>
 
   </Accordion>
diff --git a/extensions/voice-call/doctor-contract-api.test.ts b/extensions/voice-call/doctor-contract-api.test.ts
index d468e18bcff1..1f0decc5e070 100644
--- a/extensions/voice-call/doctor-contract-api.test.ts
+++ b/extensions/voice-call/doctor-contract-api.test.ts
@@ -46,6 +46,9 @@ function installStateRuntime(): void {
       openChannelIngressQueue: (() => {
         throw new Error("openChannelIngressQueue is not used by voice-call doctor tests");
       }) as never,
+      openChannelIngressDrain: (() => {
+        throw new Error("openChannelIngressDrain is not used by voice-call doctor tests");
+      }) as never,
     },
   });
 }
diff --git a/extensions/voice-call/src/manager.restore.test.ts b/extensions/voice-call/src/manager.restore.test.ts
index 73769eb710b7..16ac9db50fd2 100644
--- a/extensions/voice-call/src/manager.restore.test.ts
+++ b/extensions/voice-call/src/manager.restore.test.ts
@@ -28,6 +28,9 @@ function installStateRuntime(): void {
       openChannelIngressQueue: (() => {
         throw new Error("openChannelIngressQueue is not used by voice-call restore tests");
       }) as never,
+      openChannelIngressDrain: (() => {
+        throw new Error("openChannelIngressDrain is not used by voice-call restore tests");
+      }) as never,
     },
   });
 }
diff --git a/extensions/voice-call/src/manager.test-harness.ts b/extensions/voice-call/src/manager.test-harness.ts
index 92fa0399e15c..f576d775f079 100644
--- a/extensions/voice-call/src/manager.test-harness.ts
+++ b/extensions/voice-call/src/manager.test-harness.ts
@@ -89,6 +89,9 @@ function createVoiceCallStateRuntimeForTests(): VoiceCallStateRuntime["state"] {
     openChannelIngressQueue: (() => {
       throw new Error("openChannelIngressQueue is not used by voice-call manager tests");
     }) as VoiceCallStateRuntime["state"]["openChannelIngressQueue"],
+    openChannelIngressDrain: (() => {
+      throw new Error("openChannelIngressDrain is not used by voice-call manager tests");
+    }) as VoiceCallStateRuntime["state"]["openChannelIngressDrain"],
   };
 }
 
diff --git a/extensions/voice-call/src/manager/events.test.ts b/extensions/voice-call/src/manager/events.test.ts
index ddf3bae11e88..431ee956e694 100644
--- a/extensions/voice-call/src/manager/events.test.ts
+++ b/extensions/voice-call/src/manager/events.test.ts
@@ -58,6 +58,9 @@ function installStateRuntime(): void {
       openChannelIngressQueue: (() => {
         throw new Error("openChannelIngressQueue is not used by voice-call event tests");
       }) as never,
+      openChannelIngressDrain: (() => {
+        throw new Error("openChannelIngressDrain is not used by voice-call event tests");
+      }) as never,
     },
   });
 }
diff --git a/extensions/voice-call/src/manager/store.test.ts b/extensions/voice-call/src/manager/store.test.ts
index 95eda7ff6a22..ab072c8afb2e 100644
--- a/extensions/voice-call/src/manager/store.test.ts
+++ b/extensions/voice-call/src/manager/store.test.ts
@@ -33,6 +33,9 @@ function installStateRuntime(): void {
       openChannelIngressQueue: (() => {
         throw new Error("openChannelIngressQueue is not used by voice-call store tests");
       }) as never,
+      openChannelIngressDrain: (() => {
+        throw new Error("openChannelIngressDrain is not used by voice-call store tests");
+      }) as never,
     },
   });
 }
@@ -93,6 +96,9 @@ describe("voice-call call record store", () => {
         openChannelIngressQueue: (() => {
           throw new Error("openChannelIngressQueue is not used by voice-call store tests");
         }) as never,
+        openChannelIngressDrain: (() => {
+          throw new Error("openChannelIngressDrain is not used by voice-call store tests");
+        }) as never,
       },
     });
 
diff --git a/extensions/voice-call/src/runtime-state.ts b/extensions/voice-call/src/runtime-state.ts
index b59eadf2c61d..73d6af764300 100644
--- a/extensions/voice-call/src/runtime-state.ts
+++ b/extensions/voice-call/src/runtime-state.ts
@@ -7,7 +7,11 @@ import { createPluginRuntimeStore, type PluginRuntime } from "openclaw/plugin-sd
 export type VoiceCallStateRuntime = {
   state: Pick<
     PluginRuntime["state"],
-    "resolveStateDir" | "openKeyedStore" | "openSyncKeyedStore" | "openChannelIngressQueue"
+    | "resolveStateDir"
+    | "openKeyedStore"
+    | "openSyncKeyedStore"
+    | "openChannelIngressQueue"
+    | "openChannelIngressDrain"
   >;
 };
 
diff --git a/extensions/voice-call/src/webhook.hangup-once.lifecycle.test.ts b/extensions/voice-call/src/webhook.hangup-once.lifecycle.test.ts
index 3c43b7b3a17c..43308867c7ce 100644
--- a/extensions/voice-call/src/webhook.hangup-once.lifecycle.test.ts
+++ b/extensions/voice-call/src/webhook.hangup-once.lifecycle.test.ts
@@ -26,6 +26,11 @@ function installStateRuntime(): void {
           "openChannelIngressQueue is not used by voice-call webhook lifecycle tests",
         );
       }) as never,
+      openChannelIngressDrain: (() => {
+        throw new Error(
+          "openChannelIngressDrain is not used by voice-call webhook lifecycle tests",
+        );
+      }) as never,
     },
   });
 }
diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs
index dd93852126d1..26039d850f63 100644
--- a/scripts/plugin-sdk-surface-report.mjs
+++ b/scripts/plugin-sdk-surface-report.mjs
@@ -162,8 +162,9 @@ const defaultPublicDeprecatedExportsByEntrypointBudget = Object.freeze({
   "channel-lifecycle": 23,
   // Registry sweep: 77 packages, zero fetch failures; channel-ingress and dead aliases
   // had zero consumers.
-  "channel-message": 230,
-  "channel-message-runtime": 227,
+  // +11 each: durable channel-ingress drain seam (drain/lifecycle/claim/retry) mirrored by compat (#108656).
+  "channel-message": 241,
+  "channel-message-runtime": 238,
   "channel-pairing-paths": 1,
   // Deprecated pairing/conversation exports from the SQLite pairing migration
   // landed on main (#105802) without entrypoint pins; not touched by this PR.
@@ -253,7 +254,10 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
       // +4: shared audio-energy stats and speech-threshold gate through realtime-voice.
       // +2: supplemental sender decision and outbound text chunk sequencer.
       // +2: shared realtime voice session harness through realtime-voice.
-      8014,
+      // +24: narrowed durable channel-ingress drain seam — factory, lifecycle binding,
+      // tuning constants, and telegram-consumed claim helpers with compat mirrors,
+      // after harvesting exports orphaned by the split-out WhatsApp adapter (#108656).
+      8038,
       env,
     ),
     publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
@@ -281,7 +285,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
       // +3: PCM16/mu-law energy readers and speech-threshold gate factory.
       // +2: supplemental sender decision and outbound text chunk sequencer.
       // +1: shared realtime voice session harness through realtime-voice.
-      4479,
+      // +9: narrowed drain seam functions and compat mirrors after the
+      // WhatsApp-split harvest (#108656).
+      4489,
       env,
     ),
     publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv(
@@ -298,7 +304,9 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
       // +8: shared channel helpers mirrored by deprecated barrels.
       // +3: receipt/snapshot exports through deprecated channel barrels.
       // +1: unified implicit-mention config type through deprecated config-types.
-      2990,
+      // +24: narrowed drain seam compat mirrors in the channel-message
+      // deprecation-window barrels (#108656).
+      3014,
       env,
     ),
     publicWildcardReexports: readPluginSdkSurfaceBudgetEnv(
diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts
index 79df9fa3624d..711753bc3378 100644
--- a/src/auto-reply/get-reply-options.types.ts
+++ b/src/auto-reply/get-reply-options.types.ts
@@ -46,17 +46,35 @@ export type QueuedReplyDeliveryCorrelation = {
   begin: () => (() => void) | void;
 };
 
-/** Lifecycle hooks for queued follow-up replies. */
-export type QueuedReplyLifecycle = {
-  /** Stable cancellation owner used to keep collect-mode batches authorization-safe. */
-  ownerKey?: string;
-  /** Return false when the external owner rejects this queue identity. */
-  onEnqueued?: () => boolean | void;
-  /** Retires this source's cancellation ownership while retaining its live identity. */
+/**
+ * Exclusive: each lifecycle is its own collect-admission identity.
+ * Cancel-only: share collect identity via ownerKey (gateway chat.send).
+ */
+export type TurnAdoptionAdmission = "exclusive" | "cancel-only";
+
+/**
+ * Canonical turn-ownership lifecycle (adopt / defer / abandon / settle).
+ * Single surface for durable ingress, gateway cancel identity, and reply-lane transfer.
+ */
+export type TurnAdoptionLifecycle = {
+  /**
+   * Admission isolation mode (closed). Exclusive isolates collect identity per
+   * lifecycle; cancel-only shares via ownerKey. Never inferred from onAbandoned.
+   * Durable ingress sets exclusive; gateway cancel identity sets cancel-only.
+   */
+  admission?: TurnAdoptionAdmission;
+  onAdopted: () => void | Promise<void>;
+  /** Return false to reject followup enqueue. */
+  onDeferred?: () => boolean | void;
+  /** Deferred turn finished without owning the reply lane. */
+  onAbandoned?: () => void;
+  /** Always fires when the followup ownership cycle ends (admitted or not). Gateway cleanup. */
+  onSettled?: () => void;
+  /** Retires cancellation ownership while retaining live identity. */
   onCancellationRetired?: () => void;
-  /** Called after the queued turn owns the reply lane, before model/tool execution. */
-  onAdmitted?: () => void | Promise<void>;
-  onComplete?: () => void;
+  /** Stable cancellation owner for collect-mode batches. */
+  ownerKey?: string;
+  abortSignal?: AbortSignal;
 };
 
 /** Partial assistant payload emitted during streaming or replacement updates. */
@@ -94,11 +112,9 @@ export type GetReplyOptions = {
   /** Notifies when an agent run actually starts (useful for webchat command handling). */
   onAgentRunStart?: (runId: string) => void;
   /**
-   * Called after the restart-recovery delivery-context persist attempt
-   * completes (context may be absent when source delivery is suppressed).
-   * Channels may complete ingress ownership here without waiting for settle.
+   * Canonical adoption lifecycle (adopted / deferred / abandoned / settled + pre-adoption abort).
    */
-  onTurnAdopted?: () => void | Promise<void>;
+  turnAdoptionLifecycle?: TurnAdoptionLifecycle;
   /** Shared lifecycle owner for the current user-turn transcript append. */
   userTurnTranscriptRecorder?: UserTurnTranscriptRecorder;
   /** Current user turn is already durable; replay it without appending another copy. */
@@ -295,8 +311,6 @@ export type GetReplyOptions = {
   taskSuggestionDeliveryMode?: TaskSuggestionDeliveryMode;
   /** Starts delivery tracking when this turn later drains as a queued followup. */
   queuedDeliveryCorrelations?: QueuedReplyDeliveryCorrelation[];
-  /** Tracks ownership transfer when this turn later drains as a queued followup. */
-  queuedFollowupLifecycle?: QueuedReplyLifecycle;
   /** Called after a queued followup owns the reply lane, before its model run starts. */
   onQueuedFollowupAdmitted?: () => Promise<void> | void;
   /** Allow channel-owned progress UI while final/source reply delivery remains message-tool-only. */
diff --git a/src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts b/src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts
index 581fe92bc408..00fb5d24fab7 100644
--- a/src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts
+++ b/src/auto-reply/reply/agent-runner.misc.runreplyagent.test.ts
@@ -3402,7 +3402,7 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
     sendPolicyDenied?: boolean;
     isHeartbeat?: boolean;
     replyOperation?: ReturnType<typeof createReplyOperation>;
-    queuedLifecycle?: FollowupRun["queuedLifecycle"];
+    turnAdoptionLifecycle?: FollowupRun["turnAdoptionLifecycle"];
   }) {
     const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-stranded-"));
     const storePath = path.join(tmp, "sessions.json");
@@ -3459,7 +3459,9 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
       ...(params.strandedReplyRetry ? { strandedReplyRetry: true } : {}),
       enqueuedAt: Date.now(),
       ...(params.transcriptPrompt ? { transcriptPrompt: params.transcriptPrompt } : {}),
-      ...(params.queuedLifecycle ? { queuedLifecycle: params.queuedLifecycle } : {}),
+      ...(params.turnAdoptionLifecycle
+        ? { turnAdoptionLifecycle: params.turnAdoptionLifecycle }
+        : {}),
       run: {
         agentId: "main",
         agentDir: "/tmp/agent",
@@ -3527,9 +3529,9 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
 
   it("enqueues a one-shot recovery retry by default for substantive stranded finals", async () => {
     const parentOnComplete = vi.fn();
-    const parentLifecycle = { onComplete: parentOnComplete };
+    const parentLifecycle = { onAdopted: async () => {}, onSettled: parentOnComplete };
     const { finalAssistantText } = await runPrivateFinalCase({
-      queuedLifecycle: parentLifecycle,
+      turnAdoptionLifecycle: parentLifecycle,
     });
 
     expect(warnPrivateFinalSpy).toHaveBeenCalledTimes(1);
@@ -3542,8 +3544,8 @@ describe("runReplyAgent private message_tool_only final warning (#85714)", () =>
     expect(retryRun?.prompt).toContain("message(action=send)");
     expect(retryRun?.prompt).toContain(finalAssistantText);
     // System retry must not inherit the client turn's one-shot lifecycle identity.
-    expect(retryRun?.queuedLifecycle).toBeUndefined();
-    expect(parentLifecycle.onComplete).toBe(parentOnComplete);
+    expect(retryRun?.turnAdoptionLifecycle).toBeUndefined();
+    expect(parentLifecycle.onSettled).toBe(parentOnComplete);
     expect(parentOnComplete).not.toHaveBeenCalled();
   });
 
diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts
index e55940cb6c6e..477cd24d06be 100644
--- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts
+++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts
@@ -623,12 +623,12 @@ describe("runReplyAgent active steering", () => {
         return true;
       },
     );
-    const onTurnAdopted = vi.fn(async () => {
+    const onAdopted = vi.fn(async () => {
       events.push("adoption-finalizer");
       throw finalizerError;
     });
     const { run, typing } = createMinimalRun({
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       isActive: true,
       isStreaming: true,
       shouldSteer: true,
@@ -638,7 +638,7 @@ describe("runReplyAgent active steering", () => {
     await expect(run()).resolves.toBeUndefined();
 
     expect(events).toEqual(["transcript-committed", "adoption-finalizer"]);
-    expect(onTurnAdopted).toHaveBeenCalledTimes(1);
+    expect(onAdopted).toHaveBeenCalledTimes(1);
     expect(state.queueEmbeddedAgentMessageMock).toHaveBeenCalledTimes(1);
     expect(vi.mocked(enqueueFollowupRun)).not.toHaveBeenCalled();
     expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
@@ -658,9 +658,9 @@ describe("runReplyAgent active steering", () => {
       target: "none",
       gatewayHealth: "live",
     });
-    const onTurnAdopted = vi.fn();
+    const onAdopted = vi.fn();
     const { run } = createMinimalRun({
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       isActive: true,
       isStreaming: true,
       shouldSteer: true,
@@ -680,7 +680,7 @@ describe("runReplyAgent active steering", () => {
 
     expect(state.beforeAgentReplyRunMock).toHaveBeenCalledOnce();
     expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
-    expect(onTurnAdopted).not.toHaveBeenCalled();
+    expect(onAdopted).not.toHaveBeenCalled();
   });
 
   it("admits an ordinary rejected steering turn with durable recovery state", async () => {
@@ -690,13 +690,13 @@ describe("runReplyAgent active steering", () => {
     };
     const sessionStore = { main: sessionEntry };
     const storePath = await createSessionStoreFile(sessionEntry);
-    const onTurnAdopted = vi.fn(async () => {
+    const onAdopted = vi.fn(async () => {
       expect((await readStoredMainSession(storePath)).restartRecoveryBeforeAgentReplyState).toBe(
         "admitted",
       );
     });
     const { followupRun, run, sourceTurnId } = createMinimalRun({
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       isActive: true,
       isStreaming: true,
       shouldSteer: true,
@@ -725,7 +725,7 @@ describe("runReplyAgent active steering", () => {
     await expect(run()).resolves.toEqual(expect.objectContaining({ text: "final" }));
 
     expect(state.beforeAgentReplyRunMock).not.toHaveBeenCalled();
-    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(onAdopted).toHaveBeenCalledOnce();
     expect(state.runEmbeddedAgentMock).toHaveBeenCalledOnce();
   });
 });
@@ -1348,10 +1348,10 @@ describe("runReplyAgent pending final delivery capture", () => {
     const completedEntry = await readStoredMainSession(storePath);
     expect(completedEntry.restartRecoveryTerminalRunIds).toEqual([first.sourceTurnId]);
     state.runEmbeddedAgentMock.mockClear();
-    const onTurnAdopted = vi.fn();
+    const onAdopted = vi.fn();
     const completedStore = { main: completedEntry };
     const duplicate = createMinimalRun({
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       sessionCtx,
       runOverrides: { messageProvider: "discord" },
       sessionEntry: completedEntry,
@@ -1370,7 +1370,7 @@ describe("runReplyAgent pending final delivery capture", () => {
 
     await expect(duplicate.run()).resolves.toBeUndefined();
 
-    expect(onTurnAdopted).not.toHaveBeenCalled();
+    expect(onAdopted).not.toHaveBeenCalled();
     expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
     expect((await readStoredMainSession(storePath)).restartRecoveryTerminalRunIds).toEqual([
       first.sourceTurnId,
@@ -1402,11 +1402,11 @@ describe("runReplyAgent pending final delivery capture", () => {
     };
     const sessionStore = { main: sessionEntry };
     const storePath = await createSessionStoreFile(sessionEntry);
-    const onTurnAdopted = vi.fn();
+    const onAdopted = vi.fn();
     const duplicate = createMinimalRun({
       isActive: true,
       shouldSteer: true,
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       sessionCtx,
       runOverrides: { messageProvider: "discord" },
       sessionEntry,
@@ -1426,7 +1426,7 @@ describe("runReplyAgent pending final delivery capture", () => {
     await expect(duplicate.run()).resolves.toBeUndefined();
 
     expect(duplicate.sourceTurnId).toBe(sourceTurnId);
-    expect(onTurnAdopted).not.toHaveBeenCalled();
+    expect(onAdopted).not.toHaveBeenCalled();
     expect(state.queueEmbeddedAgentMessageMock).not.toHaveBeenCalled();
     expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
     expect(await readStoredMainSession(storePath)).toMatchObject({
@@ -1461,11 +1461,11 @@ describe("runReplyAgent pending final delivery capture", () => {
     };
     const sessionStore = { main: sessionEntry };
     const storePath = await createSessionStoreFile(sessionEntry);
-    const onTurnAdopted = vi.fn();
+    const onAdopted = vi.fn();
     const duplicate = createMinimalRun({
       isActive: true,
       shouldSteer: true,
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       sessionCtx,
       runOverrides: { messageProvider: "discord" },
       sessionEntry,
@@ -1477,7 +1477,7 @@ describe("runReplyAgent pending final delivery capture", () => {
     await expect(duplicate.run()).resolves.toBeUndefined();
 
     expect(duplicate.sourceTurnId).toBe(sourceTurnId);
-    expect(onTurnAdopted).not.toHaveBeenCalled();
+    expect(onAdopted).not.toHaveBeenCalled();
     expect(state.queueEmbeddedAgentMessageMock).not.toHaveBeenCalled();
     expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
     const stored = await readStoredMainSession(storePath);
@@ -1864,9 +1864,9 @@ describe("runReplyAgent pending final delivery capture", () => {
     };
     const sessionStore = { main: sessionEntry };
     const storePath = await createSessionStoreFile(sessionEntry);
-    const onTurnAdopted = vi.fn();
+    const onAdopted = vi.fn();
     const { run } = createMinimalRun({
-      opts: { onTurnAdopted },
+      opts: { turnAdoptionLifecycle: { onAdopted } },
       sessionCtx: {
         Provider: "webchat",
         OriginatingChannel: "webchat",
@@ -1880,7 +1880,7 @@ describe("runReplyAgent pending final delivery capture", () => {
 
     await expect(run()).rejects.toThrow("restart recovery claim changed before agent adoption");
 
-    expect(onTurnAdopted).not.toHaveBeenCalled();
+    expect(onAdopted).not.toHaveBeenCalled();
     expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled();
     expect(await readStoredMainSession(storePath)).toMatchObject({
       abortedLastRun: true,
@@ -1945,7 +1945,7 @@ describe("runReplyAgent pending final delivery capture", () => {
     }
   });
 
-  it("fires onTurnAdopted after restart recovery delivery context persist completes", async () => {
+  it("fires onAdopted after restart recovery delivery context persist completes", async () => {
     const sessionEntry: SessionEntry = {
       sessionId: "session",
       updatedAt: Date.now(),
@@ -1959,7 +1959,7 @@ describe("runReplyAgent pending final delivery capture", () => {
       messageId: "1503645939964055592",
     });
     const events: string[] = [];
-    const onTurnAdopted = vi.fn(async () => {
+    const onAdopted = vi.fn(async () => {
       const storedAtAdoption = await readStoredMainSession(storePath);
       expect(storedAtAdoption.restartRecoveryDeliveryContext).toEqual({
         channel: "discord",
@@ -2000,7 +2000,10 @@ describe("runReplyAgent pending final delivery capture", () => {
     });
 
     const { followupRun, run, sourceTurnId } = createMinimalRun({
-      opts: { onTurnAdopted, sourceReplyDeliveryMode: "message_tool_only" },
+      opts: {
+        turnAdoptionLifecycle: { onAdopted },
+        sourceReplyDeliveryMode: "message_tool_only",
+      },
       sessionCtx: {
         Provider: "discord",
         OriginatingChannel: "discord",
@@ -2028,7 +2031,7 @@ describe("runReplyAgent pending final delivery capture", () => {
 
     await run();
 
-    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(onAdopted).toHaveBeenCalledOnce();
     expect(events).toEqual(["adopted", "agent-run"]);
     expect(
       (await readStoredMainSession(storePath)).restartRecoverySourceReplyDeliveryMode,
@@ -2054,13 +2057,16 @@ describe("runReplyAgent pending final delivery capture", () => {
     };
     const sessionStore = { main: sessionEntry };
     const storePath = await createSessionStoreFile(sessionEntry);
-    const onTurnAdopted = vi.fn(async () => {
+    const onAdopted = vi.fn(async () => {
       expect(
         (await readStoredMainSession(storePath)).restartRecoverySameChannelThreadRequired,
       ).toBe(true);
     });
     const { followupRun, run, sourceTurnId } = createMinimalRun({
-      opts: { onTurnAdopted, sourceReplyDeliveryMode: "message_tool_only" },
+      opts: {
+        turnAdoptionLifecycle: { onAdopted },
+        sourceReplyDeliveryMode: "message_tool_only",
+      },
       sessionCtx: {
         Provider: "slack",
         OriginatingChannel: "slack",
@@ -2085,7 +2091,7 @@ describe("runReplyAgent pending final delivery capture", () => {
 
     await run();
 
-    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(onAdopted).toHaveBeenCalledOnce();
     expect(
       (await readStoredMainSession(storePath)).restartRecoverySameChannelThreadRequired,
     ).toBeUndefined();
@@ -2128,8 +2134,10 @@ describe("runReplyAgent pending final delivery capture", () => {
     state.runEmbeddedAgentMock.mockImplementationOnce(runHookBackedEmbeddedAgent);
     const { followupRun, run, sourceTurnId } = createMinimalRun({
       opts: {
-        onTurnAdopted: async () => {
-          events.push("adopted");
+        turnAdoptionLifecycle: {
+          onAdopted: async () => {
+            events.push("adopted");
+          },
         },
       },
       sessionCtx: {
@@ -2266,7 +2274,7 @@ describe("runReplyAgent pending final delivery capture", () => {
     });
   });
 
-  it("fires onTurnAdopted for suppressed-delivery runs before the agent turn", async () => {
+  it("fires onAdopted for suppressed-delivery runs before the agent turn", async () => {
     const sessionEntry: SessionEntry = {
       sessionId: "session",
       updatedAt: Date.now(),
@@ -2274,7 +2282,7 @@ describe("runReplyAgent pending final delivery capture", () => {
     const sessionStore = { main: sessionEntry };
     const storePath = await createSessionStoreFile(sessionEntry);
     const events: string[] = [];
-    const onTurnAdopted = vi.fn(async () => {
+    const onAdopted = vi.fn(async () => {
       const storedAtAdoption = await readStoredMainSession(storePath);
       expect(storedAtAdoption.restartRecoveryDeliveryContext).toBeUndefined();
       expect(storedAtAdoption.restartRecoveryDeliveryRunId).toBeUndefined();
@@ -2290,7 +2298,7 @@ describe("runReplyAgent pending final delivery capture", () => {
 
     const { run } = createMinimalRun({
       opts: {
-        onTurnAdopted,
+        turnAdoptionLifecycle: { onAdopted },
         sourceReplyDeliveryMode: "message_tool_only",
       },
       sessionCtx: {
@@ -2311,7 +2319,7 @@ describe("runReplyAgent pending final delivery capture", () => {
 
     await run();
 
-    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(onAdopted).toHaveBeenCalledOnce();
     expect(events).toEqual(["adopted", "agent-run"]);
   });
 
diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts
index c90f459b2671..6b6899b3f070 100644
--- a/src/auto-reply/reply/agent-runner.ts
+++ b/src/auto-reply/reply/agent-runner.ts
@@ -27,6 +27,7 @@ import { resolveFastModeState } from "../../agents/fast-mode.js";
 import { resolveModelAuthMode } from "../../agents/model-auth.js";
 import { isCliProvider } from "../../agents/model-selection.js";
 import { deriveContextPromptTokens, hasNonzeroUsage } from "../../agents/usage.js";
+import { isIngressAdoptionLostError } from "../../channels/message/ingress-drain.js";
 import { enqueueCommitmentExtraction } from "../../commitments/runtime.js";
 import type { OpenClawConfig } from "../../config/config.js";
 import {
@@ -1233,6 +1234,8 @@ export async function runReplyAgent(params: {
     replyThreadingOverride,
     replyOperation: providedReplyOperation,
   } = params;
+  // One lifecycle for all adoption sites in this run.
+  const turnAdoptionLifecycle = opts?.turnAdoptionLifecycle;
   let activeSessionEntry = sessionEntry;
   const activeSessionStore = sessionStore;
   let activeIsNewSession = isNewSession;
@@ -1410,7 +1413,7 @@ export async function runReplyAgent(params: {
       {
         steeringMode: "all",
         ...(followupRun.images?.length ? { images: followupRun.images } : {}),
-        ...(opts?.onTurnAdopted ? { waitForTranscriptCommit: true } : {}),
+        ...(turnAdoptionLifecycle ? { waitForTranscriptCommit: true } : {}),
         ...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}),
         ...(followupRun.run.sourceReplyDeliveryMode
           ? { sourceReplyDeliveryMode: followupRun.run.sourceReplyDeliveryMode }
@@ -1424,10 +1427,23 @@ export async function runReplyAgent(params: {
     if (steerOutcome.queued) {
       activeReplyOperation?.recordActivity();
       try {
-        await opts?.onTurnAdopted?.();
+        await turnAdoptionLifecycle?.onAdopted();
       } catch (error) {
-        // Transcript-backed steering is already irrevocably queued here.
-        // Replaying ingress would duplicate the injected user turn.
+        if (isIngressAdoptionLostError(error)) {
+          // Claim was tombstoned/superseded/guillotined after transcript commit.
+          // Cancel the active run so steered tools do not keep executing; do not
+          // rethrow — replaying ingress would duplicate the injected user turn.
+          const abortKey = sessionKey ?? queueKey;
+          if (abortKey) {
+            replyRunRegistry.abort(abortKey);
+          }
+          logVerbose(
+            `queue: active session ${steerSessionId} adoption lost after transcript commit (${error.code}); aborting steered turn without ingress replay`,
+          );
+          typing.cleanup();
+          return undefined;
+        }
+        // Ordinary callback failures: transcript-backed steering is irrevocable.
         logVerbose(
           `queue: active session ${steerSessionId} adoption finalizer failed after transcript commit: ${String(
             error,
@@ -1870,7 +1886,7 @@ export async function runReplyAgent(params: {
     // Adoption marks run start and must never be spool-replayed (would re-run tools).
     // Suppressed delivery persists only the user transcript; crashed suppressed runs die
     // silently. Deliverable turns atomically persist transcript plus recovery ownership.
-    await opts?.onTurnAdopted?.();
+    await turnAdoptionLifecycle?.onAdopted();
     const runOutcome = await withBeforeAgentReplyObserver(
       {
         beforeDispatch: async () => {
diff --git a/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts
index 939051741c78..ce15087493cf 100644
--- a/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts
+++ b/src/auto-reply/reply/dispatch-from-config.base.test-utils.ts
@@ -1273,9 +1273,10 @@ describe("dispatchReplyFromConfig", () => {
         cfg: emptyConfig,
         dispatcher,
         replyOptions: {
-          queuedFollowupLifecycle: {
-            onEnqueued: vi.fn(),
-            onComplete: vi.fn(),
+          turnAdoptionLifecycle: {
+            onAdopted: async () => {},
+            onDeferred: vi.fn(),
+            onSettled: vi.fn(),
           },
         },
         replyResolver,
diff --git a/src/auto-reply/reply/dispatch-from-config.lifecycle.ts b/src/auto-reply/reply/dispatch-from-config.lifecycle.ts
index 45fa5068b134..56b0f53bded9 100644
--- a/src/auto-reply/reply/dispatch-from-config.lifecycle.ts
+++ b/src/auto-reply/reply/dispatch-from-config.lifecycle.ts
@@ -160,7 +160,7 @@ export function createDispatchReplyOperationCoordinator(params: {
     const allowGatewayQueueResolution =
       phase === "dispatch" &&
       replyTurnKind === "visible" &&
-      params.replyOptions?.queuedFollowupLifecycle !== undefined &&
+      params.replyOptions?.turnAdoptionLifecycle !== undefined &&
       replyRunRegistry.get(params.dispatchOperationSessionKey) !== undefined;
     if (allowGatewayQueueResolution) {
       // Gateway turns need to reach getReplyFromConfig while the owner is active;
diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts
index 8ef66e27f1a0..b8ca962752f5 100644
--- a/src/auto-reply/reply/followup-runner.test.ts
+++ b/src/auto-reply/reply/followup-runner.test.ts
@@ -410,12 +410,12 @@ async function loadFreshFollowupRunnerModuleForTest() {
     runCliAgent: (params: unknown) => runCliAgentMock(params),
   }));
   vi.doMock("./queue.js", () => ({
-    admitFollowupRunLifecycle: async (run: Pick<FollowupRun, "queuedLifecycle">) => {
-      await run.queuedLifecycle?.onAdmitted?.();
+    admitFollowupRunLifecycle: async (run: Pick<FollowupRun, "turnAdoptionLifecycle">) => {
+      await run.turnAdoptionLifecycle?.onAdopted?.();
     },
     clearFollowupQueue: clearFollowupQueueForFollowupTest,
-    completeFollowupRunLifecycle: (run: Pick<FollowupRun, "queuedLifecycle">) =>
-      run.queuedLifecycle?.onComplete?.(),
+    completeFollowupRunLifecycle: (run: Pick<FollowupRun, "turnAdoptionLifecycle">) =>
+      run.turnAdoptionLifecycle?.onSettled?.(),
     enqueueFollowupRun: enqueueFollowupRunForFollowupTest,
     isFollowupRunAborted: (run: Pick<FollowupRun, "abortSignal" | "queueAbortSignal">) =>
       run.abortSignal?.aborted === true || run.queueAbortSignal?.aborted === true,
@@ -902,13 +902,15 @@ describe("createFollowupRunner reply-lane admission", () => {
 
     const pending = runner(
       createQueuedRun({
-        queuedLifecycle: {
-          onAdmitted: async () => {
+        turnAdoptionLifecycle: {
+          onAdopted: async () => {
             events.push("admission-started");
             await admissionBarrier;
             events.push("admitted");
           },
-          onComplete: () => events.push("complete"),
+          onSettled: () => events.push("complete"),
+          admission: "exclusive",
+          onAbandoned: () => {},
         },
         run: { provider: "anthropic", model: "claude" },
       }),
@@ -942,13 +944,15 @@ describe("createFollowupRunner reply-lane admission", () => {
     const pending = runner(
       createQueuedRun({
         abortSignal: abortController.signal,
-        queuedLifecycle: {
-          onAdmitted: async () => {
+        turnAdoptionLifecycle: {
+          onAdopted: async () => {
             events.push("admission-started");
             await admissionBarrier;
             events.push("admitted");
           },
-          onComplete: () => events.push("complete"),
+          onSettled: () => events.push("complete"),
+          admission: "exclusive",
+          onAbandoned: () => {},
         },
         run: { provider: "anthropic", model: "claude" },
       }),
@@ -1297,7 +1301,7 @@ describe("createFollowupRunner reply-lane admission", () => {
             model: "claude",
             sessionKey: "main",
           },
-          queuedLifecycle: { onComplete },
+          turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
         }),
       ),
     ).rejects.toThrow("session load failed");
@@ -2564,8 +2568,9 @@ describe("createFollowupRunner runtime config", () => {
         createQueuedRun({
           originatingChannel: "telegram",
           originatingTo: "chat-1",
-          queuedLifecycle: {
-            onComplete: () => {
+          turnAdoptionLifecycle: {
+            onAdopted: async () => {},
+            onSettled: () => {
               operationResultDuringCompletion = replyRunRegistryForTest.get("main")?.result;
             },
           },
@@ -2660,8 +2665,9 @@ describe("createFollowupRunner runtime config", () => {
         createQueuedRun({
           originatingChannel: "telegram",
           originatingTo: "chat-1",
-          queuedLifecycle: {
-            onComplete: () => {
+          turnAdoptionLifecycle: {
+            onAdopted: async () => {},
+            onSettled: () => {
               operationResultDuringCompletion = replyRunRegistryForTest.get("main")?.result;
             },
           },
@@ -3061,8 +3067,9 @@ describe("createFollowupRunner runtime config", () => {
       createQueuedRun({
         originatingChannel: "telegram",
         originatingTo: "chat-1",
-        queuedLifecycle: {
-          onComplete: () => {
+        turnAdoptionLifecycle: {
+          onAdopted: async () => {},
+          onSettled: () => {
             operationResultDuringCompletion = replyRunRegistryForTest.get("main")?.result;
           },
         },
@@ -6038,7 +6045,7 @@ describe("createFollowupRunner messaging delivery and dedupe", () => {
     const finalText =
       "Here is the answer the queued user asked for. It includes enough detail to be a visible response, and it has another sentence so the substantive-final detector treats it as a real reply.";
     const parentOnComplete = vi.fn();
-    const parentLifecycle = { onComplete: parentOnComplete };
+    const parentLifecycle = { onAdopted: async () => {}, onSettled: parentOnComplete };
     const queued = baseQueuedRun("discord");
     const { onBlockReply } = await runMessagingCase({
       agentResult: {
@@ -6049,7 +6056,7 @@ describe("createFollowupRunner messaging delivery and dedupe", () => {
         ...queued,
         originatingChannel: "discord",
         originatingTo: "channel:C1",
-        queuedLifecycle: parentLifecycle,
+        turnAdoptionLifecycle: parentLifecycle,
         run: {
           ...queued.run,
           sourceReplyDeliveryMode: "message_tool_only",
@@ -6072,7 +6079,7 @@ describe("createFollowupRunner messaging delivery and dedupe", () => {
     expect(retry?.prompt).toContain("message(action=send)");
     expect(retry?.prompt).toContain(finalText);
     // System retry detaches from the client turn lifecycle; parent completion owns onComplete once.
-    expect(retry?.queuedLifecycle).toBeUndefined();
+    expect(retry?.turnAdoptionLifecycle).toBeUndefined();
     expect(parentOnComplete).toHaveBeenCalledTimes(1);
   });
 
diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts
index c0f45f19ad65..a5717df43c4a 100644
--- a/src/auto-reply/reply/get-reply-run.ts
+++ b/src/auto-reply/reply/get-reply-run.ts
@@ -1414,14 +1414,17 @@ export async function runPreparedReply(
   );
   // Abort-signal attachment for queued followups:
   // - room_event: always inherit (source admission fence / ambient cancel).
-  // - Gateway-owned lifecycle (chat.send): always inherit so Esc can cancel a
-  //   turn after chat.send terminalizes while still queued.
+  // - Gateway-owned lifecycle (chat.send / turnAdoptionLifecycle): always inherit
+  //   so Esc can cancel a turn after chat.send terminalizes while still queued.
   // - plain user_request without lifecycle: deliberately detach from the
   //   source/active-lane signal so a superseded parent abort does not cancel a
   //   still-valid queued user turn.
+  const hasQueuedOwnershipLifecycle = Boolean(opts?.turnAdoptionLifecycle);
   const queuedFollowupAbortSignal =
-    opts?.queuedFollowupLifecycle || inboundEventKind === "room_event"
-      ? (opts?.queuedFollowupAbortSignal ?? opts?.abortSignal)
+    hasQueuedOwnershipLifecycle || inboundEventKind === "room_event"
+      ? (opts?.queuedFollowupAbortSignal ??
+        opts?.turnAdoptionLifecycle?.abortSignal ??
+        opts?.abortSignal)
       : undefined;
   const replyRoute = resolveEffectiveReplyRoute({
     ctx: {
@@ -1531,7 +1534,7 @@ export async function runPreparedReply(
     currentInboundContext,
     ...(queuedFollowupAbortSignal ? { abortSignal: queuedFollowupAbortSignal } : {}),
     deliveryCorrelations: opts?.queuedDeliveryCorrelations,
-    queuedLifecycle: opts?.queuedFollowupLifecycle,
+    turnAdoptionLifecycle: opts?.turnAdoptionLifecycle,
     onReplyAdmissionWaitChange: opts?.onReplyAdmissionWaitChange,
     messageId: sessionCtx.MessageSidFull ?? sessionCtx.MessageSid,
     summaryLine: baseBodyTrimmedRaw,
diff --git a/src/auto-reply/reply/queue.collect.test.ts b/src/auto-reply/reply/queue.collect.test.ts
index e7bee5df1c20..1202b5e0bb10 100644
--- a/src/auto-reply/reply/queue.collect.test.ts
+++ b/src/auto-reply/reply/queue.collect.test.ts
@@ -30,13 +30,52 @@ import { clearFollowupQueue, getExistingFollowupQueue } from "./queue/state.js";
 installQueueRuntimeErrorSilencer();
 
 describe("followup queue collect routing", () => {
+  it("marks exclusive admission without onAbandoned and isolates collect identity", () => {
+    // Failure window: cancel-only used to be inferred from missing onAbandoned,
+    // so exclusive admission without onAbandoned shared collect identity.
+    const exclusiveNoAbandon = createRun({ prompt: "exclusive a" });
+    exclusiveNoAbandon.turnAdoptionLifecycle = {
+      admission: "exclusive",
+      onAdopted: async () => {},
+    };
+    const exclusiveSibling = createRun({ prompt: "exclusive b" });
+    exclusiveSibling.turnAdoptionLifecycle = {
+      admission: "exclusive",
+      onAdopted: async () => {},
+    };
+    const cancelOnly = createRun({ prompt: "cancel-only" });
+    cancelOnly.turnAdoptionLifecycle = {
+      admission: "cancel-only",
+      ownerKey: "gw:owner",
+      onAdopted: async () => {},
+    };
+    const cancelOnlyShared = createRun({ prompt: "cancel-only shared" });
+    cancelOnlyShared.turnAdoptionLifecycle = {
+      admission: "cancel-only",
+      ownerKey: "gw:owner",
+      onAdopted: async () => {},
+    };
+
+    const exclusiveA = resolveFollowupDeliveryContextKey(exclusiveNoAbandon);
+    const exclusiveB = resolveFollowupDeliveryContextKey(exclusiveSibling);
+    expect(exclusiveA).not.toEqual(exclusiveB);
+
+    const cancelA = resolveFollowupDeliveryContextKey(cancelOnly);
+    const cancelB = resolveFollowupDeliveryContextKey(cancelOnlyShared);
+    expect(cancelA).toEqual(cancelB);
+  });
+
   it("retries lifecycle admission after a callback rejection", async () => {
     const onAdmitted = vi
       .fn<() => Promise<void>>()
       .mockRejectedValueOnce(new Error("admission failed"))
       .mockResolvedValueOnce();
     const run = createRun({ prompt: "retry admission" });
-    run.queuedLifecycle = { onAdmitted };
+    run.turnAdoptionLifecycle = {
+      onAdopted: onAdmitted,
+      admission: "exclusive",
+      onAbandoned: () => {},
+    };
 
     await expect(admitFollowupRunLifecycle(run)).rejects.toThrow("admission failed");
     await expect(admitFollowupRunLifecycle(run)).resolves.toBeUndefined();
@@ -61,7 +100,12 @@ describe("followup queue collect routing", () => {
       events.push("complete");
     });
     const run = createRun({ prompt: "complete during admission" });
-    run.queuedLifecycle = { onAdmitted, onComplete };
+    run.turnAdoptionLifecycle = {
+      onAdopted: onAdmitted,
+      onSettled: onComplete,
+      admission: "exclusive",
+      onAbandoned: () => {},
+    };
 
     const admission = admitFollowupRunLifecycle(run);
     await admissionStarted.promise;
@@ -84,7 +128,7 @@ describe("followup queue collect routing", () => {
     const key = `test-rejected-lifecycle-${Date.now()}`;
     const onEnqueued = vi.fn(() => false);
     const run = createRun({ prompt: "duplicate owner" });
-    run.queuedLifecycle = { onEnqueued };
+    run.turnAdoptionLifecycle = { onAdopted: async () => {}, onDeferred: onEnqueued };
 
     const enqueued = enqueueFollowupRun(key, run, {
       mode: "followup",
@@ -748,7 +792,7 @@ describe("followup queue collect routing", () => {
             originatingTo: "channel:A",
             originatingChatType: "channel",
           }),
-          queuedLifecycle: { onComplete },
+          turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
         },
         settings,
       );
@@ -783,7 +827,11 @@ describe("followup queue collect routing", () => {
         key,
         {
           ...createRun({ prompt: "rejected" }),
-          queuedLifecycle: { onEnqueued, onComplete },
+          turnAdoptionLifecycle: {
+            onAdopted: async () => {},
+            onDeferred: onEnqueued,
+            onSettled: onComplete,
+          },
         },
         settings,
       ),
@@ -1580,7 +1628,7 @@ describe("followup queue collect routing", () => {
           originatingTo: "same-target",
           originatingChatType: "direct",
         }),
-        queuedLifecycle: { onComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
       },
       settings,
     );
@@ -1907,7 +1955,7 @@ describe("followup queue collect routing", () => {
     };
     const controller = new AbortController();
     const begin = () => () => undefined;
-    const lifecycle = { onComplete: () => undefined };
+    const lifecycle = { onAdopted: async () => {}, onSettled: () => undefined };
 
     enqueueFollowupRun(
       key,
@@ -1927,7 +1975,7 @@ describe("followup queue collect routing", () => {
     first.currentInboundContext = { text: "room event body" };
     first.abortSignal = controller.signal;
     first.deliveryCorrelations = [{ begin }];
-    first.queuedLifecycle = lifecycle;
+    first.turnAdoptionLifecycle = lifecycle;
     enqueueFollowupRun(
       key,
       createRun({
@@ -1948,7 +1996,7 @@ describe("followup queue collect routing", () => {
     expect(calls[0]?.currentInboundContext?.text).toBe("room event body");
     expect(calls[0]?.abortSignal).toBe(controller.signal);
     expect(calls[0]?.deliveryCorrelations?.[0]?.begin).toBe(begin);
-    expect(calls[0]?.queuedLifecycle).toBe(lifecycle);
+    expect(calls[0]?.turnAdoptionLifecycle).toBe(lifecycle);
     expect(calls[1]?.prompt).toBe("second");
   });
 
@@ -2336,7 +2384,7 @@ describe("followup queue collect routing", () => {
             originatingChannel: "webchat",
             originatingTo: "session:main",
           }),
-          queuedLifecycle: { ownerKey },
+          turnAdoptionLifecycle: { onAdopted: async () => {}, ownerKey },
         },
         settings,
       );
@@ -3122,7 +3170,7 @@ describe("followup queue collect routing", () => {
       {
         ...createRun({ prompt: "aborted" }),
         abortSignal: controller.signal,
-        queuedLifecycle: { onComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
       },
       settings,
     );
@@ -3411,7 +3459,7 @@ describe("followup queue collect routing", () => {
         currentInboundContext: { text: "live context" },
         abortSignal: controller.signal,
         deliveryCorrelations: [{ begin }],
-        queuedLifecycle: { onComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
       },
       settings,
     );
@@ -3429,7 +3477,7 @@ describe("followup queue collect routing", () => {
     expect(calls[1]?.currentInboundAudio).toBe(true);
     expect(calls[1]?.currentInboundContext?.text).toBe("live context");
     expect(calls[1]?.abortSignal).toBe(controller.signal);
-    expect(calls[1]?.queuedLifecycle?.onComplete).toBe(onComplete);
+    expect(calls[1]?.turnAdoptionLifecycle?.onSettled).toBe(onComplete);
     expect(calls[1]?.deliveryCorrelations?.[0]?.begin).toBe(begin);
   });
 
@@ -3501,7 +3549,7 @@ describe("followup queue collect routing", () => {
         currentInboundEventKind: "room_event",
         currentInboundContext: { text: "dropped context" },
         abortSignal: controller.signal,
-        queuedLifecycle: { onComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
       },
       settings,
     );
@@ -3537,9 +3585,10 @@ describe("followup queue collect routing", () => {
         {
           ...createRun({ prompt }),
           abortSignal: new AbortController().signal,
-          queuedLifecycle: {
+          turnAdoptionLifecycle: {
+            onAdopted: async () => {},
             onCancellationRetired: sourceCancellationRetirements[index],
-            onComplete: sourceCompletions[index],
+            onSettled: sourceCompletions[index],
           },
         },
         settings,
@@ -3551,12 +3600,12 @@ describe("followup queue collect routing", () => {
       calls.push(run);
       if (calls.length === 1) {
         expect(run.prompt).toContain("[Queue overflow] Dropped 2 messages due to cap.");
-        await run.queuedLifecycle?.onAdmitted?.();
+        await run.turnAdoptionLifecycle?.onAdopted?.();
         expect(sourceCancellationRetirements[0]).toHaveBeenCalledTimes(1);
         expect(sourceCancellationRetirements[1]).not.toHaveBeenCalled();
         expect(sourceCompletions[0]).not.toHaveBeenCalled();
         expect(sourceCompletions[1]).not.toHaveBeenCalled();
-        run.queuedLifecycle?.onComplete?.();
+        run.turnAdoptionLifecycle?.onSettled?.();
         expect(sourceCompletions[0]).toHaveBeenCalledTimes(1);
         expect(sourceCompletions[1]).toHaveBeenCalledTimes(1);
         return;
@@ -3587,11 +3636,13 @@ describe("followup queue collect routing", () => {
       key,
       {
         ...createRun({ prompt: "dropped lifecycle source" }),
-        queuedLifecycle: {
-          onAdmitted: async () => {
+        turnAdoptionLifecycle: {
+          onAdopted: async () => {
             events.push("source-admitted");
           },
-          onComplete: sourceComplete,
+          onSettled: sourceComplete,
+          admission: "exclusive",
+          onAbandoned: () => {},
         },
       },
       settings,
@@ -3601,10 +3652,10 @@ describe("followup queue collect routing", () => {
     scheduleFollowupDrain(key, async (run) => {
       if (run.prompt.includes("[Queue overflow]")) {
         events.push("summary-started");
-        expect(run.queuedLifecycle?.onAdmitted).toEqual(expect.any(Function));
-        await run.queuedLifecycle?.onAdmitted?.();
+        expect(run.turnAdoptionLifecycle?.onAdopted).toEqual(expect.any(Function));
+        await run.turnAdoptionLifecycle?.onAdopted?.();
         events.push("model");
-        run.queuedLifecycle?.onComplete?.();
+        run.turnAdoptionLifecycle?.onSettled?.();
         return;
       }
       events.push("live-followup");
@@ -3632,7 +3683,7 @@ describe("followup queue collect routing", () => {
     let attempts = 0;
     const runFollowup = async (run: FollowupRun) => {
       calls.push(run);
-      expect(run.queuedLifecycle).toBeUndefined();
+      expect(run.turnAdoptionLifecycle).toBeUndefined();
       attempts += 1;
       if (attempts === 1) {
         firstAttempt.resolve();
@@ -3654,7 +3705,7 @@ describe("followup queue collect routing", () => {
         ...createRun({ prompt: "dropped ambient" }),
         currentInboundEventKind: "room_event",
         currentInboundContext: { text: "dropped context" },
-        queuedLifecycle: { onComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
       },
       settings,
     );
@@ -3671,7 +3722,7 @@ describe("followup queue collect routing", () => {
     expect(getExistingFollowupQueue(key)?.summarySources[0]?.currentInboundEventKind).toBe(
       "room_event",
     );
-    expect(getExistingFollowupQueue(key)?.summarySources[0]?.queuedLifecycle).toBeDefined();
+    expect(getExistingFollowupQueue(key)?.summarySources[0]?.turnAdoptionLifecycle).toBeDefined();
     expect(getExistingFollowupQueue(key)?.summarySources[0]?.currentInboundContext).toBeUndefined();
 
     scheduleFollowupDrain(key, runFollowup);
@@ -3703,7 +3754,7 @@ describe("followup queue collect routing", () => {
       {
         ...createRun({ prompt: "first" }),
         abortSignal: new AbortController().signal,
-        queuedLifecycle: { onComplete: firstComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: firstComplete },
       },
       settings,
     );
@@ -3712,7 +3763,7 @@ describe("followup queue collect routing", () => {
       {
         ...createRun({ prompt: "second" }),
         abortSignal: new AbortController().signal,
-        queuedLifecycle: { onComplete: secondComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: secondComplete },
       },
       settings,
     );
@@ -3736,14 +3787,16 @@ describe("followup queue collect routing", () => {
     const settings: QueueSettings = { mode: "collect", debounceMs: 0 };
 
     const first = createRun({ prompt: "first" });
-    first.queuedLifecycle = {
-      onAdmitted: async () => {
+    first.turnAdoptionLifecycle = {
+      onAdopted: async () => {
         events.push("first-admitted");
       },
+      admission: "exclusive",
+      onAbandoned: () => {},
     };
     const second = createRun({ prompt: "second" });
-    second.queuedLifecycle = {
-      onAdmitted: vi
+    second.turnAdoptionLifecycle = {
+      onAdopted: vi
         .fn<() => Promise<void>>()
         .mockImplementationOnce(async () => {
           events.push("second-rejected");
@@ -3752,6 +3805,8 @@ describe("followup queue collect routing", () => {
         .mockImplementationOnce(async () => {
           events.push("second-admitted");
         }),
+      admission: "exclusive",
+      onAbandoned: () => {},
     };
 
     enqueueFollowupRun(key, first, settings);
@@ -3785,7 +3840,7 @@ describe("followup queue collect routing", () => {
       "second-admitted",
       "model:second",
     ]);
-    expect(second.queuedLifecycle.onAdmitted).toHaveBeenCalledTimes(2);
+    expect(second.turnAdoptionLifecycle.onAdopted).toHaveBeenCalledTimes(2);
   });
 
   it("collects transcript-owned turns under one aggregate recorder", async () => {
@@ -3819,7 +3874,7 @@ describe("followup queue collect routing", () => {
           currentInboundContext: { text: "shared gateway context", promptJoiner: " " },
           deliveryCorrelations: [deliveryCorrelation],
           abortSignal: new AbortController().signal,
-          queuedLifecycle: { onComplete },
+          turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
         },
         settings,
       );
@@ -3920,9 +3975,10 @@ describe("followup queue collect routing", () => {
           transcriptPrompt: `${prompt} transcript`,
           userTurnTranscriptRecorder: createRecorder(`${prompt} transcript`),
           abortSignal,
-          queuedLifecycle: {
+          turnAdoptionLifecycle: {
+            onAdopted: async () => {},
             onCancellationRetired: sourceCancellationRetirements[index],
-            onComplete: sourceCompletions[index],
+            onSettled: sourceCompletions[index],
           },
         },
         settings,
@@ -3934,7 +3990,7 @@ describe("followup queue collect routing", () => {
       if (calls.length === 1) {
         expect(run.abortSignal).toBeDefined();
         expect(run.abortSignal).not.toBe(survivor.signal);
-        await run.queuedLifecycle?.onAdmitted?.();
+        await run.turnAdoptionLifecycle?.onAdopted?.();
         expect(sourceCancellationRetirements[0]).toHaveBeenCalledTimes(1);
         expect(sourceCancellationRetirements[1]).not.toHaveBeenCalled();
         expect(sourceCompletions[0]).not.toHaveBeenCalled();
@@ -3968,7 +4024,7 @@ describe("followup queue collect routing", () => {
     scheduleFollowupDrain(key, async (run) => {
       expect(run.abortSignal).toBeUndefined();
       expect(run.queueAbortSignal?.aborted).toBe(false);
-      await run.queuedLifecycle?.onAdmitted?.();
+      await run.turnAdoptionLifecycle?.onAdopted?.();
       clearFollowupQueue(key);
       expect(run.queueAbortSignal?.aborted).toBe(true);
       done.resolve();
@@ -3990,7 +4046,7 @@ describe("followup queue collect routing", () => {
       const enqueueSource = (prompt: string, onComplete: () => void, abortSignal?: AbortSignal) => {
         const source: FollowupRun = {
           ...createRun({ prompt }),
-          queuedLifecycle: { onComplete },
+          turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
         };
         if (abortSignal) {
           source.abortSignal = abortSignal;
@@ -4052,7 +4108,7 @@ describe("followup queue collect routing", () => {
       {
         ...createRun({ prompt: "owner A summary" }),
         abortSignal: new AbortController().signal,
-        queuedLifecycle: { onComplete: summarizedComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: summarizedComplete },
       },
       settings,
     );
@@ -4061,7 +4117,7 @@ describe("followup queue collect routing", () => {
       {
         ...createRun({ prompt: "owner B live" }),
         abortSignal: aborted.signal,
-        queuedLifecycle: { onComplete: abortedComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: abortedComplete },
       },
       settings,
     );
@@ -4105,7 +4161,7 @@ describe("followup queue collect routing", () => {
       {
         ...createRun({ prompt: "elided and cancelled" }),
         abortSignal: elided.signal,
-        queuedLifecycle: { onComplete: elidedComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: elidedComplete },
       },
       settings,
     );
@@ -4139,7 +4195,7 @@ describe("followup queue collect routing", () => {
       key,
       {
         ...createRun({ prompt: "elided source" }),
-        queuedLifecycle: { onComplete: elidedComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: elidedComplete },
       },
       settings,
     );
@@ -4147,7 +4203,7 @@ describe("followup queue collect routing", () => {
       key,
       {
         ...createRun({ prompt: "retained source" }),
-        queuedLifecycle: { onComplete: retainedComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: retainedComplete },
       },
       settings,
     );
@@ -4158,7 +4214,7 @@ describe("followup queue collect routing", () => {
       if (calls.length === 1) {
         expect(run.prompt).toContain("Dropped 2 messages");
         expect(run.prompt).toContain("retained source");
-        await run.queuedLifecycle?.onAdmitted?.();
+        await run.turnAdoptionLifecycle?.onAdopted?.();
         expect(getExistingFollowupQueue(key)?.summaryElisions).toEqual([]);
         expect(getExistingFollowupQueue(key)?.droppedCount).toBe(0);
         throw new Error("admitted summary failure");
@@ -4186,14 +4242,16 @@ describe("followup queue collect routing", () => {
     };
 
     const first = createRun({ prompt: "first dropped" });
-    first.queuedLifecycle = {
-      onAdmitted: async () => {
+    first.turnAdoptionLifecycle = {
+      onAdopted: async () => {
         events.push("first-admitted");
       },
+      admission: "exclusive",
+      onAbandoned: () => {},
     };
     const second = createRun({ prompt: "second dropped" });
-    second.queuedLifecycle = {
-      onAdmitted: vi
+    second.turnAdoptionLifecycle = {
+      onAdopted: vi
         .fn<() => Promise<void>>()
         .mockImplementationOnce(async () => {
           events.push("second-rejected");
@@ -4202,6 +4260,8 @@ describe("followup queue collect routing", () => {
         .mockImplementationOnce(async () => {
           events.push("second-admitted");
         }),
+      admission: "exclusive",
+      onAbandoned: () => {},
     };
 
     enqueueFollowupRun(key, first, settings);
@@ -4238,7 +4298,7 @@ describe("followup queue collect routing", () => {
       "summary-model",
       "live-followup",
     ]);
-    expect(second.queuedLifecycle.onAdmitted).toHaveBeenCalledTimes(2);
+    expect(second.turnAdoptionLifecycle.onAdopted).toHaveBeenCalledTimes(2);
   });
 });
 
diff --git a/src/auto-reply/reply/queue.in-flight.test.ts b/src/auto-reply/reply/queue.in-flight.test.ts
index d1e3fc1d677b..4829e80aa1af 100644
--- a/src/auto-reply/reply/queue.in-flight.test.ts
+++ b/src/auto-reply/reply/queue.in-flight.test.ts
@@ -44,11 +44,11 @@ describe("followup queue in-flight ownership", () => {
       const calls: FollowupRun[] = [];
       const active = {
         ...createRun({ prompt: "active" }),
-        queuedLifecycle: { onComplete: activeComplete },
+        turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: activeComplete },
       };
       const runFollowup = async (run: FollowupRun) => {
         calls.push(run);
-        await run.queuedLifecycle?.onAdmitted?.();
+        await run.turnAdoptionLifecycle?.onAdopted?.();
         if (run === active) {
           entered.resolve();
           await release.promise;
@@ -68,7 +68,7 @@ describe("followup queue in-flight ownership", () => {
             key,
             {
               ...createRun({ prompt: "pending" }),
-              queuedLifecycle: { onComplete: pendingComplete },
+              turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: pendingComplete },
             },
             createSettings(dropPolicy),
             "none",
@@ -111,7 +111,7 @@ describe("followup queue in-flight ownership", () => {
     const rejectedComplete = vi.fn();
     const active = createRun({ prompt: "active" });
     const runFollowup = async (run: FollowupRun) => {
-      await run.queuedLifecycle?.onAdmitted?.();
+      await run.turnAdoptionLifecycle?.onAdopted?.();
       if (run === active) {
         entered.resolve();
         await release.promise;
@@ -134,9 +134,10 @@ describe("followup queue in-flight ownership", () => {
           key,
           {
             ...createRun({ prompt: "rejected" }),
-            queuedLifecycle: {
-              onEnqueued: rejectedEnqueued,
-              onComplete: rejectedComplete,
+            turnAdoptionLifecycle: {
+              onAdopted: async () => {},
+              onDeferred: rejectedEnqueued,
+              onSettled: rejectedComplete,
             },
           },
           createSettings("new"),
@@ -179,7 +180,7 @@ describe("followup queue in-flight ownership", () => {
         originatingTo: "channel:A",
         originatingChatType: "channel",
       }),
-      queuedLifecycle: { onComplete },
+      turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: onComplete },
     }));
     const runFollowup = async (run: FollowupRun) => {
       if (!aggregate) {
@@ -207,7 +208,7 @@ describe("followup queue in-flight ownership", () => {
           key,
           {
             ...createRun({ prompt: "pending-old" }),
-            queuedLifecycle: { onComplete: pendingComplete },
+            turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: pendingComplete },
           },
           oldSettings,
           "none",
@@ -221,7 +222,7 @@ describe("followup queue in-flight ownership", () => {
       expect(pendingComplete).toHaveBeenCalledOnce();
       expect(groupCompletions.map((complete) => complete.mock.calls.length)).toEqual([0, 0]);
 
-      await aggregate?.queuedLifecycle?.onAdmitted?.();
+      await aggregate?.turnAdoptionLifecycle?.onAdopted?.();
       expect(queue?.items.map((item) => item.prompt)).toEqual(["survivor"]);
       expect(queue?.inFlight.size).toBe(2);
       expect(getFollowupQueueDepth(key)).toBe(1);
@@ -231,7 +232,7 @@ describe("followup queue in-flight ownership", () => {
           key,
           {
             ...createRun({ prompt: "rejected-new" }),
-            queuedLifecycle: { onComplete: rejectedComplete },
+            turnAdoptionLifecycle: { onAdopted: async () => {}, onSettled: rejectedComplete },
           },
           { ...initialSettings, cap: 1, dropPolicy: "new" },
           "none",
diff --git a/src/auto-reply/reply/queue/drain.ts b/src/auto-reply/reply/queue/drain.ts
index 5fac023a6f09..08e1b2b1b002 100644
--- a/src/auto-reply/reply/queue/drain.ts
+++ b/src/auto-reply/reply/queue/drain.ts
@@ -49,16 +49,28 @@ const FOLLOWUP_RUN_CALLBACKS = resolveGlobalMap<string, (run: FollowupRun) => Pr
 
 const QUEUED_ADMISSION_OWNER_STATE_KEY = Symbol.for("openclaw.queuedAdmissionOwnerState");
 const queuedAdmissionOwnerState = resolveGlobalSingleton(QUEUED_ADMISSION_OWNER_STATE_KEY, () => ({
-  keys: new WeakMap<NonNullable<FollowupRun["queuedLifecycle"]>, string>(),
+  keys: new WeakMap<NonNullable<FollowupRun["turnAdoptionLifecycle"]>, string>(),
   nextId: 1,
 }));
 
-function resolveQueuedLifecycleDeliveryKey(lifecycle: FollowupRun["queuedLifecycle"]): string {
+function hasExclusiveTurnAdmission(
+  lifecycle: FollowupRun["turnAdoptionLifecycle"],
+): lifecycle is NonNullable<FollowupRun["turnAdoptionLifecycle"]> & {
+  admission: "exclusive";
+} {
+  return lifecycle?.admission === "exclusive";
+}
+
+function resolveTurnAdoptionLifecycleDeliveryKey(
+  lifecycle: FollowupRun["turnAdoptionLifecycle"],
+): string {
   if (!lifecycle) {
     return "";
   }
   const explicitOwnerKey = lifecycle.ownerKey ?? "";
-  if (!lifecycle.onAdmitted) {
+  // Closed admission marker — never infer exclusive from onAbandoned presence.
+  // Cancel-only owners share collect identity via ownerKey alone.
+  if (!hasExclusiveTurnAdmission(lifecycle)) {
     return explicitOwnerKey;
   }
   let admissionOwnerKey = queuedAdmissionOwnerState.keys.get(lifecycle);
@@ -73,7 +85,9 @@ function resolveQueuedLifecycleDeliveryKey(lifecycle: FollowupRun["queuedLifecyc
 
 function assertSingleAdmissionOwner(items: readonly FollowupRun[]): void {
   const owners = new Set(
-    items.flatMap((item) => (item.queuedLifecycle?.onAdmitted ? [item.queuedLifecycle] : [])),
+    items.flatMap((item) =>
+      hasExclusiveTurnAdmission(item.turnAdoptionLifecycle) ? [item.turnAdoptionLifecycle] : [],
+    ),
   );
   if (owners.size > 1) {
     throw new Error("followup queue cannot aggregate distinct admission lifecycles");
@@ -180,7 +194,7 @@ export function resolveFollowupDeliveryContextKey(run: FollowupRun): string {
     run.originatingReplyToMode ?? "",
     normalizeChatType(run.originatingChatType) ?? "",
     resolveFollowupAuthorizationKey(execution),
-    run.queuedLifecycle?.ownerKey ?? "",
+    run.turnAdoptionLifecycle?.ownerKey ?? "",
     normalizeOptionalString(execution.runtimePolicySessionKey ?? execution.sessionKey) ?? "",
     execution.messageProvider ?? "",
     JSON.stringify([...new Set(execution.clientCaps ?? [])].toSorted()),
@@ -209,7 +223,7 @@ export function resolveFollowupDeliveryContextKey(run: FollowupRun): string {
     execution.suppressNextUserMessagePersistence === true,
     execution.suppressTranscriptOnlyAssistantPersistence === true,
     execution.blockReplyBreak,
-    resolveQueuedLifecycleDeliveryKey(run.queuedLifecycle),
+    resolveTurnAdoptionLifecycleDeliveryKey(run.turnAdoptionLifecycle),
   ]);
 }
 
@@ -297,7 +311,7 @@ type FollowupRuntimeMetadata = Pick<
   | "abortSignal"
   | "queueAbortSignal"
   | "deliveryCorrelations"
-  | "queuedLifecycle"
+  | "turnAdoptionLifecycle"
   | "onReplyAdmissionWaitChange"
 >;
 
@@ -409,7 +423,7 @@ function resolveAggregateOwner(items: readonly FollowupRun[]): FollowupRun | und
   // later transport-only source has no cancellation identity.
   return (
     items.findLast((item) => item.abortSignal) ??
-    items.findLast((item) => item.queuedLifecycle) ??
+    items.findLast((item) => item.turnAdoptionLifecycle) ??
     items.at(-1)
   );
 }
@@ -542,7 +556,7 @@ function collectRuntimeMetadata(
     abortSignal,
     queueAbortSignal: items.find((item) => item.queueAbortSignal)?.queueAbortSignal,
     deliveryCorrelations: deliveryCorrelations.length > 0 ? deliveryCorrelations : undefined,
-    queuedLifecycle: items.length === 1 ? items[0]?.queuedLifecycle : undefined,
+    turnAdoptionLifecycle: items.length === 1 ? items[0]?.turnAdoptionLifecycle : undefined,
     onReplyAdmissionWaitChange:
       admissionWaitCallbacks.size > 0
         ? (waiting) => {
@@ -655,7 +669,7 @@ function releaseQueueSummaryDeliveryForRetry(
     if (sourceIndex >= 0) {
       queue.summarySources[sourceIndex] = createOverflowSummaryRetrySource(source);
     }
-    if (!source.queuedLifecycle) {
+    if (!source.turnAdoptionLifecycle) {
       completeFollowupRunLifecycle(source);
     }
   }
@@ -698,7 +712,7 @@ async function runQueueSummaryDelivery(
   const cancellation = createAggregateCancellation(protectedSources);
   const needsAdmission =
     protectedSources.length > 1 ||
-    protectedSources.some((source) => source.queuedLifecycle?.onAdmitted);
+    protectedSources.some((source) => hasExclusiveTurnAdmission(source.turnAdoptionLifecycle));
   const onAdmitted = needsAdmission
     ? async () => {
         if (admitted) {
@@ -884,7 +898,7 @@ export function createOverflowSummaryRetrySource(source: FollowupRun): FollowupR
     originatingReplyToMode: source.originatingReplyToMode,
     originatingChatType: source.originatingChatType,
     abortSignal: source.abortSignal,
-    queuedLifecycle: source.queuedLifecycle,
+    turnAdoptionLifecycle: source.turnAdoptionLifecycle,
     onReplyAdmissionWaitChange: source.onReplyAdmissionWaitChange,
     ...(source.currentInboundEventKind === "room_event"
       ? { currentInboundEventKind: "room_event" }
@@ -948,12 +962,14 @@ async function runSyntheticOverflowSummary(params: {
     onReplyAdmissionWaitChange: collectRuntimeMetadata(params.sources).onReplyAdmissionWaitChange,
     ...(params.onAdmitted
       ? {
-          queuedLifecycle: {
-            onAdmitted: async () => {
+          turnAdoptionLifecycle: {
+            // Synthetic aggregate owner — not a durable exclusive ingress identity.
+            admission: "cancel-only" as const,
+            onAdopted: async () => {
               await params.onAdmitted?.();
               admitted = true;
             },
-            onComplete: () => {
+            onSettled: () => {
               if (admitted) {
                 for (const source of params.sources) {
                   completeFollowupRunLifecycle(source);
@@ -1229,7 +1245,9 @@ export function scheduleFollowupDrain(
             };
             const needsGroupAdmission =
               activeGroupItems.length > 1 ||
-              activeGroupItems.some((item) => item.queuedLifecycle?.onAdmitted);
+              activeGroupItems.some((item) =>
+                hasExclusiveTurnAdmission(item.turnAdoptionLifecycle),
+              );
             const consumeAdmittedGroup = () => {
               cancellation.admit();
               admitted = true;
@@ -1264,9 +1282,11 @@ export function scheduleFollowupDrain(
                 ...collectRuntimeMetadata(activeGroupItems, cancellation.signal),
                 ...(needsGroupAdmission
                   ? {
-                      queuedLifecycle: {
-                        onAdmitted: admitGroupSources,
-                        onComplete: () => {
+                      turnAdoptionLifecycle: {
+                        // Synthetic aggregate owner — sources keep their own admission.
+                        admission: "cancel-only" as const,
+                        onAdopted: admitGroupSources,
+                        onSettled: () => {
                           if (admitted) {
                             completeGroup();
                           }
diff --git a/src/auto-reply/reply/queue/types.ts b/src/auto-reply/reply/queue/types.ts
index c52055cea01a..c68cfe6f5298 100644
--- a/src/auto-reply/reply/queue/types.ts
+++ b/src/auto-reply/reply/queue/types.ts
@@ -17,9 +17,9 @@ import type { UserTurnTranscriptRecorder } from "../../../sessions/user-turn-tra
 import type { SkillSnapshot } from "../../../skills/types.js";
 import type {
   QueuedReplyDeliveryCorrelation,
-  QueuedReplyLifecycle,
   SourceReplyDeliveryMode,
   TaskSuggestionDeliveryMode,
+  TurnAdoptionLifecycle,
 } from "../../get-reply-options.types.js";
 import type { OriginatingChannelType } from "../../templating.js";
 import type { ElevatedLevel, ReasoningLevel, ThinkLevel, VerboseLevel } from "../directives.js";
@@ -35,6 +35,15 @@ export type QueueSettings = {
   dropPolicy?: QueueDropPolicy;
 };
 
+export type ResolveQueueSettingsParams = {
+  cfg: OpenClawConfig;
+  channel?: string;
+  sessionEntry?: SessionEntry;
+  inlineMode?: QueueMode;
+  inlineOptions?: Partial<QueueSettings>;
+  pluginDebounceMs?: number;
+};
+
 export type QueueDedupeMode = "message-id" | "prompt" | "none";
 
 type QueueInsertPosition = "tail" | "front";
@@ -72,7 +81,8 @@ export type FollowupRun = {
   /** Queue-owned cancellation fence used when lifecycle cleanup invalidates pending work. */
   queueAbortSignal?: AbortSignal;
   deliveryCorrelations?: QueuedReplyDeliveryCorrelation[];
-  queuedLifecycle?: QueuedReplyLifecycle;
+  /** Canonical ownership lifecycle for durable ingress / reply-lane transfer. */
+  turnAdoptionLifecycle?: TurnAdoptionLifecycle;
   /** Dispatch-scoped freshness owner for a queued delivery-barrier wait. */
   onReplyAdmissionWaitChange?: (waiting: boolean) => void;
   /** Provider message ID, when available (for deduplication). */
@@ -198,76 +208,87 @@ export function resolveFollowupAbortSignal(
   return signals.length > 1 ? AbortSignal.any(signals) : signals[0];
 }
 
-const enqueuedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
-const admittedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
-const admittingFollowupLifecycles = new WeakMap<QueuedReplyLifecycle, Promise<void>>();
-const retiredFollowupCancellationLifecycles = new WeakSet<QueuedReplyLifecycle>();
-const completedFollowupLifecycles = new WeakSet<QueuedReplyLifecycle>();
-const completedFollowupLifecycleCallbacks = new WeakSet<QueuedReplyLifecycle>();
+const enqueuedTurnAdoptionLifecycles = new WeakSet<TurnAdoptionLifecycle>();
+const admittedTurnAdoptionLifecycles = new WeakSet<TurnAdoptionLifecycle>();
+const admittingTurnAdoptionLifecycles = new WeakMap<TurnAdoptionLifecycle, Promise<void>>();
+const retiredTurnAdoptionCancellationLifecycles = new WeakSet<TurnAdoptionLifecycle>();
+const completedTurnAdoptionLifecycles = new WeakSet<TurnAdoptionLifecycle>();
+const completedTurnAdoptionLifecycleCallbacks = new WeakSet<TurnAdoptionLifecycle>();
 
-export function markFollowupRunEnqueued(run: Pick<FollowupRun, "queuedLifecycle">): boolean {
-  const lifecycle = run.queuedLifecycle;
-  if (!lifecycle || enqueuedFollowupLifecycles.has(lifecycle)) {
-    return true;
+type FollowupLifecycleRun = Pick<FollowupRun, "turnAdoptionLifecycle">;
+
+export function markFollowupRunEnqueued(run: FollowupLifecycleRun): boolean {
+  const lifecycle = run.turnAdoptionLifecycle;
+  if (lifecycle && !enqueuedTurnAdoptionLifecycles.has(lifecycle)) {
+    if (lifecycle.onDeferred?.() === false) {
+      return false;
+    }
+    enqueuedTurnAdoptionLifecycles.add(lifecycle);
   }
-  if (lifecycle.onEnqueued?.() === false) {
-    return false;
-  }
-  enqueuedFollowupLifecycles.add(lifecycle);
   return true;
 }
 
-export function retireFollowupRunCancellation(run: Pick<FollowupRun, "queuedLifecycle">): void {
-  const lifecycle = run.queuedLifecycle;
-  if (!lifecycle || retiredFollowupCancellationLifecycles.has(lifecycle)) {
+export function retireFollowupRunCancellation(run: FollowupLifecycleRun): void {
+  const lifecycle = run.turnAdoptionLifecycle;
+  if (!lifecycle || retiredTurnAdoptionCancellationLifecycles.has(lifecycle)) {
     return;
   }
-  retiredFollowupCancellationLifecycles.add(lifecycle);
+  retiredTurnAdoptionCancellationLifecycles.add(lifecycle);
   lifecycle.onCancellationRetired?.();
 }
 
-export async function admitFollowupRunLifecycle(
-  run: Pick<FollowupRun, "queuedLifecycle">,
-): Promise<void> {
-  const lifecycle = run.queuedLifecycle;
-  if (!lifecycle || admittedFollowupLifecycles.has(lifecycle)) {
+export async function admitFollowupRunLifecycle(run: FollowupLifecycleRun): Promise<void> {
+  const lifecycle = run.turnAdoptionLifecycle;
+  if (!lifecycle || admittedTurnAdoptionLifecycles.has(lifecycle)) {
     return;
   }
-  const existing = admittingFollowupLifecycles.get(lifecycle);
+  const existing = admittingTurnAdoptionLifecycles.get(lifecycle);
   if (existing) {
     await existing;
     return;
   }
-  if (completedFollowupLifecycles.has(lifecycle)) {
+  if (completedTurnAdoptionLifecycles.has(lifecycle)) {
     throw new Error("followup run lifecycle completed before admission");
   }
-  const admission = Promise.resolve()
-    .then(async () => await lifecycle.onAdmitted?.())
-    .then(() => {
-      admittedFollowupLifecycles.add(lifecycle);
-    });
-  admittingFollowupLifecycles.set(lifecycle, admission);
+
+  const admission = Promise.resolve().then(async () => {
+    if (!admittedTurnAdoptionLifecycles.has(lifecycle)) {
+      await lifecycle.onAdopted();
+      admittedTurnAdoptionLifecycles.add(lifecycle);
+    }
+  });
+
+  admittingTurnAdoptionLifecycles.set(lifecycle, admission);
   try {
     await admission;
   } finally {
-    admittingFollowupLifecycles.delete(lifecycle);
+    admittingTurnAdoptionLifecycles.delete(lifecycle);
   }
 }
 
-export function completeFollowupRunLifecycle(run: Pick<FollowupRun, "queuedLifecycle">): void {
-  const lifecycle = run.queuedLifecycle;
-  if (!lifecycle || completedFollowupLifecycles.has(lifecycle)) {
-    return;
-  }
-  completedFollowupLifecycles.add(lifecycle);
+export function completeFollowupRunLifecycle(run: FollowupLifecycleRun): void {
+  const lifecycle = run.turnAdoptionLifecycle;
+
   const finish = () => {
-    if (completedFollowupLifecycleCallbacks.has(lifecycle)) {
+    if (!lifecycle || completedTurnAdoptionLifecycleCallbacks.has(lifecycle)) {
       return;
     }
-    completedFollowupLifecycleCallbacks.add(lifecycle);
-    lifecycle.onComplete?.();
+    completedTurnAdoptionLifecycleCallbacks.add(lifecycle);
+    // onSettled must run even when onAbandoned throws (gateway/plugin cleanup).
+    try {
+      if (!admittedTurnAdoptionLifecycles.has(lifecycle)) {
+        lifecycle.onAbandoned?.();
+      }
+    } finally {
+      lifecycle.onSettled?.();
+    }
   };
-  const admission = admittingFollowupLifecycles.get(lifecycle);
+
+  if (lifecycle && !completedTurnAdoptionLifecycles.has(lifecycle)) {
+    completedTurnAdoptionLifecycles.add(lifecycle);
+  }
+
+  const admission = lifecycle ? admittingTurnAdoptionLifecycles.get(lifecycle) : undefined;
   if (!admission) {
     finish();
     return;
@@ -276,12 +297,3 @@ export function completeFollowupRunLifecycle(run: Pick<FollowupRun, "queuedLifec
   // the in-flight admission attempt so adoption and abandonment cannot race.
   void admission.then(finish, finish).catch(() => {});
 }
-
-export type ResolveQueueSettingsParams = {
-  cfg: OpenClawConfig;
-  channel?: string;
-  sessionEntry?: SessionEntry;
-  inlineMode?: QueueMode;
-  inlineOptions?: Partial<QueueSettings>;
-  pluginDebounceMs?: number;
-};
diff --git a/src/auto-reply/reply/stranded-reply-recovery.test.ts b/src/auto-reply/reply/stranded-reply-recovery.test.ts
index 593ac5255e11..92383e17addf 100644
--- a/src/auto-reply/reply/stranded-reply-recovery.test.ts
+++ b/src/auto-reply/reply/stranded-reply-recovery.test.ts
@@ -6,13 +6,17 @@ import { createMockFollowupRun } from "./test-helpers.js";
 const STRANDED_REPLY_RETRY_MARKER = "stranded-reply-retry";
 
 describe("buildStrandedReplyRetryFollowupRun lifecycle ownership", () => {
-  it("does not share the client turn's queuedLifecycle with the system retry", () => {
+  it("does not share the client turn's turnAdoptionLifecycle with the system retry", () => {
     const onComplete = vi.fn();
     const onEnqueued = vi.fn(() => true);
     const parent = createMockFollowupRun({
       prompt: "user question",
       transcriptPrompt: "user question",
-      queuedLifecycle: { onComplete, onEnqueued },
+      turnAdoptionLifecycle: {
+        onAdopted: async () => {},
+        onSettled: onComplete,
+        onDeferred: onEnqueued,
+      },
       admissionSessionId: "sess-rotated",
       onReplyAdmissionWaitChange: vi.fn(),
     });
@@ -22,7 +26,7 @@ describe("buildStrandedReplyRetryFollowupRun lifecycle ownership", () => {
       sourceReplyDeliveryMode: "message_tool_only",
     });
 
-    expect(retry.queuedLifecycle).toBeUndefined();
+    expect(retry.turnAdoptionLifecycle).toBeUndefined();
     expect(retry.strandedReplyRetry).toBe(true);
     expect(retry.summaryLine).toBe(STRANDED_REPLY_RETRY_MARKER);
     // Session routing stays; only the client-turn lifecycle identity is detached.
diff --git a/src/auto-reply/reply/stranded-reply-recovery.ts b/src/auto-reply/reply/stranded-reply-recovery.ts
index 5c637ed465e6..e424b01c722f 100644
--- a/src/auto-reply/reply/stranded-reply-recovery.ts
+++ b/src/auto-reply/reply/stranded-reply-recovery.ts
@@ -43,9 +43,10 @@ export function buildStrandedReplyRetryFollowupRun(
     userTurnTranscriptRecorder: undefined,
     currentInboundContext: undefined,
     // Internally generated system turn: the client turn's lifecycle (gateway cancel
-    // identity) completes with the parent run. queuedLifecycle is one-shot WeakSet-tracked,
-    // so a shared object would be double-owned and free cancel while the retry still runs.
-    queuedLifecycle: undefined,
+    // identity) completes with the parent run. turnAdoptionLifecycle is one-shot
+    // WeakSet-tracked, so a shared object would be double-owned and free cancel
+    // while the retry still runs.
+    turnAdoptionLifecycle: undefined,
     run: {
       ...base.run,
       sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
diff --git a/src/channels/message/index.ts b/src/channels/message/index.ts
index 2b0a232ac8df..43303fefd0cb 100644
--- a/src/channels/message/index.ts
+++ b/src/channels/message/index.ts
@@ -4,6 +4,16 @@ export { deriveDurableFinalDeliveryRequirements } from "./capabilities.js";
 export { defineChannelMessageAdapter } from "./adapter.js";
 export { createChannelMessageAdapterFromOutbound } from "./outbound-bridge.js";
 export { createDurableInboundReceiveJournalFromQueue } from "./durable-receive.js";
+export { INGRESS_CLAIM_PROCESS_ID, processPidFromOwnerId } from "./ingress-claim-owner.js";
+export {
+  bindIngressLifecycleToReplyOptions,
+  createChannelIngressDrain,
+  DEFAULT_INGRESS_ADOPTION_STALL_MS,
+} from "./ingress-drain.js";
+export {
+  DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+} from "./ingress-retry-policy.js";
 
 export {
   verifyChannelMessageAdapterCapabilityProofs,
@@ -30,6 +40,11 @@ export {
   createTypingCallbacks,
   resolveChannelSourceReplyDeliveryMode,
 } from "./reply-pipeline.js";
+export type {
+  ChannelIngressDispatchLifecycle,
+  ChannelIngressDrain,
+  ChannelIngressDrainDispatchResult,
+} from "./ingress-drain.js";
 export type {
   ChannelIngressQueue,
   ChannelIngressQueueClaim,
@@ -38,6 +53,7 @@ export type {
   ChannelIngressQueueRecord,
 } from "./ingress-queue.js";
 export type { MessageAckPolicy, MessageReceiveContext } from "./receive.js";
+export type { IngressNonRetryableFailure } from "./ingress-retry-policy.js";
 export type {
   ChannelMessageAdapterShape,
   ChannelMessageDurableFinalAdapter,
diff --git a/src/channels/turn/kernel.test.ts b/src/channels/turn/kernel.test.ts
index f865785bb9b3..e046eb37732e 100644
--- a/src/channels/turn/kernel.test.ts
+++ b/src/channels/turn/kernel.test.ts
@@ -1083,9 +1083,9 @@ describe("channel turn kernel", () => {
     expect(events).toEqual(["record", "afterRecord", "dispatch"]);
   });
 
-  it("threads onTurnAdopted into assembled reply options and fires after recovery persist attempt", async () => {
+  it("threads turnAdoptionLifecycle into assembled reply options and fires after recovery persist attempt", async () => {
     const events: string[] = [];
-    const onTurnAdopted = vi.fn(async () => {
+    const onAdopted = vi.fn(async () => {
       events.push("adopted");
     });
     const dispatchReplyWithBufferedBlockDispatcher = vi.fn(
@@ -1093,7 +1093,7 @@ describe("channel turn kernel", () => {
         events.push("dispatch-start");
         // Persist attempt completes before adoption (agent-runner contract).
         events.push("recovery-persist");
-        await params.replyOptions?.onTurnAdopted?.();
+        await params.replyOptions?.turnAdoptionLifecycle?.onAdopted();
         events.push("settle");
         return {
           queuedFinal: true,
@@ -1114,14 +1114,16 @@ describe("channel turn kernel", () => {
       delivery: {
         deliver: vi.fn(async () => undefined),
       },
-      onTurnAdopted,
+      turnAdoptionLifecycle: { onAdopted },
     });
 
-    expect(onTurnAdopted).toHaveBeenCalledOnce();
+    expect(onAdopted).toHaveBeenCalledOnce();
     expect(events).toEqual(["record", "dispatch-start", "recovery-persist", "adopted", "settle"]);
     expect(dispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledWith(
       expect.objectContaining({
-        replyOptions: expect.objectContaining({ onTurnAdopted }),
+        replyOptions: expect.objectContaining({
+          turnAdoptionLifecycle: expect.objectContaining({ onAdopted }),
+        }),
       }),
     );
   });
diff --git a/src/channels/turn/kernel.ts b/src/channels/turn/kernel.ts
index f2f9689d2d7a..6b97ab9e09e9 100644
--- a/src/channels/turn/kernel.ts
+++ b/src/channels/turn/kernel.ts
@@ -190,11 +190,14 @@ export const recordDroppedChannelInboundHistory = recordDroppedChannelTurnHistor
 function resolveAssembledReplyPipeline(
   params: AssembledChannelTurn,
 ): Pick<AssembledChannelTurn, "dispatcherOptions" | "replyOptions"> {
-  const onTurnAdopted = params.onTurnAdopted ?? params.replyOptions?.onTurnAdopted;
+  const turnAdoptionLifecycle =
+    params.turnAdoptionLifecycle ?? params.replyOptions?.turnAdoptionLifecycle;
   if (!params.replyPipeline) {
     return {
       dispatcherOptions: params.dispatcherOptions,
-      replyOptions: onTurnAdopted ? { ...params.replyOptions, onTurnAdopted } : params.replyOptions,
+      replyOptions: turnAdoptionLifecycle
+        ? { ...params.replyOptions, turnAdoptionLifecycle }
+        : params.replyOptions,
     };
   }
   const { onModelSelected, ...replyPipeline } = createChannelReplyPipeline({
@@ -212,7 +215,7 @@ function resolveAssembledReplyPipeline(
     replyOptions: {
       onModelSelected,
       ...params.replyOptions,
-      ...(onTurnAdopted ? { onTurnAdopted } : {}),
+      ...(turnAdoptionLifecycle ? { turnAdoptionLifecycle } : {}),
     },
   };
 }
@@ -709,7 +712,7 @@ async function runChannelTurn<
   const admission = resolved.admission ?? preflightAdmission ?? ({ kind: "dispatch" } as const);
   let result: ChannelTurnResult<TDispatchResult>;
   try {
-    // Prepared runDispatch was assembled earlier and ignores late options (including onTurnAdopted).
+    // Prepared runDispatch was assembled earlier and ignores late options.
     const dispatchResult = await dispatchResolvedChannelTurn(
       "runDispatch" in resolved
         ? {
@@ -729,7 +732,9 @@ async function runChannelTurn<
             admission,
             log: params.log,
             messageId: input.id,
-            ...(params.onTurnAdopted ? { onTurnAdopted: params.onTurnAdopted } : {}),
+            ...(params.turnAdoptionLifecycle
+              ? { turnAdoptionLifecycle: params.turnAdoptionLifecycle }
+              : {}),
           },
     );
     result = dispatchResult.dispatched ? { ...dispatchResult, admission } : dispatchResult;
diff --git a/src/channels/turn/types.ts b/src/channels/turn/types.ts
index 9d1935dd996f..05f2e65fe07d 100644
--- a/src/channels/turn/types.ts
+++ b/src/channels/turn/types.ts
@@ -1,6 +1,9 @@
 // Type contracts for channel turn normalization, admission, dispatch, and delivery.
 import type { CommandTurnKind } from "../../auto-reply/command-turn-context.js";
-import type { GetReplyOptions } from "../../auto-reply/get-reply-options.types.js";
+import type {
+  GetReplyOptions,
+  TurnAdoptionLifecycle,
+} from "../../auto-reply/get-reply-options.types.js";
 import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
 import type { DispatchFromConfigResult } from "../../auto-reply/reply/dispatch-from-config.types.js";
 import type { GetReplyFromConfig } from "../../auto-reply/reply/get-reply.types.js";
@@ -275,11 +278,8 @@ export type AssembledChannelTurn = {
   botLoopProtection?: ChannelBotLoopProtectionFacts;
   log?: (event: ChannelTurnLogEvent) => void;
   messageId?: string;
-  /**
-   * Observes turn adoption without waiting for settle. Threaded into
-   * replyOptions for the agent runner (after recovery persist attempt).
-   */
-  onTurnAdopted?: () => void | Promise<void>;
+  /** Canonical adoption lifecycle threaded into replyOptions. */
+  turnAdoptionLifecycle?: TurnAdoptionLifecycle;
 };
 
 /** Channel turn with dispatch runner already prepared. */
@@ -383,10 +383,6 @@ export type RunChannelTurnParams<TRaw, TDispatchResult = DispatchFromConfigResul
   raw: TRaw;
   adapter: ChannelTurnAdapter<TRaw, TDispatchResult>;
   log?: (event: ChannelTurnLogEvent) => void;
-  /**
-   * Observes turn adoption without waiting for settle. Fired after the
-   * recovery-context persist attempt (context may be absent when source
-   * delivery is suppressed). Default callers still await full settle.
-   */
-  onTurnAdopted?: () => void | Promise<void>;
+  /** Canonical adoption lifecycle for this turn. */
+  turnAdoptionLifecycle?: TurnAdoptionLifecycle;
 };
diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts
index 9b7a289c9f10..004b7b12807a 100644
--- a/src/gateway/server-methods/chat.ts
+++ b/src/gateway/server-methods/chat.ts
@@ -1303,9 +1303,12 @@ export const chatHandlers: GatewayRequestHandlers = {
                   abortSignal: activeRunAbort.controller.signal,
                   // Keep a Gateway-owned cancel identity after this chat.send
                   // terminalizes while the prompt waits in followup/collect queue.
-                  queuedFollowupLifecycle: {
+                  turnAdoptionLifecycle: {
+                    // Gateway cancel identity only — share collect key via ownerKey.
+                    admission: "cancel-only",
                     ownerKey: queuedFollowupOwnerKey,
-                    onEnqueued: () => {
+                    onAdopted: async () => {},
+                    onDeferred: () => {
                       queuedFollowupEnqueued = registerQueuedChatTurn({
                         chatQueuedTurns: ensureChatQueuedTurns(context),
                         runId: clientRunId,
@@ -1325,7 +1328,7 @@ export const chatHandlers: GatewayRequestHandlers = {
                         activeRunAbort.controller,
                       );
                     },
-                    onComplete: () => {
+                    onSettled: () => {
                       completeQueuedChatTurn(
                         ensureChatQueuedTurns(context),
                         clientRunId,
diff --git a/src/gateway/server-restart-sentinel.test.ts b/src/gateway/server-restart-sentinel.test.ts
index 4f6dcae04e72..9c8ede22a01d 100644
--- a/src/gateway/server-restart-sentinel.test.ts
+++ b/src/gateway/server-restart-sentinel.test.ts
@@ -684,7 +684,7 @@ describe("scheduleRestartSentinelWake", () => {
       },
     } as Awaited<ReturnType<typeof mocks.readRestartSentinel>>);
     mocks.recordInboundSessionAndDispatchReply.mockImplementationOnce(async (params) => {
-      await params.onTurnAdopted?.();
+      await params.turnAdoptionLifecycle?.onAdopted();
       await params.deliver({
         text: "done",
         replyToId: "restart-sentinel:agent:main:main:agentTurn:123",
@@ -809,7 +809,7 @@ describe("scheduleRestartSentinelWake", () => {
 
   it("fences an adopted generic turn in its explicit queue state directory", async () => {
     mocks.recordInboundSessionAndDispatchReply.mockImplementationOnce(async (params) => {
-      await params.onTurnAdopted?.();
+      await params.turnAdoptionLifecycle?.onAdopted();
     });
 
     await deliverQueuedSessionDelivery({
diff --git a/src/gateway/server-restart-sentinel.ts b/src/gateway/server-restart-sentinel.ts
index e8b32648a2fc..6c3e424bd63f 100644
--- a/src/gateway/server-restart-sentinel.ts
+++ b/src/gateway/server-restart-sentinel.ts
@@ -361,11 +361,14 @@ export async function deliverQueuedSessionDelivery(params: {
     },
     // Preflight remains retryable. Ownership starts only after the agent runner
     // has durably adopted the turn and before it can execute tools or reply.
-    onTurnAdopted: () =>
-      markSessionDeliveryAttemptStarted(
-        params.entry,
-        ...sessionDeliveryStateDirArgs(params.stateDir),
-      ),
+    turnAdoptionLifecycle: {
+      admission: "cancel-only",
+      onAdopted: () =>
+        markSessionDeliveryAttemptStarted(
+          params.entry,
+          ...sessionDeliveryStateDirArgs(params.stateDir),
+        ),
+    },
     delivery: {
       preparePayload: (payload) => {
         if (isRestartContinuationBusyPayload(payload)) {
diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts
index dbe3ab815e4f..d30af5cb280b 100644
--- a/src/gateway/server.chat.gateway-server-chat-b.test.ts
+++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts
@@ -3918,12 +3918,12 @@ describe("gateway server chat", () => {
         getRuntimeConfig: () => ({}),
         dedupe: new Map(),
       } as unknown as GatewayRequestContext;
-      let queuedLifecycle: GetReplyOptions["queuedFollowupLifecycle"];
+      let turnAdoptionLifecycle: GetReplyOptions["turnAdoptionLifecycle"];
       const dispatchRelease = createDeferred();
       dispatchInboundMessageMock.mockImplementationOnce(async (args: unknown) => {
-        queuedLifecycle = (args as { replyOptions?: GetReplyOptions }).replyOptions
-          ?.queuedFollowupLifecycle;
-        queuedLifecycle?.onEnqueued?.();
+        turnAdoptionLifecycle = (args as { replyOptions?: GetReplyOptions }).replyOptions
+          ?.turnAdoptionLifecycle;
+        turnAdoptionLifecycle?.onDeferred?.();
         await dispatchRelease.promise;
         return {};
       });
@@ -3963,8 +3963,8 @@ describe("gateway server chat", () => {
         context,
       });
 
-      await vi.waitFor(() => expect(queuedLifecycle).toBeDefined(), FAST_WAIT_OPTS);
-      expect(queuedLifecycle?.ownerKey).toBe("connection:conn-tui");
+      await vi.waitFor(() => expect(turnAdoptionLifecycle).toBeDefined(), FAST_WAIT_OPTS);
+      expect(turnAdoptionLifecycle?.ownerKey).toBe("connection:conn-tui");
       expect(broadcast).not.toHaveBeenCalledWith(
         "chat",
         expect.objectContaining({ runId: "idem-queued-followup", state: "final" }),
@@ -4034,18 +4034,18 @@ describe("gateway server chat", () => {
       queuedEntry?.controller.abort();
       expect(context.chatQueuedTurns.has("idem-queued-followup")).toBe(false);
 
-      queuedLifecycle?.onComplete?.();
+      turnAdoptionLifecycle?.onSettled?.();
       expect(context.chatQueuedTurns.has("idem-queued-followup")).toBe(false);
       await vi.waitFor(
         () => expect(context.removeChatRun).toHaveBeenCalledTimes(1),
         FAST_WAIT_OPTS,
       );
 
-      let failedDispatchLifecycle: GetReplyOptions["queuedFollowupLifecycle"];
+      let failedDispatchLifecycle: GetReplyOptions["turnAdoptionLifecycle"];
       dispatchInboundMessageMock.mockImplementationOnce(async (args: unknown) => {
         failedDispatchLifecycle = (args as { replyOptions?: GetReplyOptions }).replyOptions
-          ?.queuedFollowupLifecycle;
-        failedDispatchLifecycle?.onEnqueued?.();
+          ?.turnAdoptionLifecycle;
+        failedDispatchLifecycle?.onDeferred?.();
         throw new Error("post-enqueue bookkeeping failed");
       });
       await expectDefined(
@@ -4098,7 +4098,7 @@ describe("gateway server chat", () => {
         payload: { status: "ok" },
       });
       expect(context.chatQueuedTurns.has("idem-queued-followup-post-error")).toBe(true);
-      failedDispatchLifecycle?.onComplete?.();
+      failedDispatchLifecycle?.onSettled?.();
       expect(context.chatQueuedTurns.has("idem-queued-followup-post-error")).toBe(false);
     } finally {
       dispatchInboundMessageMock.mockReset();
diff --git a/src/plugin-sdk/channel-outbound.ts b/src/plugin-sdk/channel-outbound.ts
index e5dd201e206d..63471b3a6c2c 100644
--- a/src/plugin-sdk/channel-outbound.ts
+++ b/src/plugin-sdk/channel-outbound.ts
@@ -15,10 +15,21 @@ const loadChannelMessageRuntimeModule = createLazyRuntimeModule(
 
 export type { DurableMessageBatchSendResult } from "../channels/message/runtime.js";
 export {
+  bindIngressLifecycleToReplyOptions,
+  createChannelIngressDrain,
   createReplyPrefixContext,
   createReplyPrefixOptions,
   createTypingCallbacks,
   createChannelReplyPipeline as createChannelMessageReplyPipeline,
+  // Narrow drain seam by maintainer decision (#108924): factory, lifecycle binding,
+  // tuning constants, and processPidFromOwnerId (telegram transport display). All other
+  // claim/retry/adoption internals stay core-owned; test helpers live on the
+  // private-local plugin-state-test-runtime subpath.
+  DEFAULT_INGRESS_ADOPTION_STALL_MS,
+  DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
+  DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+  INGRESS_CLAIM_PROCESS_ID,
+  processPidFromOwnerId,
   resolveChannelSourceReplyDeliveryMode as resolveChannelMessageSourceReplyDeliveryMode,
 } from "../channels/message/index.js";
 // Bare interval/stop orchestration for channels that own their typing renewal
@@ -136,6 +147,7 @@ export type {
   ChannelMessageSendTextContext,
   ChannelMessageUnknownSendContext,
   ChannelMessageUnknownSendReconciliationResult,
+  ChannelIngressDrain,
   ChannelIngressQueue,
   ChannelIngressQueueClaim,
   ChannelIngressQueueClaimRef,
diff --git a/src/plugin-sdk/plugin-state-test-runtime.ts b/src/plugin-sdk/plugin-state-test-runtime.ts
index 0cde97beea62..fb3750b473e2 100644
--- a/src/plugin-sdk/plugin-state-test-runtime.ts
+++ b/src/plugin-sdk/plugin-state-test-runtime.ts
@@ -17,3 +17,16 @@ export {
   closeOpenClawStateDatabaseForTest,
   openOpenClawStateDatabase,
 } from "../state/openclaw-state-db.js";
+// Test-only ingress reliability helpers: core predicates polling/webhook tests
+// assert directly; excluded from the public SDK surface (private-local subpath).
+export {
+  INGRESS_CLAIM_LEASE_MS,
+  isIngressClaimOwnedByOtherLiveProcess,
+} from "../channels/message/ingress-claim-owner.js";
+export {
+  resolveIngressRetryDelayMs,
+  shouldDeadLetterRetryableIngressEvent,
+} from "../channels/message/ingress-retry-policy.js";
+// Test-only pairing-store seeding so channel tests exercise the real
+// store-backed authorization path instead of injecting fake readers.
+export { addChannelAllowFromStoreEntry } from "../pairing/pairing-store.js";
diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts
index 8320513b846a..4b0f93898cf3 100644
--- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts
+++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts
@@ -822,6 +822,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial<PluginRuntime> =
       openChannelIngressQueue: vi.fn(() => {
         throw new Error("openChannelIngressQueue mock is not configured");
       }) as unknown as PluginRuntime["state"]["openChannelIngressQueue"],
+      openChannelIngressDrain: vi.fn(() => {
+        throw new Error("openChannelIngressDrain mock is not configured");
+      }) as unknown as PluginRuntime["state"]["openChannelIngressDrain"],
     },
     tasks: {
       runs: {
diff --git a/src/plugins/registry-runtime.ts b/src/plugins/registry-runtime.ts
index 647bdd1affcd..1a6c67fdc89e 100644
--- a/src/plugins/registry-runtime.ts
+++ b/src/plugins/registry-runtime.ts
@@ -1,5 +1,6 @@
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
 import { normalizeOptionalAgentRuntimeId } from "../agents/agent-runtime-id.js";
+import { createChannelIngressDrain } from "../channels/message/ingress-drain.js";
 import { createChannelIngressQueue } from "../channels/message/ingress-queue.js";
 import type { SessionEntry } from "../config/sessions/types.js";
 import {
@@ -467,7 +468,9 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) {
         };
         if (prop === "state") {
           const baseState = getRuntimeProperty();
-          const assertPluginStateAllowed = (methodName: "openBlobStore" | "openKeyedStore") => {
+          const assertPluginStateAllowed = (
+            methodName: "openBlobStore" | "openKeyedStore" | "openChannelIngressDrain",
+          ) => {
             const record =
               pluginRuntimeRecordById.get(pluginId) ??
               registry.plugins.find((entry) => entry.id === pluginId);
@@ -506,6 +509,40 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) {
                 stateDir,
               });
             },
+            openChannelIngressDrain: <TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
+              options: Omit<
+                Parameters<
+                  typeof createChannelIngressDrain<TPayload, TMetadata, TCompletedMetadata>
+                >[0],
+                "queue"
+              > & {
+                queue?: ReturnType<
+                  typeof createChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>
+                >;
+                accountId?: string;
+                stateDir?: string;
+              },
+            ) => {
+              assertPluginStateAllowed("openChannelIngressDrain");
+              const stateDir = options.stateDir ?? baseState.resolveStateDir();
+              const queue =
+                options.queue ??
+                createChannelIngressQueue<TPayload, TMetadata, TCompletedMetadata>({
+                  channelId: pluginId,
+                  accountId: options.accountId,
+                  stateDir,
+                });
+              const {
+                queue: _queue,
+                accountId: _accountId,
+                stateDir: _stateDir,
+                ...drainOptions
+              } = options;
+              return createChannelIngressDrain<TPayload, TMetadata, TCompletedMetadata>({
+                ...drainOptions,
+                queue,
+              });
+            },
           } satisfies PluginRuntime["state"];
         }
         if (prop === "config") {
diff --git a/src/plugins/runtime/index.ts b/src/plugins/runtime/index.ts
index b48f88f94a3c..4030f65e3b0a 100644
--- a/src/plugins/runtime/index.ts
+++ b/src/plugins/runtime/index.ts
@@ -347,6 +347,11 @@ export function createPluginRuntime(_options: CreatePluginRuntimeOptions = {}):
           "openChannelIngressQueue is only available through the plugin runtime proxy.",
         );
       },
+      openChannelIngressDrain: () => {
+        throw new Error(
+          "openChannelIngressDrain is only available through the plugin runtime proxy.",
+        );
+      },
     },
     tasks,
     taskFlow,
diff --git a/src/plugins/runtime/types-core.ts b/src/plugins/runtime/types-core.ts
index 3db174cc999c..e79025d04600 100644
--- a/src/plugins/runtime/types-core.ts
+++ b/src/plugins/runtime/types-core.ts
@@ -1,4 +1,5 @@
 // Core runtime types define system, config, and task helper contracts for plugins.
+import type { CreateChannelIngressDrainOptions } from "../../channels/message/ingress-drain.js";
 import type { CreateChannelIngressQueueOptions } from "../../channels/message/ingress-queue.js";
 import type { ConfigMutationBase } from "../../config/mutation-types.js";
 import type { SessionPluginJsonValue } from "../../config/sessions/types.js";
@@ -437,6 +438,20 @@ export type PluginRuntimeCore = {
       TMetadata,
       TCompletedMetadata
     >;
+    openChannelIngressDrain: <TPayload, TMetadata = unknown, TCompletedMetadata = unknown>(
+      options: Omit<
+        CreateChannelIngressDrainOptions<TPayload, TMetadata, TCompletedMetadata>,
+        "queue"
+      > & {
+        queue?: import("../../channels/message/ingress-queue.js").ChannelIngressQueue<
+          TPayload,
+          TMetadata,
+          TCompletedMetadata
+        >;
+        accountId?: string;
+        stateDir?: string;
+      },
+    ) => import("../../channels/message/ingress-drain.js").ChannelIngressDrain;
   };
   tasks: {
     runs: PluginRuntimeTaskRuns;