fix(auto-reply): serialize reply session initialization

obviyus · obviyus · commit d2da8c79d9b8 · 2026-06-25T13:10:35.000-07:00
diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts
@@ -10,6 +10,7 @@ import {
 import * as bootstrapCache from "../../agents/bootstrap-cache.js";
 import type { OpenClawConfig } from "../../config/config.js";
 import type { SessionEntry } from "../../config/sessions.js";
+import { runExclusiveSessionStoreWrite } from "../../config/sessions/store-writer.js";
 import { formatZonedTimestamp } from "../../infra/format-time/format-datetime.ts";
 import {
   testing as sessionBindingTesting,
@@ -468,6 +469,49 @@ afterEach(async () => {
   resetSystemEventsForTest();
   await sessionMcpTesting.resetSessionMcpRuntimeManager();
 });
+describe("initSessionState guarded initialization", () => {
+  it("serializes concurrent initializers before reading the guarded snapshot", async () => {
+    const storePath = await createStorePath("openclaw-session-init-race-");
+    const sessionKey = "agent:main:telegram:chat:42";
+    await writeSessionStoreFast(storePath, {
+      [sessionKey]: {
+        sessionId: "existing-session",
+        updatedAt: 100,
+      },
+    });
+    const cfg = { session: { store: storePath } } as OpenClawConfig;
+    let releaseWriter = () => {};
+    const writerReleased = new Promise<void>((resolve) => {
+      releaseWriter = resolve;
+    });
+    let markWriterStarted = () => {};
+    const writerStarted = new Promise<void>((resolve) => {
+      markWriterStarted = resolve;
+    });
+    const heldWriter = runExclusiveSessionStoreWrite(storePath, async () => {
+      markWriterStarted();
+      await writerReleased;
+    });
+    await writerStarted;
+
+    const turns = Array.from({ length: 8 }, (_, index) =>
+      initSessionState({
+        ctx: {
+          Body: `turn ${index}`,
+          SessionKey: sessionKey,
+        },
+        cfg,
+        commandAuthorized: true,
+      }),
+    );
+
+    releaseWriter();
+    await heldWriter;
+
+    await expect(Promise.all(turns)).resolves.toHaveLength(8);
+  });
+});
+
 describe("initSessionState thread forking", () => {
   it("forks a new session from the parent session file", async () => {
     const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
diff --git a/src/auto-reply/reply/session.ts b/src/auto-reply/reply/session.ts
@@ -35,6 +35,7 @@ import {
 } from "../../config/sessions/session-accessor.js";
 import { resolveSessionKey } from "../../config/sessions/session-key.js";
 import { resolveMaintenanceConfigFromInput } from "../../config/sessions/store-maintenance.js";
+import { runExclusiveSessionStoreWrite } from "../../config/sessions/store-writer.js";
 import { parseSessionThreadInfoFast } from "../../config/sessions/thread-info.js";
 import {
   DEFAULT_RESET_TRIGGERS,
@@ -186,6 +187,14 @@ export type InitSessionStateParams = {
   resumeRequestedSession?: boolean;
 };
 
+type InitSessionStateAttemptContext = {
+  agentId: string;
+  conversationBindingContext: ReturnType<typeof resolveSessionConversationBindingContext>;
+  isSystemEvent: boolean;
+  sessionCtxForState: MsgContext;
+  storePath: string;
+};
+
 function resolveSessionConversationBindingContext(
   cfg: OpenClawConfig,
   ctx: MsgContext,
@@ -244,32 +253,19 @@ function resolveBoundConversationSessionKey(params: {
   return binding.targetSessionKey;
 }
 
-/** Initializes or reuses the reply session state for one inbound turn. */
-export async function initSessionState(params: InitSessionStateParams): Promise<SessionInitResult> {
-  return await initSessionStateAttempt(params, false);
-}
-
-async function initSessionStateAttempt(
+function resolveInitSessionStateAttemptContext(
   params: InitSessionStateParams,
-  staleSnapshotRetried: boolean,
-): Promise<SessionInitResult> {
-  const { ctx, cfg, commandAuthorized } = params;
-  // Heartbeat, cron-event, and exec-event runs should NEVER trigger session
-  // resets or conversation binding retargeting. These are automated system
-  // events, not user interactions that should affect session continuity.
-  // See #58409 for details on silent session reset bug.
+): InitSessionStateAttemptContext {
+  const { cfg, ctx } = params;
+  // Automated system events must not reset sessions or retarget conversation bindings.
   const isSystemEvent =
     ctx.Provider === "heartbeat" || ctx.Provider === "cron-event" || ctx.Provider === "exec-event";
   const conversationBindingContext = isSystemEvent
     ? null
     : resolveSessionConversationBindingContext(cfg, ctx);
-  // Native slash commands (Telegram/Discord/Slack) are delivered on a separate
-  // "slash session" key, but should mutate the target chat session.
+  // Slash/menu commands may arrive on a transport session while targeting the chat session.
+  // Prefer explicit command target before binding lookup so command mutations land there.
   const commandTargetSessionKey = resolveCommandTurnTargetSessionKey(ctx);
-  // Native slash/menu commands can arrive on a transport-specific "slash session"
-  // while explicitly targeting an existing chat session. Honor that explicit target
-  // before any binding lookup so command-side mutations land on the intended session.
-  // Priority: commandTargetSessionKey > boundConversation > route.
   const targetSessionKey =
     commandTargetSessionKey ??
     resolveBoundConversationSessionKey({
@@ -281,20 +277,54 @@ async function initSessionStateAttempt(
     targetSessionKey && targetSessionKey !== ctx.SessionKey
       ? { ...ctx, SessionKey: targetSessionKey }
       : ctx;
-  const sessionCfg = cfg.session;
-  const maintenanceConfig = resolveMaintenanceConfigFromInput(sessionCfg?.maintenance);
-  const mainKey = normalizeMainKey(sessionCfg?.mainKey);
   const agentId = resolveSessionAgentId({
     sessionKey: sessionCtxForState.SessionKey,
     config: cfg,
     fallbackAgentId: sessionCtxForState.AgentId,
   });
+  return {
+    agentId,
+    conversationBindingContext,
+    isSystemEvent,
+    sessionCtxForState,
+    storePath: resolveStorePath(cfg.session?.store, { agentId }),
+  };
+}
+
+/** Initializes or reuses the reply session state for one inbound turn. */
+export async function initSessionState(params: InitSessionStateParams): Promise<SessionInitResult> {
+  return await initSessionStateAttempt(params, false);
+}
+
+async function initSessionStateAttempt(
+  params: InitSessionStateParams,
+  staleSnapshotRetried: boolean,
+): Promise<SessionInitResult> {
+  const attemptContext = resolveInitSessionStateAttemptContext(params);
+  // Guarded revision checks only serialize correctly when the snapshot and
+  // commit share the same writer lane.
+  return await runExclusiveSessionStoreWrite(
+    attemptContext.storePath,
+    async () => await initSessionStateAttemptLocked(params, attemptContext, staleSnapshotRetried),
+  );
+}
+
+async function initSessionStateAttemptLocked(
+  params: InitSessionStateParams,
+  attemptContext: InitSessionStateAttemptContext,
+  staleSnapshotRetried: boolean,
+): Promise<SessionInitResult> {
+  const { ctx, cfg, commandAuthorized } = params;
+  const { agentId, conversationBindingContext, isSystemEvent, sessionCtxForState, storePath } =
+    attemptContext;
+  const sessionCfg = cfg.session;
+  const maintenanceConfig = resolveMaintenanceConfigFromInput(sessionCfg?.maintenance);
+  const mainKey = normalizeMainKey(sessionCfg?.mainKey);
   const groupResolution = resolveGroupSessionKey(sessionCtxForState) ?? undefined;
   const resetTriggers = sessionCfg?.resetTriggers?.length
     ? sessionCfg.resetTriggers
     : DEFAULT_RESET_TRIGGERS;
   const sessionScope = sessionCfg?.scope ?? "per-sender";
-  const storePath = resolveStorePath(sessionCfg?.store, { agentId });
   const ingressTimingEnabled = process.env.OPENCLAW_DEBUG_INGRESS_TIMING === "1";
 
   let sessionEntry: SessionEntry;
@@ -858,7 +888,7 @@ async function initSessionStateAttempt(
   });
   if (!committed.ok) {
     if (!staleSnapshotRetried) {
-      return await initSessionStateAttempt(params, true);
+      return await initSessionStateAttemptLocked(params, attemptContext, true);
     }
     throw new Error(`reply session initialization conflicted for ${sessionKey}`);
   }
diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts
@@ -1776,6 +1776,7 @@ export async function commitReplySessionInitialization(params: {
       activeSessionKey: params.activeSessionKey,
       maintenanceConfig: params.maintenanceConfig,
       onWarn: params.onMaintenanceWarning,
+      reentrant: true,
       skipSaveWhenResult: (result) => !result.ok,
     },
   );
diff --git a/src/config/sessions/store-writer.test.ts b/src/config/sessions/store-writer.test.ts
@@ -36,6 +36,76 @@ describe("session store writer", () => {
     expect(getSessionStoreWriterQueueSizeForTest()).toBe(0);
   });
 
+  it("runs nested writes for the active store without requeueing behind itself", async () => {
+    const storePath = "/tmp/openclaw-store.json";
+    const order: string[] = [];
+
+    const result = await runExclusiveSessionStoreWrite(storePath, async () => {
+      order.push("outer:start");
+      const nested = await runExclusiveSessionStoreWrite(
+        storePath,
+        async () => {
+          order.push("inner");
+          return "nested-result";
+        },
+        { reentrant: true },
+      );
+      order.push("outer:end");
+      return nested;
+    });
+
+    expect(result).toBe("nested-result");
+    expect(order).toEqual(["outer:start", "inner", "outer:end"]);
+    expect(getSessionStoreWriterQueueSizeForTest()).toBe(0);
+  });
+
+  it("does not leak active writer state to async children after the writer returns", async () => {
+    const storePath = "/tmp/openclaw-store.json";
+    const order: string[] = [];
+    let releaseChild = () => {};
+    const childReleased = new Promise<void>((resolve) => {
+      releaseChild = resolve;
+    });
+    let child: Promise<string> = Promise.resolve("not-started");
+
+    await runExclusiveSessionStoreWrite(storePath, async () => {
+      child = (async () => {
+        await childReleased;
+        return await runExclusiveSessionStoreWrite(storePath, async () => {
+          order.push("child");
+          return "child-result";
+        });
+      })();
+    });
+
+    let releaseBlocker = () => {};
+    const blockerReleased = new Promise<void>((resolve) => {
+      releaseBlocker = resolve;
+    });
+    let markBlockerStarted = () => {};
+    const blockerStarted = new Promise<void>((resolve) => {
+      markBlockerStarted = resolve;
+    });
+    const blocker = runExclusiveSessionStoreWrite(storePath, async () => {
+      order.push("blocker:start");
+      markBlockerStarted();
+      await blockerReleased;
+      order.push("blocker:end");
+    });
+    await blockerStarted;
+
+    releaseChild();
+    await Promise.resolve();
+    expect(order).toEqual(["blocker:start"]);
+
+    releaseBlocker();
+    await Promise.all([blocker, child]);
+
+    expect(order).toEqual(["blocker:start", "blocker:end", "child"]);
+    expect(await child).toBe("child-result");
+    expect(getSessionStoreWriterQueueSizeForTest()).toBe(0);
+  });
+
   it("rejects empty store paths before enqueuing work", async () => {
     await expect(runExclusiveSessionStoreWrite("", async () => undefined)).rejects.toThrow(
       /storePath must be a non-empty string/,
diff --git a/src/config/sessions/store-writer.ts b/src/config/sessions/store-writer.ts
@@ -2,14 +2,20 @@
 import { runQueuedStoreWrite } from "../../shared/store-writer-queue.js";
 import { WRITER_QUEUES } from "./store-writer-state.js";
 
+export type RunExclusiveSessionStoreWriteOptions = {
+  reentrant?: boolean;
+};
+
 export async function runExclusiveSessionStoreWrite<T>(
   storePath: string,
   fn: () => Promise<T>,
+  opts: RunExclusiveSessionStoreWriteOptions = {},
 ): Promise<T> {
   return await runQueuedStoreWrite({
     queues: WRITER_QUEUES,
     storePath,
     label: "runExclusiveSessionStoreWrite",
     fn,
+    reentrant: opts.reentrant,
   });
 }
diff --git a/src/config/sessions/store.ts b/src/config/sessions/store.ts
@@ -204,6 +204,8 @@ type SaveSessionStoreOptions = {
 };
 
 type UpdateSessionStoreOptions<T> = SaveSessionStoreOptions & {
+  /** Allow a nested mutation only when the caller already owns this store writer lane. */
+  reentrant?: boolean;
   /**
    * Specialized callers can prove their mutator made no changes through its result.
    * When true, the writer-owned object cache is restored and sessions.json is untouched.
@@ -1019,19 +1021,23 @@ export async function updateSessionStore<T>(
   mutator: (store: Record<string, SessionEntry>) => Promise<T> | T,
   opts?: UpdateSessionStoreOptions<T>,
 ): Promise<T> {
-  return await runExclusiveSessionStoreWrite(storePath, async () => {
-    const store = loadMutableSessionStoreForWriter(storePath);
-    const result = await mutator(store);
-    if (opts?.skipSaveWhenResult?.(result)) {
-      restoreUnchangedSessionStoreCache(storePath, store);
+  return await runExclusiveSessionStoreWrite(
+    storePath,
+    async () => {
+      const store = loadMutableSessionStoreForWriter(storePath);
+      const result = await mutator(store);
+      if (opts?.skipSaveWhenResult?.(result)) {
+        restoreUnchangedSessionStoreCache(storePath, store);
+        return result;
+      }
+      await saveSessionStoreUnlocked(storePath, store, {
+        ...opts,
+        singleEntryPersistence: opts?.resolveSingleEntryPersistence?.(result) ?? undefined,
+      });
       return result;
-    }
-    await saveSessionStoreUnlocked(storePath, store, {
-      ...opts,
-      singleEntryPersistence: opts?.resolveSingleEntryPersistence?.(result) ?? undefined,
-    });
-    return result;
-  });
+    },
+    { reentrant: opts?.reentrant },
+  );
 }
 
 function cloneSessionEntryProjectionSnapshot(
diff --git a/src/shared/store-writer-queue.ts b/src/shared/store-writer-queue.ts
