fix: keep channel messages queued during restart recovery (#175078)

* fix: preserve channel messages during restart recovery

* refactor: keep reply recovery coordination outside admission

* test: use a typed Gateway context for recovery admission

* test: expect pending recovery in command owner coverage

docs/concepts/queue.md

@@ -213,6 +213,13 @@ Repeated preparation errors stop renewing the waiting heartbeat, allowing the
 existing ingress watchdog to apply its retry and dead-letter policy. Normal
 capacity waits keep their heartbeat.
 
+Transient session admission conflicts follow the same ingress retry policy as
+other retryable failures: by default, a message becomes eligible for dead-lettering
+only after both eight failed attempts and 24 hours, with capped retry backoff. The
+oldest pending message keeps its place ahead of later messages in the same lane.
+A terminal restart-recovery tombstone still fails immediately so an authorized
+reset can proceed.
+
 ## Lanes and scope
 
 - Applies to auto-reply agent runs across all inbound channels that use the gateway reply pipeline (WhatsApp web, Telegram, Slack, Discord, Signal, iMessage, webchat, etc.).

docs/gateway/restart-recovery.md

@@ -823,10 +823,11 @@ receipts remain intact; recovery uses its normal transcript replay checks.
 
 If a Gateway left a dead writer and an unfinished recovery cycle,
 `sessions.recover` reconciles that writer
-and starts a continuation in the same session. A new Control UI message also reconciles this
-state before admission, so a rejected send cannot trap the conversation in a
-"conversation changed" retry loop. Both paths preserve the session key and
-transcript. A live run or cloud worker still prevents this repair. Tombstoned
+and starts a continuation in the same session. New Control UI and channel messages
+also reconcile this state before admission, so a rejected send cannot trap the
+conversation in a "conversation changed" retry loop. Channel messages wait behind
+live recovery ownership and resume admission when it settles. These paths preserve
+the session key and transcript. A live run or cloud worker still prevents this repair. Tombstoned
 sessions retain their separate recovery path into a new session.
 
 If recovery fails during preparation before the agent starts, the Gateway restores

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

@@ -1027,49 +1027,81 @@ describe("Telegram durable ingress coalescing", () => {
     }
   });
 
