Skip to content

Commit 30a2edf

Browse files
openperfopenperf
committed
fix(agents): serialize new-session resolution per session key
Concurrent commands for the same session key minted separate sessionIds because resolveSession ran before the per-session execution lane, so a second OpenAI-compatible request that arrived mid-turn forked an isolated, memory-less session. Reserve the brand-new-session path per session key so concurrent commands adopt one sessionId. Fixes #84575
1 parent d29c3a5 commit 30a2edf

3 files changed

Lines changed: 258 additions & 2 deletions

File tree

‎src/agents/agent-command.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,7 @@ import {
9797
resolveInternalEventTranscriptBody,
9898
} from "./command/attempt-execution.shared.js";
9999
import { resolveAgentRunContext } from "./command/run-context.js";
100-
import { resolveSession } from "./command/session.js";
100+
import { resolveSessionWithReservation } from "./command/session-resolution-reservation.js";
101101
import type { AgentCommandIngressOpts, AgentCommandOpts } from "./command/types.js";
102102
import { DEFAULT_MODEL, DEFAULT_PROVIDER } from "./defaults.js";
103103
import {
@@ -715,13 +715,14 @@ async function prepareAgentCommandExecution(opts: AgentCommandOpts, runtime: Run
715715
const commandOpts = toSessionKey
716716
? { ...opts, to: undefined, sessionKey: explicitSessionKey }
717717
: opts;
718-
const sessionResolution = resolveSession({
718+
const sessionResolution = await resolveSessionWithReservation({
719719
cfg,
720720
to: commandOpts.to,
721721
sessionId: commandOpts.sessionId,
722722
sessionKey: explicitSessionKey,
723723
agentId: agentIdOverride,
724724
clone: false,
725+
suppressVisibleSessionEffects: opts.sessionEffects === "internal",
725726
});
726727

727728
const { sessionId, sessionKey, storePath, isNewSession, persistedThinking, persistedVerbose } =

‎src/agents/command/session-resolution-reservation.test.ts‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
import fs from "node:fs/promises";
2+
import os from "node:os";
3+
import path from "node:path";
4+
import { afterEach, describe, expect, it } from "vitest";
5+
import { replaceSessionEntry } from "../../config/sessions/session-accessor.js";
6+
import { clearSessionStoreCaches } from "../../config/sessions/store-cache.js";
7+
import { withSessionStoreWriterForTest } from "../../config/sessions/store-writer.js";
8+
import { loadSessionStore } from "../../config/sessions/store.js";
9+
import type { OpenClawConfig } from "../../config/types.openclaw.js";
10+
import { resolveSessionWithReservation } from "./session-resolution-reservation.js";
11+
import { resolveSession } from "./session.js";
12+
13+
/** Flush queued micro/macro tasks so an in-flight reservation reaches its persist. */
14+
async function flushPendingWork(): Promise<void> {
15+
for (let i = 0; i < 5; i += 1) {
16+
await new Promise((resolve) => {
17+
setTimeout(resolve, 0);
18+
});
19+
}
20+
}
21+
22+
async function withTempStore<T>(
23+
run: (params: { cfg: OpenClawConfig; storePath: string }) => Promise<T>,
24+
): Promise<T> {
25+
const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-reservation-"));
26+
const storePath = path.join(dir, "sessions.json");
27+
const cfg = { session: { store: storePath, mainKey: "main" } } as OpenClawConfig;
28+
try {
29+
return await run({ cfg, storePath });
30+
} finally {
31+
clearSessionStoreCaches();
32+
await fs.rm(dir, { recursive: true, force: true });
33+
}
34+
}
35+
36+
afterEach(() => {
37+
clearSessionStoreCaches();
38+
});
39+
40+
describe("resolveSessionWithReservation", () => {
41+
const sessionKey = "agent:main:finn:c1";
42+
43+
it("forks distinct sessionIds without the reservation lock (regression repro)", async () => {
44+
await withTempStore(async ({ cfg }) => {
45+
const first = resolveSession({ cfg, sessionKey });
46+
const second = resolveSession({ cfg, sessionKey });
47+
expect(first.isNewSession).toBe(true);
48+
expect(second.isNewSession).toBe(true);
49+
// Without serialization both requests mint their own id, so the second
50+
// request runs in an isolated, memory-less session.
51+
expect(first.sessionId).not.toBe(second.sessionId);
52+
});
53+
});
54+
55+
it("gives concurrent same-key requests one shared sessionId", async () => {
56+
await withTempStore(async ({ cfg, storePath }) => {
57+
const [first, second] = await Promise.all([
58+
resolveSessionWithReservation({ cfg, sessionKey }),
59+
resolveSessionWithReservation({ cfg, sessionKey }),
60+
]);
61+
expect(first.sessionId).toBe(second.sessionId);
62+
// Exactly one resolution created the session; the other adopted it.
63+
expect([first.isNewSession, second.isNewSession].filter(Boolean)).toHaveLength(1);
64+
// The reserved mapping is persisted so later turns resume the same session.
65+
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
66+
expect(persisted?.sessionId).toBe(first.sessionId);
67+
});
68+
});
69+
70+
it("reuses the reserved id for a follow-up after the first request", async () => {
71+
await withTempStore(async ({ cfg }) => {
72+
const first = await resolveSessionWithReservation({ cfg, sessionKey });
73+
const second = await resolveSessionWithReservation({ cfg, sessionKey });
74+
expect(second.sessionId).toBe(first.sessionId);
75+
expect(second.isNewSession).toBe(false);
76+
});
77+
});
78+
79+
it("leaves the explicit sessionId path unchanged", async () => {
80+
await withTempStore(async ({ cfg }) => {
81+
const resolution = await resolveSessionWithReservation({
82+
cfg,
83+
sessionId: "explicit-123",
84+
});
85+
expect(resolution.sessionId).toBe("explicit-123");
86+
});
87+
});
88+
89+
it("does not write a visible store row for internal handoffs (suppressVisibleSessionEffects)", async () => {
90+
await withTempStore(async ({ cfg, storePath }) => {
91+
const resolution = await resolveSessionWithReservation({
92+
cfg,
93+
sessionKey,
94+
suppressVisibleSessionEffects: true,
95+
});
96+
expect(resolution.isNewSession).toBe(true);
97+
expect(resolution.sessionId).toBeTruthy();
98+
// Internal handoffs must not leak a visible session-store row.
99+
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
100+
expect(persisted).toBeUndefined();
101+
});
102+
});
103+
104+
it("adopts a concurrent rebind instead of reverting it to the reserved id", async () => {
105+
await withTempStore(async ({ cfg, storePath }) => {
106+
const now = Date.now();
107+
let releaseHold: () => void = () => {};
108+
const held = new Promise<void>((resolve) => {
109+
releaseHold = resolve;
110+
});
111+
let signalHeld: () => void = () => {};
112+
const queueHeld = new Promise<void>((resolve) => {
113+
signalHeld = resolve;
114+
});
115+
116+
// Hold the per-store writer queue so the competing rebind and the
117+
// reservation's persist run under the lock in a deterministic order.
118+
const holder = withSessionStoreWriterForTest(storePath, async () => {
119+
signalHeld();
120+
await held;
121+
});
122+
await queueHeld;
123+
124+
// A non-reservation writer (models /reset or another endpoint) claims the
125+
// key with a different sessionId; it is queued ahead of the reservation
126+
// persist while the reservation still observes an empty key.
127+
const rebind = replaceSessionEntry(
128+
{ sessionKey, storePath },
129+
{ sessionId: "rebind-id", updatedAt: now, sessionStartedAt: now },
130+
);
131+
132+
// The reservation mints its own id and its persist queues behind the
133+
// rebind, so the rebind wins the store write lock first.
134+
const reservation = resolveSessionWithReservation({ cfg, sessionKey });
135+
await flushPendingWork();
136+
137+
releaseHold();
138+
const [, resolution] = await Promise.all([rebind, reservation, holder]);
139+
140+
// The reservation must not stamp its stale minted id back over the newer
141+
// rebind; it adopts the winning identity instead.
142+
expect(resolution.sessionId).toBe("rebind-id");
143+
const persisted = loadSessionStore(storePath, { skipCache: true })[sessionKey];
144+
expect(persisted?.sessionId).toBe("rebind-id");
145+
});
146+
});
147+
});

‎src/agents/command/session-resolution-reservation.ts‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
import type { OpenClawConfig } from "../../config/types.openclaw.js";
2+
import { persistSessionEntry } from "./attempt-execution.shared.js";
3+
import { resolveSession } from "./session.js";
4+
5+
// Return shape of resolveSession; kept local so session.ts need not widen its
6+
// exported surface just to name the reservation wrapper's return type.
7+
type SessionResolution = ReturnType<typeof resolveSession>;
8+
9+
// Per-session-key serialization for resolving a brand-new session identity.
10+
//
11+
// Concurrent commands that target the same session key must agree on a single
12+
// `sessionId`. `resolveSession` mints a fresh `crypto.randomUUID()` whenever the
13+
// key has no fresh stored entry, and that resolution runs before the per-session
14+
// execution lane is entered (the lane is keyed by session key but only
15+
// serializes execution). Two requests that race here -- e.g. a follow-up sent
16+
// while the first turn is still in flight on the OpenAI-compatible endpoint --
17+
// each mint their own id, so the later one runs in an isolated, memory-less
18+
// session. Serializing the mint-and-reserve step closes that window: the first
19+
// request persists the key-to-id mapping and any concurrent request adopts it
20+
// instead of forking a second session.
21+
const SESSION_RESERVATION_QUEUE = new Map<string, Promise<unknown>>();
22+
23+
async function serializeReservation<T>(key: string, task: () => Promise<T>): Promise<T> {
24+
const prev = SESSION_RESERVATION_QUEUE.get(key) ?? Promise.resolve();
25+
const next = prev.then(task, task);
26+
SESSION_RESERVATION_QUEUE.set(key, next);
27+
try {
28+
return await next;
29+
} finally {
30+
if (SESSION_RESERVATION_QUEUE.get(key) === next) {
31+
SESSION_RESERVATION_QUEUE.delete(key);
32+
}
33+
}
34+
}
35+
36+
export type ResolveSessionInput = {
37+
cfg: OpenClawConfig;
38+
to?: string;
39+
sessionId?: string;
40+
sessionKey?: string;
41+
agentId?: string;
42+
clone?: boolean;
43+
/**
44+
* Internal handoffs (sessionEffects === "internal") must not write visible
45+
* session-store rows. Skip the reservation persist entirely; internal runs
46+
* are orchestrated, not user-fired, so there is no same-key race to protect.
47+
*/
48+
suppressVisibleSessionEffects?: boolean;
49+
};
50+
51+
/**
52+
* Resolve a session like {@link resolveSession}, but serialize the brand-new
53+
* session path per session key so concurrent commands cannot fork separate
54+
* sessions for the same key.
55+
*/
56+
export async function resolveSessionWithReservation(
57+
opts: ResolveSessionInput,
58+
): Promise<SessionResolution> {
59+
const resolution = resolveSession(opts);
60+
// Only a brand-new session mints a random id and is therefore racy. An
61+
// explicit session id or an existing fresh entry already pins the identity,
62+
// so those paths skip the lock entirely and keep steady-state turns lock-free.
63+
// Internal handoffs also skip reservation so they cannot create a visible
64+
// session-store row that the suppressVisibleSessionEffects contract forbids.
65+
if (
66+
opts.suppressVisibleSessionEffects ||
67+
opts.sessionId?.trim() ||
68+
!resolution.isNewSession ||
69+
!resolution.sessionKey
70+
) {
71+
return resolution;
72+
}
73+
const reservationKey = `${resolution.storePath}::${resolution.sessionKey}`;
74+
return await serializeReservation(reservationKey, async () => {
75+
// Re-resolve inside the critical section: a concurrent command may have
76+
// reserved an id between the initial resolve and acquiring the lock.
77+
const rechecked = resolveSession(opts);
78+
if (!rechecked.isNewSession || !rechecked.sessionKey || !rechecked.sessionStore) {
79+
return rechecked;
80+
}
81+
const reservedSessionId = rechecked.sessionId;
82+
// Id of the row resolveSession just observed: undefined when the key is
83+
// free, or the stale/expired entry this new session legitimately replaces.
84+
const observedSessionId = rechecked.sessionStore[rechecked.sessionKey]?.sessionId;
85+
const now = Date.now();
86+
// Compare-and-set under the store write lock: the reservation may only claim
87+
// the key when it is still free, already points at our reserved id, or still
88+
// holds the exact stale row we resolved against. A concurrent /new, /reset,
89+
// or same-key writer that rebound the key to a different id between resolve
90+
// and persist must win -- stamping our id back would silently revert it.
91+
const persisted = await persistSessionEntry({
92+
sessionStore: rechecked.sessionStore,
93+
sessionKey: rechecked.sessionKey,
94+
storePath: rechecked.storePath,
95+
entry: { sessionId: reservedSessionId, updatedAt: now, sessionStartedAt: now },
96+
shouldPersist: (existing) =>
97+
!existing ||
98+
existing.sessionId === reservedSessionId ||
99+
existing.sessionId === observedSessionId,
100+
});
101+
// The guard rejected the write because another writer rebound the key; adopt
102+
// the winning identity instead of returning the stale id we minted.
103+
if (persisted && persisted.sessionId !== reservedSessionId) {
104+
return resolveSession(opts);
105+
}
106+
return rechecked;
107+
});
108+
}

0 commit comments

Comments
 (0)