From a38c075762dba4485fcd32dc4be5077d53afd446 Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Thu, 25 Jun 2026 12:11:06 -0700 Subject: [PATCH] fix(auto-reply): serialize reply session initialization (cherry picked from commit d2da8c79d9b8199915ba280fe88a1be5f8d3f6b8) --- src/auto-reply/reply/session.test.ts | 44 +++++++++++++ src/auto-reply/reply/session.ts | 78 ++++++++++++++++-------- src/config/sessions/session-accessor.ts | 1 + src/config/sessions/store-writer.test.ts | 70 +++++++++++++++++++++ src/config/sessions/store-writer.ts | 6 ++ src/config/sessions/store.ts | 30 +++++---- src/shared/store-writer-queue.ts | 43 ++++++++++++- 7 files changed, 235 insertions(+), 37 deletions(-) diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts index 0e5824c052ea..d7e4a8f24c6e 100644 --- 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((resolve) => { + releaseWriter = resolve; + }); + let markWriterStarted = () => {}; + const writerStarted = new Promise((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 index c075867fa0bb..58cbd8701cd2 100644 --- 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; + 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 { - return await initSessionStateAttempt(params, false); -} - -async function initSessionStateAttempt( +function resolveInitSessionStateAttemptContext( params: InitSessionStateParams, - staleSnapshotRetried: boolean, -): Promise { - 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 { + return await initSessionStateAttempt(params, false); +} + +async function initSessionStateAttempt( + params: InitSessionStateParams, + staleSnapshotRetried: boolean, +): Promise { + 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 { + 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 index 444208e743cd..41b661314c5a 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -1465,6 +1465,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 index 1cd370f69328..4590e5945c22 100644 --- 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((resolve) => { + releaseChild = resolve; + }); + let child: Promise = 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((resolve) => { + releaseBlocker = resolve; + }); + let markBlockerStarted = () => {}; + const blockerStarted = new Promise((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 index a015566821f2..605b686de694 100644 --- 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( storePath: string, fn: () => Promise, + opts: RunExclusiveSessionStoreWriteOptions = {}, ): Promise { 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 index 8ec5459c0d71..791782151e57 100644 --- a/src/config/sessions/store.ts +++ b/src/config/sessions/store.ts @@ -193,6 +193,8 @@ type SaveSessionStoreOptions = { }; type UpdateSessionStoreOptions = 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. @@ -1008,19 +1010,23 @@ export async function updateSessionStore( mutator: (store: Record) => Promise | T, opts?: UpdateSessionStoreOptions, ): Promise { - 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 index 3e1cb26dd00b..3cf2a675a6dc 100644 --- a/src/shared/store-writer-queue.ts +++ b/src/shared/store-writer-queue.ts @@ -1,3 +1,5 @@ +import { AsyncLocalStorage } from "node:async_hooks"; + /** Pending exclusive store write plus the promise hooks for its caller. */ export type StoreWriterTask = { /** Write operation to run once earlier tasks for the same store path finish. */ @@ -21,6 +23,39 @@ export type StoreWriterQueue = { /** Store writer queues keyed by the canonical store path. */ type StoreWriterQueues = Map; +type ActiveStoreWriter = { + active: boolean; + parent: ActiveStoreWriter | undefined; + queues: StoreWriterQueues; + storePath: string; +}; + +const activeStoreWriters = new AsyncLocalStorage(); + +function isActiveStoreWriter(queues: StoreWriterQueues, storePath: string): boolean { + let active = activeStoreWriters.getStore(); + while (active) { + if (active.active && active.queues === queues && active.storePath === storePath) { + return true; + } + active = active.parent; + } + return false; +} + +async function runActiveStoreWriter( + queues: StoreWriterQueues, + storePath: string, + fn: () => Promise, +): Promise { + const writer = { active: true, parent: activeStoreWriters.getStore(), queues, storePath }; + try { + return await activeStoreWriters.run(writer, fn); + } finally { + writer.active = false; + } +} + function getOrCreateStoreWriterQueue( queues: StoreWriterQueues, storePath: string, @@ -89,6 +124,7 @@ export async function runQueuedStoreWrite(params: { storePath: string; label: string; fn: () => Promise; + reentrant?: boolean; }): Promise { if (!params.storePath || typeof params.storePath !== "string") { throw new Error( @@ -97,10 +133,15 @@ export async function runQueuedStoreWrite(params: { )}`, ); } + // Explicit reentrancy keeps one logical read/decide/write section on the + // active lane; ordinary async children must queue behind the current writer. + if (params.reentrant === true && isActiveStoreWriter(params.queues, params.storePath)) { + return await params.fn(); + } const queue = getOrCreateStoreWriterQueue(params.queues, params.storePath); return await new Promise((resolve, reject) => { const task: StoreWriterTask = { - fn: async () => await params.fn(), + fn: async () => await runActiveStoreWriter(params.queues, params.storePath, params.fn), resolve: (value) => resolve(value as T), reject, };