-  it("bounds repeated session-start conflicts and drains the next Telegram update", async () => {
-    const poison = textUpdate({ updateId: 801, messageId: 1, text: "poison" });
-    const after = textUpdate({ updateId: 802, messageId: 2, text: "after" });
-    const poisonId = telegramQueueEventId(poison.update_id);
-    const sessionError = Object.assign(
-      new Error('Session "agent:main:telegram:direct:111" changed while starting work. Retry.'),
-      { code: "SESSION_WORK_START_CHANGED" },
-    );
-    downstreamTurns.mockImplementation(async (turn) => {
-      if ((turn.BodyForAgent ?? turn.Body ?? "").includes("poison")) {
-        throw sessionError;
-      }
-      return { queuedFinal: false, counts: { block: 0, final: 0, tool: 0 } };
-    });
-    await writeTelegramSpooledUpdate({ stateDir, update: poison });
-    await writeTelegramSpooledUpdate({ stateDir, update: after });
-    const queue = openTelegramIngressQueue({ stateDir });
-    for (let attempt = 1; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
-      const claim = await queue.claim(poisonId, { ownerId: `proof:${attempt}` });
-      if (!claim) {
-        throw new Error(`Expected setup claim ${attempt}`);
-      }
-      await queue.release(claim, {
-        lastError: sessionError.message,
-        releasedAt: Date.now() - 60 * 60 * 1_000,
+  it.each(["young", "aged"] as const)(
+    "preserves retry policy and ordering for a %s session-start conflict",
+    async (age) => {
+      vi.useFakeTimers({ toFake: ["Date"] });
+      const poison = textUpdate({ updateId: 801, messageId: 1, text: "poison" });
+      const after = textUpdate({ updateId: 802, messageId: 2, text: "after" });
+      const poisonId = telegramQueueEventId(poison.update_id);
+      const sessionError = Object.assign(
+        new Error('Session "agent:main:telegram:direct:111" changed while starting work. Retry.'),
+        { code: "SESSION_WORK_START_CHANGED" },
+      );
+      let conflictPending = true;
+      downstreamTurns.mockImplementation(async (turn) => {
+        if (conflictPending && (turn.BodyForAgent ?? turn.Body ?? "").includes("poison")) {
+          throw sessionError;
+        }
+        return { queuedFinal: false, counts: { block: 0, final: 0, tool: 0 } };
       });
-    }
-    const runtimeError = vi.fn();
-    const { monitor, telegramTransport } = await createMonitor({ onRuntimeError: runtimeError });
+      await writeTelegramSpooledUpdate({
+        stateDir,
+        update: poison,
+        now: Date.now() - (age === "aged" ? 24 * 60 * 60 * 1_000 : 0),
+      });
+      await writeTelegramSpooledUpdate({ stateDir, update: after });
+      const queue = openTelegramIngressQueue({ stateDir });
+      for (let attempt = 1; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) {
+        const claim = await queue.claim(poisonId, { ownerId: `proof:${attempt}` });
+        if (!claim) {
+          throw new Error(`Expected setup claim ${attempt}`);
+        }
+        await queue.release(claim, {
+          lastError: sessionError.message,
+          releasedAt: Date.now() - 60 * 60 * 1_000,
+        });
+      }
+      const runtimeError = vi.fn();
+      const { monitor, telegramTransport } = await createMonitor({ onRuntimeError: runtimeError });
 
-    monitor.start();
-    await monitor.waitForIdle();
-    expect(await queue.listFailed?.({ limit: "all" })).toEqual([
-      expect.objectContaining({ id: poisonId, reason: "session-start-conflict-retry-limit" }),
-    ]);
-    expect(await queue.listPending({ limit: "all" })).toEqual([]);
-    expect(
-      downstreamTurns.mock.calls.some(([turn]) =>
-        (turn.BodyForAgent ?? turn.Body ?? "").includes("after"),
-      ),
-    ).toBe(true);
+      monitor.start();
+      await monitor.waitForIdle();
+      if (age === "young") {
+        expect(await queue.listFailed?.({ limit: "all" })).toEqual([]);
+        expect(await queue.listPending({ limit: "all" })).toMatchObject([
+          { id: poisonId, attempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS },
+          { id: telegramQueueEventId(after.update_id), attempts: 0 },
+        ]);
+        expect(downstreamTurns).toHaveBeenCalledOnce();
 
-    await monitor.stop();
-    await telegramTransport.close();
-  });
+        conflictPending = false;
+        vi.setSystemTime(Date.now() + 3 * 60 * 1_000);
+        monitor.requestDrain();
+        await monitor.waitForIdle();
+        await assertSpoolTombstoned({ stateDir, updateIds: [801, 802] });
+        expect(downstreamTurns.mock.calls.map(([turn]) => turn.RawBody)).toEqual([
+          "poison",
+          "poison",
+          "after",
+        ]);
+        await monitor.admit(poison);
+        await monitor.waitForIdle();
+        expect(downstreamTurns).toHaveBeenCalledTimes(3);
+      } else {
+        expect(await queue.listFailed?.({ limit: "all" })).toEqual([
+          expect.objectContaining({ id: poisonId, reason: "retry-limit-exceeded" }),
+        ]);
+      }
+      expect(await queue.listPending({ limit: "all" })).toEqual([]);
+      expect(
+        downstreamTurns.mock.calls.some(([turn]) =>
+          (turn.BodyForAgent ?? turn.Body ?? "").includes("after"),
+        ),
+      ).toBe(true);
+
+      await monitor.stop();
+      await telegramTransport.close();
+    },
+  );
 });

src/agents/agent-command-recovery-owner.test.ts

@@ -479,7 +479,15 @@ describe("agent command restart recovery ownership", () => {
           sessionId: target.sessionId,
           target: { sessionKey, storePath: target.storePath },
         }),
