fix(sessions): reconcile client-abandoned running sessions on restart

src/agents/agent-command.live-model-switch.test.ts

@@ -328,7 +328,9 @@ vi.mock("../infra/agent-events.js", () => ({
   clearAgentRunContext: (...args: unknown[]) => state.clearAgentRunContextMock(...args),
   emitAgentEvent: (...args: unknown[]) => state.emitAgentEventMock(...args),
   getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
   onAgentEvent: vi.fn(),
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
   registerAgentRunContext: (...args: unknown[]) => state.registerAgentRunContextMock(...args),
   withAgentRunLifecycleGeneration: (_generation: string, run: () => unknown) => run(),
 }));

src/agents/embedded-agent-runner/run-state.ts

@@ -12,6 +12,10 @@ import {
   resolveActiveReplyRunSessionId,
   type ReplyBackendQueueMessageOptions,
 } from "../../auto-reply/reply/reply-run-registry.js";
+import {
+  isAgentEventLifecycleGenerationCurrent,
+  registerAgentEventLifecycleRotationHandler,
+} from "../../infra/agent-events.js";
 import { resolveGlobalSingleton } from "../../shared/global-singleton.js";
 
 /**
@@ -63,6 +67,7 @@ const EMBEDDED_RUN_STATE_KEY = Symbol.for("openclaw.embeddedRunState");
 const embeddedRunState = resolveGlobalSingleton(EMBEDDED_RUN_STATE_KEY, () => ({
   activeRuns: new Map<string, EmbeddedAgentQueueHandle>(),
   activeRunsByRunId: new Map<string, EmbeddedAgentQueueHandle>(),
+  activeRunLifecycleGenerations: new WeakMap<EmbeddedAgentQueueHandle, string>(),
   retainedAbortabilityRunIds: new Set<string>(),
   snapshots: new Map<string, ActiveEmbeddedRunSnapshot>(),
   sessionIdsByKey: new Map<string, string>(),
@@ -79,6 +84,12 @@ export const ACTIVE_EMBEDDED_RUNS =
 export const ACTIVE_EMBEDDED_RUNS_BY_RUN_ID =
   embeddedRunState.activeRunsByRunId ??
   (embeddedRunState.activeRunsByRunId = new Map<string, EmbeddedAgentQueueHandle>());
+export const ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS =
+  embeddedRunState.activeRunLifecycleGenerations ??
+  (embeddedRunState.activeRunLifecycleGenerations = new WeakMap<
+    EmbeddedAgentQueueHandle,
+    string
+  >());
 export const RETAINED_EMBEDDED_RUN_ABORTABILITY_RUN_IDS =
   embeddedRunState.retainedAbortabilityRunIds ??
   (embeddedRunState.retainedAbortabilityRunIds = new Set<string>());
@@ -104,6 +115,71 @@ export const EMBEDDED_RUN_WAITERS =
   embeddedRunState.waiters ??
   (embeddedRunState.waiters = new Map<string, Set<EmbeddedRunWaiter>>());
 
+function evictPriorLifecycleEmbeddedRuns(): void {
+  const staleHandles = new Set<EmbeddedAgentQueueHandle>();
+  for (const [sessionId, handle] of ACTIVE_EMBEDDED_RUNS) {
+    const lifecycleGeneration = ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle);
+    if (lifecycleGeneration && isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)) {
+      continue;
+    }
+    staleHandles.add(handle);
+    if (ACTIVE_EMBEDDED_RUNS.get(sessionId) === handle) {
+      ACTIVE_EMBEDDED_RUNS.delete(sessionId);
+    }
+    ACTIVE_EMBEDDED_RUN_SNAPSHOTS.delete(sessionId);
+  }
+  for (const [runId, handle] of ACTIVE_EMBEDDED_RUNS_BY_RUN_ID) {
+    const lifecycleGeneration = ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle);
+    if (lifecycleGeneration && isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)) {
+      continue;
+    }
+    staleHandles.add(handle);
+    // This index only gates the separately owned chat abort controller; absence
+    // is abortable. Keeping it would let stale ownership influence new work.
+    if (ACTIVE_EMBEDDED_RUNS_BY_RUN_ID.get(runId) === handle) {
+      ACTIVE_EMBEDDED_RUNS_BY_RUN_ID.delete(runId);
+      RETAINED_EMBEDDED_RUN_ABORTABILITY_RUN_IDS.delete(runId);
+    }
+  }
+  for (const [sessionKey, sessionId] of ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_KEY) {
+    if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) {
+      ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_KEY.delete(sessionKey);
+    }
+  }
+  for (const [sessionFile, sessionId] of ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_FILE) {
+    if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) {
+      ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_FILE.delete(sessionFile);
+    }
+  }
+  for (const [sessionId, waiters] of EMBEDDED_RUN_WAITERS) {
+    if (ACTIVE_EMBEDDED_RUNS.has(sessionId)) {
+      continue;
+    }
+    EMBEDDED_RUN_WAITERS.delete(sessionId);
+    for (const waiter of waiters) {
+      if (waiter.timer) {
+        clearTimeout(waiter.timer);
+      }
+      waiter.resolve(true);
+    }
+  }
+  const abortErrors: unknown[] = [];
+  // Remove stale ownership first so synchronous abort callbacks may register a
+  // replacement without the cleanup above erasing that current-generation run.
+  for (const handle of staleHandles) {
+    try {
+      handle.abort("restart");
+    } catch (error) {
+      abortErrors.push(error);
+    }
+  }
+  if (abortErrors.length > 0) {
+    throw new AggregateError(abortErrors, "Failed to abort stale embedded agent runs");
+  }
+}
+
+registerAgentEventLifecycleRotationHandler("embedded-agent-runs", evictPriorLifecycleEmbeddedRuns);
+
 /** Counts active embedded runs while including auto-reply registry runs for shared sessions. */
 export function getActiveEmbeddedRunCount(): number {
   let activeCount = ACTIVE_EMBEDDED_RUNS.size;
@@ -137,6 +213,20 @@ export function listActiveEmbeddedRunSessionIds(): string[] {
   ].toSorted((a, b) => a.localeCompare(b));
 }
 