-      ).resolves.toEqual({ kind: "invalidated", reason: "state_changed" });
+      ).resolves.toEqual({
+        kind: "invalidated",
+        reason: "recovery_pending",
+        sessionKey,
+        entry: expect.objectContaining({
+          sessionId: target.sessionId,
+          ...runningRecovery(lifecycleGeneration),
+        }),
+      });
       let owner: Awaited<ReturnType<typeof startOwner>> | undefined;
       const ownerStarted = createDeferred();
       if (timing === "before preparation") {

src/agents/main-session-recovery/main-session-recovery-store.ts

@@ -9,6 +9,7 @@ import { applySessionEntryReplacements } from "../../config/sessions/session-acc
 import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
 import {
   isMainSessionRecoveryPending,
+  isMainSessionRecoveryReconciliationCandidate,
   transitionMainSessionRecovery,
   type MainSessionRecoveryCommand,
   type MainSessionRecoveryOwnerClaim,
@@ -297,6 +298,18 @@ export async function claimMainSessionRecoveryOwner(params: {
   ) {
     return { kind: "not_required", entry: claim.entry, sessionKey: claim.sessionKey } as const;
   }
+  if (
+    claim.entry?.sessionId === params.sessionId &&
+    claim.sessionKey &&
+    isMainSessionRecoveryReconciliationCandidate(claim.entry)
+  ) {
+    return {
+      kind: "invalidated",
+      reason: "recovery_pending",
+      entry: claim.entry,
+      sessionKey: claim.sessionKey,
+    } as const;
+  }
   const reason = claim.transition.kind === "rejected" ? claim.transition.reason : "state_changed";
   return { kind: "invalidated", reason } as const;
 }

src/auto-reply/reply/reply-turn-admission.recovery.test.ts

@@ -12,8 +12,10 @@ import * as restartRecovery from "../../agents/main-session-recovery/main-sessio
 import { loadSessionEntry, replaceSessionEntry } from "../../config/sessions/session-accessor.js";
 import type { InternalSessionEntry as SessionEntry } from "../../config/sessions/types.js";
 import type { GatewayRecoveryRuntime } from "../../gateway/server-instance-runtime.types.js";
-import type { GatewayRequestContext } from "../../gateway/server-methods/types.js";
+import { createGatewayRequestContext } from "../../gateway/server-request-context.js";
+import { makeContextParams } from "../../gateway/server-request-context.test-support.js";
 import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
+import { bindGatewayContextResolver } from "../../plugins/runtime/gateway-request-scope.js";
 import {
   beginSessionWorkAdmission,
   consumeSessionWorkAdmissionHandoff,
@@ -23,7 +25,7 @@ import {
   runExclusiveSessionLifecycleMutation,
   type SessionWorkAdmissionLease,
 } from "../../sessions/session-lifecycle-admission.js";
-import { replyRunRegistry } from "./reply-run-registry.js";
+import { createReplyOperation, replyRunRegistry } from "./reply-run-registry.js";
 import { testing } from "./reply-run-registry.test-support.js";
 import { admitTestReplyTurn, createSessionStore } from "./reply-turn-admission.test-support.js";
 
@@ -38,7 +40,7 @@ afterEach(async () => {
   testing.resetReplyRunRegistry();
   vi.restoreAllMocks();
 });
-function createRecoveryGatewayContext() {
+function createRecoveryGatewayContext(storePath?: string) {
   const recoveryRuntime: GatewayRecoveryRuntime = {
     prepareRestartRecovery: () => undefined,
     dispatchSessionMethod: vi.fn(),
@@ -46,8 +48,11 @@ function createRecoveryGatewayContext() {
     waitForAgent: vi.fn(),
     sendRecoveryNotice: vi.fn(),
   };
-  // The recovery boundary supplies execution; admission consumes these capabilities.
-  return { getRuntimeConfig: () => ({}), recoveryRuntime } as GatewayRequestContext;
+  const context = createGatewayRequestContext(makeContextParams());
+  context.getRuntimeConfig = () => (storePath ? { session: { store: storePath } } : {});
+  context.recoveryRuntime = recoveryRuntime;
+  context.workerSessionPlacementService = { getMany: () => new Map() };
+  return context;
 }
 function complete(result: Admission | undefined) {
   if (result?.status === "owned") {
@@ -321,6 +326,136 @@ it("preserves live recovery authority while monitoring", async () => {
   await owner.released;
 });
 
+it.each(["settled", "cancelled", "replaced"] as const)(
+  "keeps visible input behind an admitted recovery fence until %s",
+  async (outcome) => {
+    const f = recoveryFixture({
+      status: undefined,
+      abortedLastRun: false,
+      mainRestartRecovery: { cycleId: "cycle-1", revision: 1, chargedAttempts: 1 },
+      restartRecoveryRuns: [
+        { runId: "admitted-recovery", lifecycleGeneration: getAgentEventLifecycleGeneration() },
+      ],
+    });
+    const predecessor = await f.begin();
+    const stored = f.read();
+    const attempted = createDeferred();
+    const claim = recoveryStore.claimMainSessionRecoveryOwner;
+    vi.spyOn(recoveryStore, "claimMainSessionRecoveryOwner").mockImplementation(async (params) => {
+      const result = await claim(params);
+      attempted.resolve();
+      return result;
+    });
+    const context = createRecoveryGatewayContext(f.storePath);
+    const admission = f.wait({ resolveGatewayContext: () => context, waitForActive: false });
+    await attempted.promise;
+    await setImmediate();
+    expect(admission.failure).toBeUndefined();
+    expect(admission.result).toBeUndefined();
+    expect(replyRunRegistry.get(sessionKey)).toBeUndefined();
+    expect(f.read()).toEqual(stored);
+    if (outcome === "cancelled") {
+      f.abort.abort();
+    } else {
+      await f.write({
+        sessionId: outcome === "replaced" ? "replacement-session" : sessionId,
+        updatedAt: Date.now(),
+        status: "done",
+      });
+    }
+    predecessor.release();
+    await admission.settled;
+    if (outcome === "replaced") {
+      expect(admission.failure).toMatchObject({ code: "SESSION_WORK_START_CHANGED" });
+      expect(admission.result).toBeUndefined();
+    } else {
+      expect(admission.failure).toBeUndefined();
+      expect(admission.result).toMatchObject(
+        outcome === "settled" ? { status: "owned" } : { status: "skipped", reason: "aborted" },
+      );
+    }
+  },
+);
+
+it("rejects command adoption without waiting behind a live recovery fence", async () => {
+  const f = recoveryFixture({
+    status: undefined,
+    abortedLastRun: false,
+    mainRestartRecovery: { cycleId: "cycle-1", revision: 1, chargedAttempts: 0 },
+    restartRecoveryRuns: [
+      { runId: "active-run", lifecycleGeneration: getAgentEventLifecycleGeneration() },
+    ],
+  });
+  const context = createRecoveryGatewayContext(f.storePath);
+  await f.begin();
+  const blocker = createReplyOperation({ sessionKey, sessionId, resetTriggered: false });
+  blocker.setPhase("running");
+  const sourceKey = "agent:main:discord:slash:source";
+  const reservation = createReplyOperation({
+    sessionKey: sourceKey,
+    sessionId: "source-session",
+    resetTriggered: false,
+  });
+  bindGatewayContextResolver(reservation, () => context);
+  f.cleanup.push(
+    () => blocker.complete(),
+    () => reservation.complete(),
+  );
+  const before = f.read();
+  const attempted = createDeferred();
+  const claim = recoveryStore.claimMainSessionRecoveryOwner;
+  vi.spyOn(recoveryStore, "claimMainSessionRecoveryOwner").mockImplementation(async (params) => {
+    const result = await claim(params);
+    attempted.resolve();
+    return result;
+  });
+  const admission = f.wait({
+    sessionId: reservation.sessionId,
+    waitForActive: false,
+    adoptOperation: reservation,
+  });
+  await attempted.promise;
+  await setImmediate();
+  expect(admission.failure).toMatchObject({
+    code: "SESSION_WORK_START_CHANGED",
+  });
+  expect(admission.result).toBeUndefined();
+  expect(replyRunRegistry.get(sessionKey)).toBe(blocker);
+  expect(replyRunRegistry.get(sourceKey)).toBe(reservation);
+  expect(reservation.key).toBe(sourceKey);
+  expect(reservation.result).toBeNull();
+  expect(f.read()).toEqual(before);
+});
+
+it("reconciles an orphaned recovery fence before admitting channel input", async () => {
+  const f = recoveryFixture({
+    status: undefined,
+    abortedLastRun: false,
+    mainRestartRecovery: { cycleId: "cycle-1", revision: 1, chargedAttempts: 1 },
+    restartRecoveryRuns: [
+      { runId: "admitted-recovery", lifecycleGeneration: getAgentEventLifecycleGeneration() },
+    ],
+    restartRecoveryDeliveryRunId: "old-channel-claim",
+    restartRecoveryDeliverySourceRunId: "old-channel-source",
+  });
+  const context = createRecoveryGatewayContext(f.storePath);
+  const retry = vi
+    .spyOn(restartRecovery, "retryRestartAbortedMainSessionRecovery")
+    .mockImplementationOnce(async (request) => {
+      expect(request.expectedRecoverySourceRunId).toBe("old-channel-source");
+      expect(f.read()).toMatchObject({
+        sessionId,
+        abortedLastRun: true,
+        restartRecoveryDeliverySourceRunId: "old-channel-source",
+      });
+      expect(replyRunRegistry.get(sessionKey)).toBeUndefined();
+      await f.write({ sessionId, updatedAt: Date.now(), status: "done" });
+      return { started: 1, settled: 0, failed: 0, skipped: 0 };
+    });
+  owned(await f.admit({ resolveGatewayContext: () => context }));
+  expect(retry).toHaveBeenCalledOnce();
+});
+
 async function busyReply(withRecoveryFence = true) {
   const fixture = recoveryFixture({ status: undefined, abortedLastRun: false });
   const reply = owned(await fixture.admit());

src/auto-reply/reply/reply-turn-admission.ts

@@ -4,7 +4,6 @@ import { MAIN_SESSION_RECOVERY_WORK_ADMISSION_OWNER } from "../../agents/main-se
 import { scheduleMainSessionRecoveryPendingTarget } from "../../agents/main-session-recovery/main-session-recovery-owner-release.js";
 import {
   claimMainSessionRecoveryOwner,
-  releaseMainSessionRecoveryOwner,
   type MainSessionRecoveryOwnerLease,
 } from "../../agents/main-session-recovery/main-session-recovery-store.js";
 import { isAgentRunRestartAbortReason } from "../../agents/run-termination.js";
@@ -72,24 +71,16 @@ import {
 import { REPLY_WORK_ADMISSION_OWNER } from "./reply-turn-admission-owner.js";
 import type { ReplyTurnAdmission, ReplyTurnAdmissionParams } from "./reply-turn-admission.types.js";
 import { bindReplyOperationDatabaseAdmission } from "./reply-turn-database-admission.js";
-import { waitForRestartRecoveryProgress } from "./reply-turn-recovery-wait.js";
+import {
+  reconcileOrWaitForReplyRecovery,
+  releaseReplyRecoveryOwner,
+  waitForRestartRecoveryProgress,
+} from "./reply-turn-recovery.js";
 
 class QueuedFollowupLifecycleInvalidatedError extends Error {}
 
 const log = createSubsystemLogger("auto-reply/reply-turn-admission");
 
-async function releaseReplyRecoveryOwner(lease: MainSessionRecoveryOwnerLease | undefined) {
-  try {
-    return await releaseMainSessionRecoveryOwner(lease);
-  } catch (error) {
-    log.warn(`failed to release main-session recovery reply owner: ${formatErrorMessage(error)}`);
-    // The durable owner schedules exact-token retries. A completed reply must
-    // not keep its successor barrier and lifecycle admission until that
-    // background repair wins a contested SQLite write.
-    return undefined;
-  }
-}
-
 /** Runs owner work with its admission marked as the initiating lifecycle context. */
 export async function runWithReplyOperationLifecycleAdmission<T>(
   operation: ReplyOperation,
@@ -478,6 +469,32 @@ export async function admitReplyTurn(
               target: { agentId: params.agentId, sessionKey: params.sessionKey, storePath },
             });
             if (ownerClaim.kind === "invalidated") {
+              if (
+                ownerClaim.reason === "recovery_pending" &&
+                !params.adoptOperation &&
+                gatewayContext?.recoveryRuntime
+              ) {
+                // Release our admission so recovery can reclaim an orphan.
+                admission?.release();
+                await reconcileOrWaitForReplyRecovery({
+                  agentId: params.agentId,
+                  sessionKey: params.sessionKey,
+                  storePath,
+                  entry: ownerClaim.entry,
+                  gatewayContext,
+                  signal: params.upstreamAbortSignal,
+                  assertCurrent: () => {
+                    params.upstreamAbortSignal?.throwIfAborted();
+                    params.assertRequestCurrent?.();
+                    workspaceReadiness?.assertCurrent();
+                    assertDatabaseOwnerCurrent();
+                    if (lifecycleGeneration !== getAgentEventLifecycleGeneration()) {
+                      rejectSessionChange();
+                    }
+                  },
+                });
+                continue;
+              }
               rejectSessionChange();
             }
             recoveryOwnerLease = ownerClaim.kind === "claimed" ? ownerClaim.lease : undefined;

src/auto-reply/reply/reply-turn-recovery-wait.ts

@@ -1,34 +0,0 @@
-import { raceWithTimeout } from "@openclaw/retry";
-import { DEFAULT_RECOVERY_DELAY_MS } from "../../agents/main-session-recovery/main-session-restart-recovery-shared.js";
-import { sessionChanges } from "../../sessions/session-row-changes.js";
-import { createDeferredCore } from "../../shared/deferred.js";
-
-export async function waitForRestartRecoveryProgress(params: {
-  agentId?: string;
-  sessionKey: string;
-  ownerRelease?: Promise<void>;
-  signal?: AbortSignal;
-}): Promise<void> {
-  const changed = createDeferredCore();
-  const unsubscribe = sessionChanges.subscribe((change) => {
-    if (
-      "all" in change ||
-      (change.sessionKey === params.sessionKey &&
-        (!params.agentId || !change.agentId || change.agentId === params.agentId))
-    ) {
-      changed.resolve();
-    }
-  });
-  // Retry deferred dispatches without spinning; also cover a commit that won
-  // just before subscription. Every wake revalidates the session and owner.
-  try {
-    await raceWithTimeout(
-      params.ownerRelease ? Promise.race([changed.promise, params.ownerRelease]) : changed.promise,
-      DEFAULT_RECOVERY_DELAY_MS,
-      () => undefined,
-      { ref: false, signal: params.signal },
-    );
-  } finally {
-    unsubscribe();
-  }
-}

src/auto-reply/reply/reply-turn-recovery.ts

@@ -0,0 +1,99 @@
+import { raceWithTimeout } from "@openclaw/retry";
+import {
+  releaseMainSessionRecoveryOwner,
+  type MainSessionRecoveryOwnerLease,
+} from "../../agents/main-session-recovery/main-session-recovery-store.js";
+import { DEFAULT_RECOVERY_DELAY_MS } from "../../agents/main-session-recovery/main-session-restart-recovery-shared.js";
+import type { InternalSessionEntry } from "../../config/sessions/types.js";
+import type { GatewayRequestContext } from "../../gateway/server-methods/types.js";
+import { formatErrorMessage } from "../../infra/errors.js";
+import { createSubsystemLogger } from "../../logging/subsystem.js";
+import { getPluginRuntimeGatewayRequestScope } from "../../plugins/runtime/gateway-request-scope.js";
+import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
+import { getSessionWorkAdmissionRelease } from "../../sessions/session-lifecycle-admission.js";
+import { sessionChanges } from "../../sessions/session-row-changes.js";
+import { createDeferredCore } from "../../shared/deferred.js";
+
+const log = createSubsystemLogger("auto-reply/reply-turn-admission");
+
+export async function releaseReplyRecoveryOwner(lease: MainSessionRecoveryOwnerLease | undefined) {
+  try {
+    return await releaseMainSessionRecoveryOwner(lease);
+  } catch (error) {
+    log.warn(`failed to release main-session recovery reply owner: ${formatErrorMessage(error)}`);
+    // The durable owner schedules exact-token retries. A completed reply must
+    // not keep its successor barrier and lifecycle admission until that
+    // background repair wins a contested SQLite write.
+    return undefined;
+  }
+}
+
+/** Retained recovery is busy work; only the Gateway owner may reclaim it. */
+export async function reconcileOrWaitForReplyRecovery(params: {
+  agentId?: string;
+  sessionKey: string;
+  storePath: string;
+  entry: InternalSessionEntry;
+  gatewayContext: GatewayRequestContext;
+  signal?: AbortSignal;
+  assertCurrent: () => void;
+}): Promise<void> {
+  const { reconcileOrphanedGatewaySessionRecovery } =
+    await import("../../gateway/session-recovery-service.js");
+  const { resolveSessionWorkerPlacementContext } =
+    await import("../../gateway/session-worker-placement-context.js");
+  const recovered = await reconcileOrphanedGatewaySessionRecovery({
+    cfg: params.gatewayContext.getRuntimeConfig(),
+    entry: params.entry,
+    target: {
+      agentId: params.agentId ?? resolveAgentIdFromSessionKey(params.sessionKey),
+      storePath: params.storePath,
+      canonicalKey: params.sessionKey,
+      storeKeys: [params.sessionKey],
+    },
+    authorizedPluginId: getPluginRuntimeGatewayRequestScope()?.pluginId,
+    commitGuard: params.assertCurrent,
+    workerPlacementContext: resolveSessionWorkerPlacementContext(params.gatewayContext),
+  });
+  if (!recovered) {
+    await waitForRestartRecoveryProgress({
+      agentId: params.agentId,
+      sessionKey: params.sessionKey,
+      signal: params.signal,
+      ownerRelease: getSessionWorkAdmissionRelease({
+        scope: params.storePath,
+        identities: [params.sessionKey, params.entry.sessionId],
+      }),
+    });
+  }
+}
+
+export async function waitForRestartRecoveryProgress(params: {
+  agentId?: string;
+  sessionKey: string;
+  ownerRelease?: Promise<void>;
+  signal?: AbortSignal;
+}): Promise<void> {
+  const changed = createDeferredCore();
+  const unsubscribe = sessionChanges.subscribe((change) => {
+    if (
+      "all" in change ||
+      (change.sessionKey === params.sessionKey &&
+        (!params.agentId || !change.agentId || change.agentId === params.agentId))
+    ) {
+      changed.resolve();
+    }
+  });
+  // Retry deferred dispatches without spinning; also cover a commit that won
+  // just before subscription. Every wake revalidates the session and owner.
+  try {
+    await raceWithTimeout(
+      params.ownerRelease ? Promise.race([changed.promise, params.ownerRelease]) : changed.promise,
+      DEFAULT_RECOVERY_DELAY_MS,
+      () => undefined,
+      { ref: false, signal: params.signal },
+    );
+  } finally {
+    unsubscribe();
+  }
+}

src/channels/message/ingress-retry-policy.test.ts

@@ -242,7 +242,7 @@ describe("ingress retry policy", () => {
     });
   });
 
-  it("bounds a wrapped session-start conflict at the configured attempt budget", () => {
+  it("keeps wrapped session-start conflicts under both retry and age gates", () => {
     const maxAttempts = 3;
     const message = 'Session "agent:main:telegram:direct:1" changed while starting work. Retry.';
     const conflict = Object.assign(new Error(message), {
@@ -295,10 +295,18 @@ describe("ingress retry policy", () => {
       now: 2_000,
     });
     expect(atLimit).toEqual({
-      kind: "fail",
-      reason: "session-start-conflict-retry-limit",
+      kind: "release",
       message: wrapped.message,
       attempt: maxAttempts,
     });
+    expect(
+      resolveIngressFailureDisposition({
+        err: wrapped,
+        event: { receivedAt: 1_000, attempts: maxAttempts - 1 },
+        formatError: coerceErrorMessage,
+        config: { maxAttempts },
+        now: 1_000 + DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
+      }),
+    ).toMatchObject({ kind: "fail", reason: "retry-limit-exceeded", attempt: maxAttempts });
   });
 });

src/channels/message/ingress-retry-policy.ts

@@ -7,10 +7,7 @@ import {
   collectNestedErrorCandidates,
   extractErrorCode,
 } from "@openclaw/normalization-core/error-coercion";
-import {
-  SESSION_RESTART_RECOVERY_TOMBSTONE_ERROR_CODE,
-  SESSION_WORK_START_CHANGED_ERROR_CODE,
-} from "../../config/sessions/work-start-error.js";
+import { SESSION_RESTART_RECOVERY_TOMBSTONE_ERROR_CODE } from "../../config/sessions/work-start-error.js";
 import { computeBackoff } from "../../infra/backoff.js";
 
 export const DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS = 8;
@@ -95,7 +92,6 @@ export function resolveIngressFailureDisposition(params: {
   now?: number;
 }): IngressFailureDisposition {
   const now = params.now ?? Date.now();
-  const maxAttempts = params.config?.maxAttempts ?? DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS;
   const attempt = (params.event.attempts ?? 0) + 1;
   const message = params.formatError(params.err);
   const nonRetryable = params.resolveNonRetryableFailure?.(params.err) ?? null;
@@ -111,11 +107,9 @@ export function resolveIngressFailureDisposition(params: {
   // Retrying this terminal generation blocks the authorized reset behind it.
   const reason = errorCodes.has(SESSION_RESTART_RECOVERY_TOMBSTONE_ERROR_CODE)
     ? "restart-recovery-tombstone"
-    : attempt >= maxAttempts && errorCodes.has(SESSION_WORK_START_CHANGED_ERROR_CODE)
-      ? "session-start-conflict-retry-limit"
-      : shouldDeadLetterRetryableIngressEvent(params.event, attempt, params.config, now)
-        ? "retry-limit-exceeded"
-        : undefined;
+    : shouldDeadLetterRetryableIngressEvent(params.event, attempt, params.config, now)
+      ? "retry-limit-exceeded"
+      : undefined;
   return reason
     ? { kind: "fail", reason, message, attempt }
     : { kind: "release", attempt, message };

src/infra/outbound/session-binding-routing.worker.test.ts

@@ -127,7 +127,7 @@ it.each(
   (["generic", "account"] as const).flatMap((kind) =>
     (["owner-retired", "database-closed"] as const).map((change) => ({ kind, change })),
   ),
-)("preserves bounded ingress classification for $kind native $change", async ({ kind, change }) => {
+)("retries native $kind $change under the ingress age gate", async ({ kind, change }) => {
   const previousRegistry = captureActivePluginRegistrySnapshot();
   const channel = kind === "generic" ? "native-binding-regression" : "imessage";
   const conversation = { channel, accountId: "default", conversationId: "native-admission" };
@@ -193,16 +193,11 @@ it.each(
         now,
       });
     expect(disposition(6)).toMatchObject({ kind: "release", attempt: 7 });
+    expect(disposition(7)).toMatchObject({ kind: "release", attempt: 8 });
     if (change === "owner-retired") {
-      expect(disposition(7)).toMatchObject({
-        kind: "fail",
-        reason: "session-start-conflict-retry-limit",
-        attempt: 8,
-      });
       expect(failure).toMatchObject({ code: "SESSION_WORK_START_CHANGED" });
     } else {
       expect(failure).toMatchObject({ code: "STATE_DATABASE_READ_ADMISSION_INVALIDATED" });
-      expect(disposition(7)).toMatchObject({ kind: "release", attempt: 8 });
     }
   } finally {
     manager?.stop();