+export function setActiveEmbeddedRunLifecycleGeneration(
+  handle: EmbeddedAgentQueueHandle,
+  lifecycleGeneration: string,
+): string {
+  // A delayed re-registration must not transfer an old driver into the new
+  // Gateway lifecycle and suppress orphan recovery again.
+  const existingLifecycleGeneration = ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle);
+  if (existingLifecycleGeneration !== undefined) {
+    return existingLifecycleGeneration;
+  }
+  ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.set(handle, lifecycleGeneration);
+  return lifecycleGeneration;
+}
+
 /** Resolves the current session id for an active run after resets or compaction. */
 export function resolveActiveEmbeddedRunSessionId(sessionKey: string): string | undefined {
   const normalizedSessionKey = sessionKey.trim();

src/agents/embedded-agent-runner/run/attempt-after-turn.ts

@@ -3,6 +3,7 @@
  */
 import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../../context-engine/host-compat.js";
 import type { ContextEngine } from "../../../context-engine/types.js";
+import { captureAgentRunLifecycleGeneration } from "../../../infra/agent-events.js";
 import { freezeDiagnosticTraceContext } from "../../../infra/diagnostic-trace-context.js";
 import { formatErrorMessage } from "../../../infra/errors.js";
 import type { getGlobalHookRunner } from "../../../plugins/hook-runner-global.js";
@@ -185,7 +186,11 @@ export async function completeEmbeddedAttemptAfterTurn(
           if (rotation.rotated) {
             sessionIdUsed = rotation.sessionId ?? sessionIdUsed;
             sessionFileUsed = rotation.sessionFile ?? sessionFileUsed;
-            updateActiveEmbeddedRunSessionFile(attempt.sessionId, sessionFileUsed);
+            updateActiveEmbeddedRunSessionFile(
+              attempt.sessionId,
+              sessionFileUsed,
+              attempt.lifecycleGeneration ?? captureAgentRunLifecycleGeneration(attempt.runId),
+            );
             log.info(
               `[compaction] rotated active transcript after automatic compaction ` +
                 `(sessionKey=${attempt.sessionKey ?? attempt.sessionId})`,

src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts

@@ -3,6 +3,7 @@
  */
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
 import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../../../auto-reply/tokens.js";
+import { captureAgentRunLifecycleGeneration } from "../../../infra/agent-events.js";
 import {
   freezeDiagnosticTraceContext,
   type DiagnosticTraceContext,
@@ -30,6 +31,7 @@ import {
   type ToolSearchTargetTranscriptProjection,
 } from "../../tool-search.js";
 import { log } from "../logger.js";
+import { setActiveEmbeddedRunLifecycleGeneration } from "../run-state.js";
 import {
   clearActiveEmbeddedRun,
   type EmbeddedAgentQueueHandle,
@@ -374,6 +376,10 @@ export function prepareEmbeddedAttemptStream(input: {
     abort: (reason) => abortActiveRunExternally(reason),
   };
   attempt.replyOperation?.attachBackend(queueHandle);
+  setActiveEmbeddedRunLifecycleGeneration(
+    queueHandle,
+    attempt.lifecycleGeneration ?? captureAgentRunLifecycleGeneration(attempt.runId),
+  );
   setActiveEmbeddedRun(attempt.sessionId, queueHandle, attempt.sessionKey, attempt.sessionFile);
 
   return {

src/agents/embedded-agent-runner/runs.generation.test.ts

@@ -0,0 +1,263 @@
+import { importFreshModule } from "openclaw/plugin-sdk/test-fixtures";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { testing as replyRunTesting } from "../../auto-reply/reply/reply-run-registry.test-support.js";
+import {
+  getAgentEventLifecycleGeneration,
+  rotateAgentEventLifecycleGeneration,
+} from "../../infra/agent-events.js";
+import {
+  listActiveEmbeddedRunSessionIds,
+  listActiveEmbeddedRunSessionKeys,
+  setActiveEmbeddedRunLifecycleGeneration,
+  type EmbeddedAgentQueueHandle,
+} from "./run-state.js";
+import {
+  clearActiveEmbeddedRun,
+  isEmbeddedAgentRunAbortableForRunId,
+  queueEmbeddedAgentMessageWithOutcomeAsync,
+  resolveActiveEmbeddedRunHandleSessionId,
+  resolveActiveEmbeddedRunHandleSessionIdBySessionFile,
+  setActiveEmbeddedRun,
+  updateActiveEmbeddedRunSessionFile,
+} from "./runs.js";
+import { testing } from "./runs.test-support.js";
+
+function createRunHandle(params: {
+  abort?: EmbeddedAgentQueueHandle["abort"];
+  queueMessage: EmbeddedAgentQueueHandle["queueMessage"];
+  runId: string;
+}): EmbeddedAgentQueueHandle {
+  return {
+    kind: "embedded",
+    runId: params.runId,
+    queueMessage: params.queueMessage,
+    isStreaming: () => true,
+    isAbortable: () => false,
+    isCompacting: () => false,
+    abort: params.abort ?? (() => {}),
+  };
+}
+
+describe("embedded run registry lifecycle generations", () => {
+  afterEach(() => {
+    testing.resetActiveEmbeddedRuns();
+    replyRunTesting.resetReplyRunRegistry();
+  });
+
+  it("rejects a delayed prior-lifecycle registration for a current session owner", async () => {
+    const priorLifecycleGeneration = getAgentEventLifecycleGeneration();
+    const staleQueueMessage = vi.fn(async () => {});
+    const staleAbort = vi.fn();
+    const staleHandle = createRunHandle({
+      abort: staleAbort,
+      queueMessage: staleQueueMessage,
+      runId: "stale-run",
+    });
+    setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration);
+
+    rotateAgentEventLifecycleGeneration();
+    const currentQueueMessage = vi.fn(async () => {});
+    const currentAbort = vi.fn();
+    setActiveEmbeddedRun(
+      "shared-session",
+      createRunHandle({
+        abort: currentAbort,
+        queueMessage: currentQueueMessage,
+        runId: "current-run",
+      }),
+      "agent:main:current",
+      "/tmp/current-session.jsonl",
+    );
+
+    setActiveEmbeddedRun(
+      "shared-session",
+      staleHandle,
+      "agent:main:stale",
+      "/tmp/stale-session.jsonl",
+    );
+    updateActiveEmbeddedRunSessionFile(
+      "shared-session",
+      "/tmp/stale-update.jsonl",
+      priorLifecycleGeneration,
+    );
+
+    await expect(
+      queueEmbeddedAgentMessageWithOutcomeAsync("shared-session", "still live"),
+    ).resolves.toMatchObject({ queued: true, target: "embedded_run" });
+    expect(currentQueueMessage).toHaveBeenCalledOnce();
+    expect(staleQueueMessage).not.toHaveBeenCalled();
+    expect(staleAbort).toHaveBeenCalledWith("restart");
+    expect(currentAbort).not.toHaveBeenCalled();
+    expect(listActiveEmbeddedRunSessionIds()).toContain("shared-session");
+    expect(listActiveEmbeddedRunSessionKeys()).toEqual(["agent:main:current"]);
+    expect(resolveActiveEmbeddedRunHandleSessionId("agent:main:stale")).toBeUndefined();
+    expect(
+      resolveActiveEmbeddedRunHandleSessionIdBySessionFile("/tmp/stale-update.jsonl"),
+    ).toBeUndefined();
+    expect(isEmbeddedAgentRunAbortableForRunId("current-run")).toBe(false);
+    expect(isEmbeddedAgentRunAbortableForRunId("stale-run")).toBe(true);
+  });
+
+  it("rejects a delayed prior-lifecycle registration without a replacement owner", async () => {
+    const priorLifecycleGeneration = getAgentEventLifecycleGeneration();
+    const staleQueueMessage = vi.fn(async () => {});
+    const staleAbort = vi.fn();
+    const staleHandle = createRunHandle({
+      abort: staleAbort,
+      queueMessage: staleQueueMessage,
+      runId: "stale-run",
+    });
+    setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration);
+
+    rotateAgentEventLifecycleGeneration();
+    setActiveEmbeddedRun(
+      "stale-session",
+      staleHandle,
+      "agent:main:stale",
+      "/tmp/stale-session.jsonl",
+    );
+
+    await expect(
+      queueEmbeddedAgentMessageWithOutcomeAsync("stale-session", "should not arrive"),
+    ).resolves.toMatchObject({ queued: false, reason: "no_active_run" });
+    expect(staleQueueMessage).not.toHaveBeenCalled();
+    expect(staleAbort).toHaveBeenCalledWith("restart");
+    expect(listActiveEmbeddedRunSessionIds()).not.toContain("stale-session");
+    expect(listActiveEmbeddedRunSessionKeys()).not.toContain("agent:main:stale");
+    expect(resolveActiveEmbeddedRunHandleSessionId("agent:main:stale")).toBeUndefined();
+    expect(
+      resolveActiveEmbeddedRunHandleSessionIdBySessionFile("/tmp/stale-session.jsonl"),
+    ).toBeUndefined();
+    expect(isEmbeddedAgentRunAbortableForRunId("stale-run")).toBe(true);
+  });
+
+  it("propagates failure to abort a delayed prior-lifecycle registration", () => {
+    const priorLifecycleGeneration = getAgentEventLifecycleGeneration();
+    const staleHandle = createRunHandle({
+      abort: () => {
+        throw new Error("stale abort failed");
+      },
+      queueMessage: vi.fn(async () => {}),
+      runId: "stale-run",
+    });
+    setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration);
+
+    rotateAgentEventLifecycleGeneration();
+
+    expect(() => setActiveEmbeddedRun("stale-session", staleHandle, "agent:main:stale")).toThrow(
+      "stale abort failed",
+    );
+    expect(listActiveEmbeddedRunSessionIds()).not.toContain("stale-session");
+  });
+
+  it("lets a current-lifecycle owner replace a stale session owner", async () => {
+    const staleQueueMessage = vi.fn(async () => {});
+    const staleAbort = vi.fn();
+    const staleHandle = createRunHandle({
+      abort: staleAbort,
+      queueMessage: staleQueueMessage,
+      runId: "stale-run",
+    });
+    setActiveEmbeddedRun(
+      "shared-session",
+      staleHandle,
+      "agent:main:stale",
+      "/tmp/stale-session.jsonl",
+    );
+
+    rotateAgentEventLifecycleGeneration();
+    const currentQueueMessage = vi.fn(async () => {});
+    const currentAbort = vi.fn();
+    setActiveEmbeddedRun(
+      "shared-session",
+      createRunHandle({
+        abort: currentAbort,
+        queueMessage: currentQueueMessage,
+        runId: "current-run",
+      }),
+      "agent:main:current",
+      "/tmp/current-session.jsonl",
+    );
+    clearActiveEmbeddedRun("shared-session", staleHandle, "agent:main:stale");
+
+    await expect(
+      queueEmbeddedAgentMessageWithOutcomeAsync("shared-session", "now current"),
+    ).resolves.toMatchObject({ queued: true, target: "embedded_run" });
+    expect(currentQueueMessage).toHaveBeenCalledOnce();
+    expect(staleQueueMessage).not.toHaveBeenCalled();
+    expect(staleAbort).toHaveBeenCalledOnce();
+    expect(staleAbort).toHaveBeenCalledWith("restart");
+    expect(currentAbort).not.toHaveBeenCalled();
+    expect(listActiveEmbeddedRunSessionIds()).toContain("shared-session");
+    expect(listActiveEmbeddedRunSessionKeys()).toEqual(["agent:main:current"]);
+    expect(resolveActiveEmbeddedRunHandleSessionId("agent:main:stale")).toBeUndefined();
+    expect(
+      resolveActiveEmbeddedRunHandleSessionIdBySessionFile("/tmp/stale-session.jsonl"),
+    ).toBeUndefined();
+  });
+
+  it("preserves a current owner registered synchronously by stale abort", async () => {
+    const currentQueueMessage = vi.fn(async () => {});
+    const currentAbort = vi.fn();
+    const currentHandle = createRunHandle({
+      abort: currentAbort,
+      queueMessage: currentQueueMessage,
+      runId: "current-run",
+    });
+    const staleAbort = vi.fn(() => {
+      setActiveEmbeddedRun(
+        "shared-session",
+        currentHandle,
+        "agent:main:current",
+        "/tmp/current-session.jsonl",
+      );
+    });
+    const staleHandle = createRunHandle({
+      abort: staleAbort,
+      queueMessage: vi.fn(async () => {}),
+      runId: "stale-run",
+    });
+    setActiveEmbeddedRun(
+      "shared-session",
+      staleHandle,
+      "agent:main:stale",
+      "/tmp/stale-session.jsonl",
+    );
+
+    rotateAgentEventLifecycleGeneration();
+
+    await expect(
+      queueEmbeddedAgentMessageWithOutcomeAsync("shared-session", "current survives"),
+    ).resolves.toMatchObject({ queued: true, target: "embedded_run" });
+    expect(staleAbort).toHaveBeenCalledWith("restart");
+    expect(currentAbort).not.toHaveBeenCalled();
+    expect(currentQueueMessage).toHaveBeenCalledOnce();
+    expect(listActiveEmbeddedRunSessionKeys()).toEqual(["agent:main:current"]);
+  });
+
+  it("evicts reply operations created by a prior hot-loaded module instance", async () => {
+    const replyRunsA = await importFreshModule<
+      typeof import("../../auto-reply/reply/reply-run-registry.js")
+    >(import.meta.url, "../../auto-reply/reply/reply-run-registry.js?scope=generation-a");
+    const operation = replyRunsA.createReplyOperation({
+      sessionKey: "agent:main:hot-loaded",
+      sessionId: "hot-loaded-session",
+      resetTriggered: false,
+    });
+    const cancel = vi.fn();
+    operation.setPhase("running");
+    operation.attachBackend({
+      kind: "embedded",
+      cancel,
+      isStreaming: () => true,
+    });
+
+    const replyRunsB = await importFreshModule<
+      typeof import("../../auto-reply/reply/reply-run-registry.js")
+    >(import.meta.url, "../../auto-reply/reply/reply-run-registry.js?scope=generation-b");
+    rotateAgentEventLifecycleGeneration();
+
+    expect(cancel).toHaveBeenCalledWith("restart");
+    expect(replyRunsB.isReplyRunActiveForSessionId("hot-loaded-session")).toBe(false);
+  });
+});

src/agents/embedded-agent-runner/runs.test.ts

@@ -10,6 +10,7 @@ import {
   isReplyRunActiveForSessionId,
 } from "../../auto-reply/reply/reply-run-registry.js";
 import { testing as replyRunTesting } from "../../auto-reply/reply/reply-run-registry.test-support.js";
+import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
 import { setDiagnosticsEnabledForProcess } from "../../infra/diagnostic-events.js";
 import { resetDiagnosticRunActivityForTest } from "../../logging/diagnostic-run-activity.js";
 import { markDiagnosticToolStartedForTest } from "../../logging/diagnostic-run-activity.test-support.js";
@@ -437,6 +438,7 @@ describe("embedded-agent runner run registry", () => {
     updateActiveEmbeddedRunSessionFile(
       "session-file-diagnostics",
       "/tmp/openclaw-run-registry-rotated.jsonl",
+      getAgentEventLifecycleGeneration(),
     );
 
     expect(getDiagnosticSessionState({ sessionId: "session-file-diagnostics" }).sessionFile).toBe(

src/agents/embedded-agent-runner/runs.ts

@@ -17,6 +17,10 @@ import {
   type ReplyOperationPhase,
   waitForReplyRunEndBySessionId,
 } from "../../auto-reply/reply/reply-run-registry.js";
+import {
+  getAgentEventLifecycleGeneration,
+  isAgentEventLifecycleGenerationCurrent,
+} from "../../infra/agent-events.js";
 import {
   getDiagnosticSessionActivitySnapshot,
   markDiagnosticEmbeddedRunEnded,
@@ -33,6 +37,7 @@ import { resolveTimerTimeoutMs } from "../../shared/number-coercion.js";
 import {
   ACTIVE_EMBEDDED_RUNS,
   ACTIVE_EMBEDDED_RUNS_BY_RUN_ID,
+  ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS,
   ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_FILE,
   ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_KEY,
   ACTIVE_EMBEDDED_RUN_SNAPSHOTS,
@@ -42,6 +47,7 @@ import {
   EMBEDDED_RUN_WAITERS,
   getActiveEmbeddedRunCount,
   RETAINED_EMBEDDED_RUN_ABORTABILITY_RUN_IDS,
+  setActiveEmbeddedRunLifecycleGeneration,
   type ActiveEmbeddedRunSnapshot,
   type AbandonedEmbeddedRun,
   type EmbeddedAgentQueueHandle,
@@ -835,6 +841,22 @@ export function setActiveEmbeddedRun(
   sessionKey?: string,
   sessionFile?: string,
 ) {
+  const currentLifecycleGeneration = getAgentEventLifecycleGeneration();
+  const incomingLifecycleGeneration = setActiveEmbeddedRunLifecycleGeneration(
+    handle,
+    currentLifecycleGeneration,
+  );
+  // The immutable handle generation rejects delayed stale registration even
+  // when rotation left no replacement owner in the session slot.
+  if (!isAgentEventLifecycleGenerationCurrent(incomingLifecycleGeneration)) {
+    try {
+      handle.abort("restart");
+    } catch (error) {
+      diag.warn(`stale run registration abort failed: sessionId=${sessionId} err=${String(error)}`);
+      throw error;
+    }
+    return;
+  }
   const previousHandle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
   const wasActive = previousHandle !== undefined;
   if (previousHandle) {
@@ -845,6 +867,7 @@ export function setActiveEmbeddedRun(
   if (handle.runId) {
     ACTIVE_EMBEDDED_RUNS_BY_RUN_ID.set(handle.runId, handle);
   }
+  clearActiveRunSessionKeys(sessionId);
   setActiveRunSessionKey(sessionKey, sessionId);
   clearActiveRunSessionFiles(sessionId);
   setActiveRunSessionFile(sessionFile, sessionId);
@@ -874,8 +897,14 @@ export function updateActiveEmbeddedRunSnapshot(
 export function updateActiveEmbeddedRunSessionFile(
   sessionId: string,
   sessionFile: string | undefined,
+  lifecycleGeneration: string,
 ): void {
-  if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) {
+  const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
+  if (
+    handle === undefined ||
+    ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle) !== lifecycleGeneration ||
+    !isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)
+  ) {
     return;
   }
   clearActiveRunSessionFiles(sessionId);

src/agents/embedded-agent-subscribe.handlers.lifecycle.test.ts

@@ -26,6 +26,9 @@ const BEFORE_AGENT_FINALIZE_EVENT = {
 
 vi.mock("../infra/agent-events.js", () => ({
   emitAgentEvent: emitAgentEventMock,
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 
 function createContext(

src/agents/main-session-restart-recovery.test.ts

@@ -4,6 +4,7 @@ import os from "node:os";
 import path from "node:path";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { GatewayClientRequestError } from "../../packages/gateway-client/src/index.js";
+import { createReplyOperation } from "../auto-reply/reply/reply-run-registry.js";
 import type { InternalSessionEntry as SessionEntry } from "../config/sessions.js";
 import * as sessionAccessor from "../config/sessions/session-accessor.js";
 import {
@@ -19,6 +20,7 @@ import {
   getAgentEventLifecycleGeneration,
   registerAgentRunContext,
   resetAgentEventsForTest,
+  rotateAgentEventLifecycleGeneration,
 } from "../infra/agent-events.js";
 import {
   getActiveGatewayRootWorkCount,
@@ -32,6 +34,14 @@ import {
   runExclusiveSessionLifecycleMutation,
 } from "../sessions/session-lifecycle-admission.js";
 import { createDeferred } from "../test-utils/deferred.js";
+import { setActiveEmbeddedRunLifecycleGeneration } from "./embedded-agent-runner/run-state.js";
+import {
+  clearActiveEmbeddedRun,
+  queueEmbeddedAgentMessageWithOutcomeAsync,
+  resolveActiveEmbeddedRunHandleSessionId,
+  setActiveEmbeddedRun,
+  type EmbeddedAgentQueueHandle,
+} from "./embedded-agent-runner/runs.js";
 import {
   INTERNAL_RUNTIME_CONTEXT_BEGIN,
   INTERNAL_RUNTIME_CONTEXT_END,
@@ -2035,6 +2045,179 @@ describe("main-session-restart-recovery", () => {
     expect(store["agent:main:already-marked"]?.abortedLastRun).toBe(false);
   });
 
+  it.each([
+    ["current owner before delayed stale registration", "current-first"],
+    ["stale owner before current registration", "stale-first"],
+  ] as const)("keeps a live session running with %s", async (_label, registrationOrder) => {
+    const sessionsDir = await makeSessionsDir();
+    const cutoff = Date.now();
+    const sessionKey = "agent:main:generation-race";
+    const sessionId = "generation-race-session";
+    await writeStore(sessionsDir, {
+      [sessionKey]: {
+        sessionId,
+        updatedAt: cutoff - 10_000,
+        status: "running",
+      },
+    });
+
+    const createHandle = (runId: string): EmbeddedAgentQueueHandle => ({
+      kind: "embedded",
+      runId,
+      queueMessage: async () => {},
+      isStreaming: () => true,
+      isCompacting: () => false,
+      abort: () => {},
+    });
+    const priorLifecycleGeneration = getAgentEventLifecycleGeneration();
+    const staleHandle = createHandle("stale-generation-run");
+    setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration);
+    if (registrationOrder === "stale-first") {
+      setActiveEmbeddedRun(sessionId, staleHandle, sessionKey);
+    }
+
+    rotateAgentEventLifecycleGeneration();
+    const currentHandle = createHandle("current-generation-run");
+    setActiveEmbeddedRun(sessionId, currentHandle, sessionKey);
+    if (registrationOrder === "current-first") {
+      setActiveEmbeddedRun(sessionId, staleHandle, sessionKey);
+    }
+
+    try {
+      await expect(
+        recoverStartupOrphanedMainSessions({ stateDir: tmpDir, updatedBeforeMs: cutoff }),
+      ).resolves.toEqual({ marked: 0, recovered: 0, failed: 0, skipped: 0 });
+      expect(callGateway).not.toHaveBeenCalled();
+      expect(
+        loadSessionEntry({
+          sessionKey,
+          storePath: path.join(sessionsDir, "sessions.json"),
+        }),
+      ).toMatchObject({ status: "running" });
+    } finally {
+      clearActiveEmbeddedRun(sessionId, currentHandle, sessionKey);
+      clearActiveEmbeddedRun(sessionId, staleHandle, sessionKey);
+    }
+  });
+
+  it("reconciles only prior-lifecycle running sessions after an in-process restart", async () => {
+    const sessionsDir = await makeSessionsDir();
+    const cutoff = Date.now();
+    const abandonedKey = "agent:main:abandoned-client";
+    const liveKey = "agent:main:live-client";
+    await writeStore(sessionsDir, {
+      [abandonedKey]: {
+        sessionId: "abandoned-session",
+        updatedAt: cutoff - 10_000,
+        status: "running",
+      },
+      [liveKey]: {
+        sessionId: "live-session",
+        updatedAt: cutoff - 10_000,
+        status: "running",
+      },
+    });
+    await writeTranscript(sessionsDir, "abandoned-session", [
+      { role: "system", content: "the client disappeared before the turn became resumable" },
+    ]);
+
+    const createHandle = (
+      runId: string,
+      queueMessage: EmbeddedAgentQueueHandle["queueMessage"] = async () => {},
+      abort: EmbeddedAgentQueueHandle["abort"] = () => {},
+    ): EmbeddedAgentQueueHandle => ({
+      kind: "embedded",
+      runId,
+      queueMessage,
+      isStreaming: () => true,
+      isCompacting: () => false,
+      abort,
+    });
+    const abandonedReply = createReplyOperation({
+      sessionKey: abandonedKey,
+      sessionId: "abandoned-session",
+      resetTriggered: false,
+    });
+    const abandonedReplyQueue = vi.fn(async () => {});
+    const abandonedReplyCancel = vi.fn();
+    abandonedReply.setPhase("running");
+    abandonedReply.attachBackend({
+      kind: "embedded",
+      cancel: abandonedReplyCancel,
+      isStreaming: () => true,
+      queueMessage: abandonedReplyQueue,
+    });
+    const abandonedEmbeddedQueue = vi.fn(async () => {});
+    const abandonedEmbeddedAbort = vi.fn();
+    const abandonedHandle = createHandle(
+      "abandoned-run",
+      abandonedEmbeddedQueue,
+      abandonedEmbeddedAbort,
+    );
+    setActiveEmbeddedRun("abandoned-session", abandonedHandle, abandonedKey);
+
+    const firstRecovery = recoverStartupOrphanedMainSessions({
+      stateDir: tmpDir,
+      updatedBeforeMs: cutoff,
+    });
+    // Advance ownership while the async store discovery above is pending. The
+    // older scan must drop the stale owner without overlooking this new live one.
+    rotateAgentEventLifecycleGeneration();
+    setActiveEmbeddedRun("abandoned-session", abandonedHandle, abandonedKey);
+
+    await expect(
+      queueEmbeddedAgentMessageWithOutcomeAsync("abandoned-session", "do not route stale"),
+    ).resolves.toMatchObject({ queued: false, reason: "no_active_run" });
+    expect(abandonedEmbeddedQueue).not.toHaveBeenCalled();
+    expect(abandonedEmbeddedAbort).toHaveBeenCalledWith("restart");
+    expect(abandonedReplyQueue).not.toHaveBeenCalled();
+    expect(abandonedReplyCancel).toHaveBeenCalledWith("restart");
+    expect(resolveActiveEmbeddedRunHandleSessionId(abandonedKey)).toBeUndefined();
+
+    const liveReply = createReplyOperation({
+      sessionKey: liveKey,
+      sessionId: "live-session",
+      resetTriggered: false,
+    });
+    const liveAbort = vi.fn();
+    const liveHandle = createHandle("live-run", undefined, liveAbort);
+    setActiveEmbeddedRun("live-session", liveHandle, liveKey);
+    try {
+      const first = await firstRecovery;
+
+      expect(first).toEqual({ marked: 1, recovered: 0, failed: 1, skipped: 0 });
+      expect(callGateway).not.toHaveBeenCalled();
+      expect(
+        loadSessionEntry({
+          sessionKey: abandonedKey,
+          storePath: path.join(sessionsDir, "sessions.json"),
+        }),
+      ).toMatchObject({
+        status: "failed",
+        abortedLastRun: true,
+      });
+      expect(
+        loadSessionEntry({
+          sessionKey: liveKey,
+          storePath: path.join(sessionsDir, "sessions.json"),
+        }),
+      ).toMatchObject({
+        status: "running",
+      });
+      expect(liveAbort).not.toHaveBeenCalled();
+      expect(liveReply.abortSignal.aborted).toBe(false);
+
+      await expect(
+        recoverStartupOrphanedMainSessions({ stateDir: tmpDir, updatedBeforeMs: cutoff }),
+      ).resolves.toEqual({ marked: 0, recovered: 0, failed: 0, skipped: 0 });
+    } finally {
+      clearActiveEmbeddedRun("abandoned-session", abandonedHandle, abandonedKey);
+      clearActiveEmbeddedRun("live-session", liveHandle, liveKey);
+      abandonedReply.complete();
+      liveReply.complete();
+    }
+  });
+
   it("recovers only the configured store for duplicate startup-orphaned session keys", async () => {
     const cutoff = Date.now();
     const defaultSessionsDir = await makeSessionsDir();

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

@@ -364,6 +364,9 @@ export async function markStartupOrphanedMainSessionsForRecovery(params: {
       ? undefined
       : normalizeStringSet(params.activeSessionKeys);
   const updatedBeforeMs = normalizeFiniteTimestamp(params.updatedBeforeMs);
+  // Lifecycle rotation synchronously evicts stale owners, so this same registry
+  // view drives both operational routing and recovery suppression. Re-read it at
+  // each check so a newer owner can still fence an older async recovery scan.
   const resolveActiveSessionIds = () =>
     providedActiveSessionIds ?? normalizeStringSet(listActiveEmbeddedRunSessionIds());
   const resolveActiveSessionKeys = () =>

src/agents/subagent-registry.announce-loop-guard.test.ts

@@ -54,7 +54,10 @@ vi.mock("../gateway/call.js", () => ({
 }));
 
 vi.mock("../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
   onAgentEvent: mocks.onAgentEvent,
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 
 vi.mock("./subagent-registry.store.sqlite.js", () => ({

src/agents/subagent-registry.archive.e2e.test.ts

@@ -57,8 +57,11 @@ vi.mock("../tasks/task-status-access.js", () => ({
 }));
 
 vi.mock("../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
   getAgentRunContext: vi.fn(() => undefined),
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
   onAgentEvent: vi.fn((_handler: unknown) => noop),
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 
 vi.mock("../config/config.js", async () => {

src/agents/subagent-registry.mocks.shared.ts

@@ -21,5 +21,8 @@ vi.mock("../gateway/call.js", () => ({
 }));
 
 vi.mock("../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
   onAgentEvent: sharedMocks.onAgentEvent,
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));

src/agents/subagent-registry.steer-restart.test.ts

@@ -50,6 +50,9 @@ vi.mock("../gateway/call.js", () => ({
 }));
 
 vi.mock("../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
   onAgentEvent: vi.fn((handler: typeof lifecycleHandler) => {
     lifecycleHandler = handler;
     return noop;

src/agents/subagent-registry.test.ts

@@ -211,8 +211,11 @@ vi.mock("../gateway/call.js", () => ({
 }));
 
 vi.mock("../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
   getAgentRunContext: mocks.getAgentRunContext,
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
   onAgentEvent: mocks.onAgentEvent,
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 
 vi.mock("../config/config.js", () => {

src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts

@@ -537,7 +537,10 @@ vi.mock("../../bindings/records.js", () => ({
 vi.mock("../../infra/agent-events.js", () => ({
   emitAgentAuditEvent: (params: unknown) => agentEventMocks.emitAgentAuditEvent(params),
   emitAgentEvent: (params: unknown) => agentEventMocks.emitAgentEvent(params),
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
   onAgentEvent: (listener: unknown) => agentEventMocks.onAgentEvent(listener),
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 vi.mock("../../plugins/conversation-binding.js", () => ({
   buildPluginBindingDeclinedText: () => "Plugin binding request was declined.",

src/auto-reply/reply/reply-run-registry.ts

@@ -5,6 +5,11 @@ import {
   isAgentRunRestartAbortReason,
 } from "../../agents/run-termination.js";
 import { createAbortError } from "../../infra/abort-signal.js";
+import {
+  getAgentEventLifecycleGeneration,
+  isAgentEventLifecycleGenerationCurrent,
+  registerAgentEventLifecycleRotationHandler,
+} from "../../infra/agent-events.js";
 import type { ImageContent } from "../../llm/types.js";
 import {
   getDiagnosticSessionActivitySnapshot,
@@ -123,6 +128,8 @@ type ReplyOperationResult =
 export type ReplyOperation = {
   readonly key: ReplyRunKey;
   readonly sessionId: string;
+  /** Gateway lifecycle that admitted this process-local owner. */
+  readonly lifecycleGeneration?: string;
   readonly routeThreadId?: string | number;
   readonly abortSignal: AbortSignal;
   readonly resetTriggered: boolean;
@@ -234,6 +241,7 @@ type ReplyRunState = {
   waitKeysBySessionId: Map<string, string>;
   waitersByKey: Map<string, Set<ReplyRunWaiter>>;
   followupAdmissionBarriersByKey: Map<string, ReplyRunFollowupAdmissionBarrier>;
+  evictOperationByOperation?: WeakMap<ReplyOperation, () => void>;
 };
 
 const REPLY_RUN_STATE_KEY = Symbol.for("openclaw.replyRunRegistry");
@@ -245,8 +253,12 @@ const replyRunState = resolveGlobalSingleton<ReplyRunState>(REPLY_RUN_STATE_KEY,
   waitKeysBySessionId: new Map<string, string>(),
   waitersByKey: new Map<string, Set<ReplyRunWaiter>>(),
   followupAdmissionBarriersByKey: new Map<string, ReplyRunFollowupAdmissionBarrier>(),
+  evictOperationByOperation: new WeakMap<ReplyOperation, () => void>(),
 }));
 replyRunState.followupAdmissionBarriersByKey ??= new Map();
+const evictReplyOperationByOperation =
+  replyRunState.evictOperationByOperation ??
+  (replyRunState.evictOperationByOperation = new WeakMap<ReplyOperation, () => void>());
 
 export const REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS = 15_000;
 // Terminal results must release the lane even if the owner never resumes.
@@ -490,7 +502,20 @@ function updateFollowupAdmissionSessionId(sessionKey: string, sessionId: string)
   }
 }
 
-function clearReplyRunState(params: { sessionKey: string; sessionId: string }): void {
+function clearReplyRunState(params: {
+  sessionKey: string;
+  sessionId: string;
+  operation: ReplyOperation;
+}): void {
+  if (replyRunState.activeRunsByKey.get(params.sessionKey) !== params.operation) {
+    if (
+      replyRunState.activeKeysBySessionId.get(params.sessionId) === params.sessionKey &&
+      replyRunState.activeSessionIdsByKey.get(params.sessionKey) !== params.sessionId
+    ) {
+      replyRunState.activeKeysBySessionId.delete(params.sessionId);
+    }
+    return;
+  }
   replyRunState.activeRunsByKey.delete(params.sessionKey);
   replyRunState.activeSessionIdsByKey.delete(params.sessionKey);
   if (replyRunState.activeKeysBySessionId.get(params.sessionId) === params.sessionKey) {
@@ -550,6 +575,7 @@ export function createReplyOperation(params: {
   let terminalRecovery = false;
   let acceptedSteeredInboundAudio = false;
   const startedAtMs = Date.now();
+  const lifecycleGeneration = getAgentEventLifecycleGeneration();
   let lastActivityAtMs = startedAtMs;
   const upstreamAbortSignal = params.upstreamAbortSignal;
   let upstreamAbortHandler: (() => void) | undefined;
@@ -580,6 +606,7 @@ export function createReplyOperation(params: {
     terminalSettleTimer.clear();
     finalizationLease.clear();
     expireReplyOperationByOperation.delete(operation);
+    evictReplyOperationByOperation.delete(operation);
     detachUpstreamAbort();
     const registeredBarrier = afterClearBarrier
       ? registerFollowupAdmissionBarrier(
@@ -598,6 +625,7 @@ export function createReplyOperation(params: {
     clearReplyRunState({
       sessionKey: currentSessionKey,
       sessionId: currentSessionId,
+      operation,
     });
     if (!registeredBarrier) {
       flushReplyOperationAfterClear(operation, currentSessionId);
@@ -642,6 +670,7 @@ export function createReplyOperation(params: {
     get sessionId() {
       return currentSessionId;
     },
+    lifecycleGeneration,
     get routeThreadId() {
       return params.routeThreadId;
     },
@@ -939,6 +968,33 @@ export function createReplyOperation(params: {
     },
   });
 
+  evictReplyOperationByOperation.set(operation, () => {
+    if (stateCleared) {
+      return;
+    }
+    if (!result) {
+      setResult({ kind: "aborted", code: "aborted_for_restart" });
+      phase = "aborted";
+    }
+    abortInternally(createAgentRunRestartAbortError());
+    let cancelError: unknown;
+    let cancelFailed = false;
+    try {
+      getAttachedBackend(operation)?.cancel("restart");
+    } catch (error) {
+      cancelFailed = true;
+      cancelError = error;
+      diag.warn(
+        `reply run lifecycle eviction cancel failed: sessionKey=${currentSessionKey} error=${String(error)}`,
+      );
+    } finally {
+      clearState();
+    }
+    if (cancelFailed) {
+      throw cancelError;
+    }
+  });
+
   replyRunState.activeRunsByKey.set(sessionKey, operation);
   replyRunState.activeSessionIdsByKey.set(sessionKey, currentSessionId);
   replyRunState.activeKeysBySessionId.set(currentSessionId, sessionKey);
@@ -1295,6 +1351,67 @@ export function listActiveReplyRunSessionKeys(): string[] {
   return [...replyRunState.activeSessionIdsByKey.keys()];
 }
 
+function evictPriorLifecycleReplyRuns(): void {
+  const errors: unknown[] = [];
+  for (const operation of replyRunState.activeRunsByKey.values()) {
+    if (
+      operation.lifecycleGeneration &&
+      isAgentEventLifecycleGenerationCurrent(operation.lifecycleGeneration)
+    ) {
+      continue;
+    }
+    const evict = evictReplyOperationByOperation.get(operation);
+    if (evict) {
+      try {
+        evict();
+      } catch (error) {
+        errors.push(error);
+        try {
+          clearReplyRunState({
+            sessionKey: operation.key,
+            sessionId: operation.sessionId,
+            operation,
+          });
+        } catch (clearError) {
+          errors.push(clearError);
+        }
+      }
+      continue;
+    }
+    // Pre-generation hot-loaded operations have no retained callback, but their
+    // public method still closes over the module instance that owns the backend.
+    try {
+      if (!operation.abortForRestart()) {
+        errors.push(new Error(`Stale reply operation was not abortable: ${operation.key}`));
+      }
+    } catch (error) {
+      errors.push(error);
+    }
+    // Admission stays occupied until the old closure clears it. If abort
+    // synchronously clears and replaces the slot, its captured stateCleared
+    // makes this completion idempotent instead of erasing the replacement.
+    try {
+      operation.complete();
+    } catch (error) {
+      errors.push(error);
+    }
+    try {
+      clearReplyRunState({
+        sessionKey: operation.key,
+        sessionId: operation.sessionId,
+        operation,
+      });
+    } catch (error) {
+      errors.push(error);
+    }
+  }
+  if (errors.length > 0) {
+    throw new AggregateError(errors, "Failed to abort stale reply runs");
+  }
+}
+
+registerAgentEventLifecycleRotationHandler("reply-runs", evictPriorLifecycleReplyRuns);
+
 const replyRunRegistryTestApi = {
   resetReplyRunRegistry(): void {
     for (const [sessionKey, sessionId] of replyRunState.activeSessionIdsByKey) {

src/cli/gateway-cli/run-loop.test.ts

@@ -211,6 +211,9 @@ vi.mock("../../tasks/runtime-internal.js", () => ({
 }));
 
 vi.mock("../../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
   rotateAgentEventLifecycleGeneration: () => rotateAgentEventLifecycleGeneration(),
 }));
 

src/commands/agent.acp.test.ts

@@ -26,10 +26,14 @@ const agentEventMocks = vi.hoisted(() => {
       }
     }),
     getAgentEventLifecycleGeneration: vi.fn(() => "test-generation"),
+    isAgentEventLifecycleGenerationCurrent: vi.fn(
+      (generation: string) => generation === "test-generation",
+    ),
     onAgentEvent: vi.fn((handler: (event: AgentEvent) => void) => {
       handlers.add(handler);
       return () => handlers.delete(handler);
     }),
+    registerAgentEventLifecycleRotationHandler: vi.fn(),
     registerAgentRunContext: vi.fn(),
     withAgentRunLifecycleGeneration: vi.fn((_generation: string, run: () => unknown) => run()),
   };

src/gateway/server-methods/agent.test-harness.ts

@@ -221,6 +221,9 @@ vi.mock("../../infra/agent-events.js", () => ({
   getAgentEventLifecycleGeneration: () => mocks.lifecycleGeneration,
   getAgentRunContext: vi.fn(() => undefined),
   hasProjectedAgentRunForSession: vi.fn(() => false),
+  isAgentEventLifecycleGenerationCurrent: (generation: string) =>
+    generation === mocks.lifecycleGeneration,
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
   registerAgentRunContext: mocks.registerAgentRunContext,
   onAgentEvent: vi.fn(),
 }));

src/infra/agent-events.ts

@@ -167,6 +167,7 @@ type AgentEventState = {
     }
   >;
   lifecycleGeneration: string;
+  lifecycleRotationHandlers?: Map<string, (lifecycleGeneration: string) => void>;
 };
 
 const AGENT_EVENT_STATE_KEY = Symbol.for("openclaw.agentEvents.state");
@@ -223,9 +224,25 @@ export function getAgentEventLifecycleGeneration(): string {
   return getAgentEventState().lifecycleGeneration;
 }
 
+export function isAgentEventLifecycleGenerationCurrent(lifecycleGeneration: string): boolean {
+  return lifecycleGeneration === getAgentEventState().lifecycleGeneration;
+}
+
+/** Registers process-local state cleanup at the gateway lifecycle boundary. */
+export function registerAgentEventLifecycleRotationHandler(
+  key: string,
+  handler: (lifecycleGeneration: string) => void,
+): void {
+  const state = getAgentEventState();
+  const handlers =
+    state.lifecycleRotationHandlers ??
+    (state.lifecycleRotationHandlers = new Map<string, (lifecycleGeneration: string) => void>());
+  handlers.set(key, handler);
+}
+
 /** Rejects work that no longer belongs to the active gateway lifecycle. */
 export function assertAgentRunLifecycleGenerationCurrent(lifecycleGeneration: string): void {
-  if (lifecycleGeneration === getAgentEventState().lifecycleGeneration) {
+  if (isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)) {
     return;
   }
   throw createAbortError("Agent run belongs to a stale gateway lifecycle");
@@ -244,6 +261,18 @@ export function captureAgentRunLifecycleGeneration(runId: string): string {
 export function rotateAgentEventLifecycleGeneration(): string {
   const state = getAgentEventState();
   state.lifecycleGeneration = randomUUID();
+  // Rotation is the liveness choke point: after it returns, no prior-generation
+  // owner is operationally reachable. Recovery and runtime consumers therefore
+  // agree that only current-generation owners can drive or receive work.
+  const errors: unknown[] = [];
+  notifyListeners(
+    state.lifecycleRotationHandlers?.values() ?? [],
+    state.lifecycleGeneration,
+    (error) => errors.push(error),
+  );
+  if (errors.length > 0) {
+    throw new AggregateError(errors, "Failed to retire stale agent lifecycle owners");
+  }
   return state.lifecycleGeneration;
 }
 

src/plugins/wired-hooks-after-tool-call.e2e.test.ts

@@ -22,6 +22,9 @@ vi.mock("../infra/agent-events.js", () => ({
   emitAgentCommandOutputEvent: vi.fn(),
   emitAgentItemEvent: vi.fn(),
   emitAgentEvent: vi.fn(),
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 
 function createToolHandlerCtx(params: {

src/plugins/wired-hooks-compaction.test.ts

@@ -19,6 +19,9 @@ vi.mock("../plugins/hook-runner-global.js", () => ({
 
 vi.mock("../infra/agent-events.js", () => ({
   emitAgentEvent: hookMocks.emitAgentEvent,
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 
 import {

src/tasks/cron-run-continuation-cleanup.test.ts

@@ -17,6 +17,9 @@ vi.mock("../config/sessions/session-accessor.js", () => ({
 }));
 vi.mock("../infra/agent-events.js", () => ({
   getAgentEventLifecycleGeneration: () => "current-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) =>
+    generation === "current-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
 }));
 vi.mock("../infra/session-delivery-queue.js", () => ({
   loadPendingSessionDeliveries: mocks.loadPendingSessionDeliveries,

src/tui/embedded-backend.test.ts

@@ -97,6 +97,9 @@ vi.mock("../agents/btw.js", () => ({
 }));
 
 vi.mock("../infra/agent-events.js", () => ({
+  getAgentEventLifecycleGeneration: () => "test-generation",
+  isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation",
+  registerAgentEventLifecycleRotationHandler: vi.fn(),
   onAgentEvent: (listener: (evt: unknown) => void) => {
     registeredListener = listener;
     return () => {