Skip to content

Commit 826c84e

Browse files
moguangyu5-designjalehman
moguangyu5-design
and
jalehman
authored
fix(config/sessions): narrow reply-session initialization revision to identity fields (#98835)
* fix(config/sessions): narrow reply-session initialization revision to identity fields The initialization guard compared the full persisted session entry, so background touches to updatedAt, heartbeat timestamps, context-budget metadata, etc. produced false-positive stale-snapshot conflicts and the "reply session initialization conflicted" error. Only sessionId and sessionFile matter for detecting a session rotation. Narrow the revision to those identity fields and add a regression test. Fixes #98672 * fix(config/sessions): merge current metadata when reply-init identity guard passes * fix(config/sessions): preserve only snapshot-drifted metadata in reply-init commit * fix: preserve cleared reply-session metadata * fix: allow same-session reply initialization drift --------- Co-authored-by: moguangyu5-design <[email protected]> Co-authored-by: Josh Lehman <[email protected]>
1 parent 55ed57a commit 826c84e

4 files changed

Lines changed: 656 additions & 8 deletions

File tree

‎src/auto-reply/reply/session.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,13 +72,13 @@ import { parseSoftResetCommand } from "./commands-reset-mode.js";
7272
import { resolveConversationBindingContextFromMessage } from "./conversation-binding-input.js";
7373
import { normalizeInboundTextNewlines } from "./inbound-text.js";
7474
import { stripMentions, stripStructuralPrefixes } from "./mentions.js";
75+
import { replyRunRegistry } from "./reply-run-registry.js";
7576
import { isResetAuthorizedForContext } from "./reset-authorization.js";
7677
import {
7778
maybeRetireLegacyMainDeliveryRoute,
7879
resolveLastChannelRaw,
7980
resolveLastToRaw,
8081
} from "./session-delivery.js";
81-
import { replyRunRegistry } from "./reply-run-registry.js";
8282
import {
8383
createReplySessionEntryHandle,
8484
type ReplySessionEntryHandle,
@@ -927,6 +927,7 @@ async function initSessionStateAttemptLocked(
927927
retiredEntry: retiredLegacyMainDelivery,
928928
sessionEntry,
929929
sessionKey,
930+
snapshotEntry: initializationSnapshot.currentEntry,
930931
storePath,
931932
});
932933
if (!committed.ok) {

‎src/config/sessions/session-accessor.reply-init-concurrency.test.ts‎

Lines changed: 226 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,226 @@
1+
import { spawn } from "node:child_process";
2+
import fs from "node:fs/promises";
3+
import os from "node:os";
4+
import path from "node:path";
5+
import process from "node:process";
6+
import { pathToFileURL } from "node:url";
7+
import { describe, expect, it, vi } from "vitest";
8+
import { loadSessionEntry, updateSessionEntry, upsertSessionEntry } from "./session-accessor.js";
9+
10+
vi.mock("../config.js", async () => ({
11+
...(await vi.importActual<typeof import("../config.js")>("../config.js")),
12+
getRuntimeConfig: vi.fn().mockReturnValue({}),
13+
}));
14+
15+
type ChildResult =
16+
| {
17+
ok: true;
18+
sessionEntry: {
19+
sessionFile?: string;
20+
sessionId?: string;
21+
updatedAt?: number;
22+
};
23+
}
24+
| {
25+
currentEntry?: {
26+
sessionId?: string;
27+
updatedAt?: number;
28+
};
29+
ok: false;
30+
reason: string;
31+
revision: string;
32+
};
33+
34+
const POLL_MS = 20;
35+
const WAIT_TIMEOUT_MS = 10_000;
36+
const SESSION_KEY = "agent:main:main";
37+
const AGENT_ID = "main";
38+
39+
async function waitForFile(filePath: string): Promise<void> {
40+
const deadline = Date.now() + WAIT_TIMEOUT_MS;
41+
while (Date.now() < deadline) {
42+
try {
43+
await fs.access(filePath);
44+
return;
45+
} catch {
46+
await new Promise((resolve) => {
47+
setTimeout(resolve, POLL_MS);
48+
});
49+
}
50+
}
51+
throw new Error(`timeout waiting for ${filePath}`);
52+
}
53+
54+
async function readJsonFile<T>(filePath: string): Promise<T> {
55+
return JSON.parse(await fs.readFile(filePath, "utf8")) as T;
56+
}
57+
58+
function createReplyInitChildScript(sessionAccessorUrl: string): string {
59+
return `
60+
const fs = await import("node:fs/promises");
61+
const {
62+
commitReplySessionInitialization,
63+
loadReplySessionInitializationSnapshot,
64+
} = await import(${JSON.stringify(sessionAccessorUrl)});
65+
66+
const POLL_MS = ${POLL_MS};
67+
const WAIT_TIMEOUT_MS = ${WAIT_TIMEOUT_MS};
68+
const SESSION_KEY = ${JSON.stringify(SESSION_KEY)};
69+
const AGENT_ID = ${JSON.stringify(AGENT_ID)};
70+
71+
async function waitForFile(filePath) {
72+
const deadline = Date.now() + WAIT_TIMEOUT_MS;
73+
while (Date.now() < deadline) {
74+
try {
75+
await fs.access(filePath);
76+
return;
77+
} catch {
78+
await new Promise((resolve) => {
79+
setTimeout(resolve, POLL_MS);
80+
});
81+
}
82+
}
83+
throw new Error(\`timeout waiting for \${filePath}\`);
84+
}
85+
86+
async function writeJsonFile(filePath, value) {
87+
await fs.writeFile(filePath, \`\${JSON.stringify(value, null, 2)}\\n\`, "utf8");
88+
}
89+
90+
const storePath = process.env.REPLY_INIT_STORE_PATH;
91+
const readyPath = process.env.REPLY_INIT_READY_PATH;
92+
const proceedPath = process.env.REPLY_INIT_PROCEED_PATH;
93+
const resultPath = process.env.REPLY_INIT_RESULT_PATH;
94+
const preparedUpdatedAt = process.env.REPLY_INIT_PREPARED_UPDATED_AT;
95+
if (!storePath || !readyPath || !proceedPath || !resultPath || !preparedUpdatedAt) {
96+
throw new Error("reply initialization child env is incomplete");
97+
}
98+
99+
const snapshot = loadReplySessionInitializationSnapshot({
100+
sessionKey: SESSION_KEY,
101+
storePath,
102+
});
103+
await writeJsonFile(readyPath, {
104+
currentEntry: snapshot.currentEntry,
105+
revision: snapshot.revision,
106+
});
107+
108+
await waitForFile(proceedPath);
109+
110+
const committed = await commitReplySessionInitialization({
111+
activeSessionKey: SESSION_KEY,
112+
agentId: AGENT_ID,
113+
expectedRevision: snapshot.revision,
114+
sessionEntry: {
115+
sessionId: "existing-session",
116+
updatedAt: Number(preparedUpdatedAt),
117+
},
118+
sessionKey: SESSION_KEY,
119+
snapshotEntry: snapshot.currentEntry,
120+
storePath,
121+
});
122+
await writeJsonFile(resultPath, committed);
123+
`;
124+
}
125+
126+
async function waitForChild(child: ReturnType<typeof spawn>): Promise<void> {
127+
let childStdout = "";
128+
let childStderr = "";
129+
child.stdout?.setEncoding("utf8");
130+
child.stderr?.setEncoding("utf8");
131+
child.stdout?.on("data", (chunk) => {
132+
childStdout += String(chunk);
133+
});
134+
child.stderr?.on("data", (chunk) => {
135+
childStderr += String(chunk);
136+
});
137+
138+
const childExit = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>(
139+
(resolve, reject) => {
140+
child.once("error", reject);
141+
child.once("exit", (code, signal) => resolve({ code, signal }));
142+
},
143+
);
144+
if (childExit.code !== 0) {
145+
throw new Error(
146+
`reply initialization child failed code=${String(childExit.code)} signal=${String(childExit.signal)}\nstdout:\n${childStdout}\nstderr:\n${childStderr}`,
147+
);
148+
}
149+
}
150+
151+
describe("reply session initialization concurrency", () => {
152+
it("commits after same-session activity from another process", async () => {
153+
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-reply-init-"));
154+
const sessionAccessorUrl = pathToFileURL(
155+
path.resolve("src/config/sessions/session-accessor.ts"),
156+
).href;
157+
const storePath = path.join(tempDir, "sessions.json");
158+
const readyPath = path.join(tempDir, "snapshot-ready.json");
159+
const proceedPath = path.join(tempDir, "proceed");
160+
const resultPath = path.join(tempDir, "result.json");
161+
const baseTime = Date.now();
162+
const activeTurnUpdatedAt = baseTime + 20;
163+
const preparedUpdatedAt = baseTime + 30;
164+
165+
try {
166+
await upsertSessionEntry(
167+
{ sessionKey: SESSION_KEY, storePath },
168+
{
169+
sessionId: "existing-session",
170+
updatedAt: baseTime,
171+
},
172+
);
173+
174+
const child = spawn(
175+
process.execPath,
176+
[
177+
"--import",
178+
"tsx",
179+
"--input-type=module",
180+
"--eval",
181+
createReplyInitChildScript(sessionAccessorUrl),
182+
],
183+
{
184+
env: {
185+
...process.env,
186+
REPLY_INIT_PREPARED_UPDATED_AT: String(preparedUpdatedAt),
187+
REPLY_INIT_PROCEED_PATH: proceedPath,
188+
REPLY_INIT_READY_PATH: readyPath,
189+
REPLY_INIT_RESULT_PATH: resultPath,
190+
REPLY_INIT_STORE_PATH: storePath,
191+
},
192+
stdio: ["ignore", "pipe", "pipe"],
193+
},
194+
);
195+
196+
await waitForFile(readyPath);
197+
const snapshot = await readJsonFile<{ currentEntry?: unknown; revision: string }>(readyPath);
198+
expect(snapshot.revision).toBe(JSON.stringify({ sessionId: "existing-session" }));
199+
200+
await updateSessionEntry(
201+
{ sessionKey: SESSION_KEY, storePath },
202+
() => ({ updatedAt: activeTurnUpdatedAt }),
203+
{ skipMaintenance: true },
204+
);
205+
await fs.writeFile(proceedPath, "go\n", "utf8");
206+
await waitForChild(child);
207+
208+
const result = await readJsonFile<ChildResult>(resultPath);
209+
expect(result).toMatchObject({
210+
ok: true,
211+
sessionEntry: {
212+
sessionId: "existing-session",
213+
updatedAt: preparedUpdatedAt,
214+
},
215+
});
216+
expect(
217+
loadSessionEntry({ readConsistency: "latest", sessionKey: SESSION_KEY, storePath }),
218+
).toMatchObject({
219+
sessionId: "existing-session",
220+
updatedAt: preparedUpdatedAt,
221+
});
222+
} finally {
223+
await fs.rm(tempDir, { recursive: true, force: true });
224+
}
225+
}, 15_000);
226+
});

0 commit comments

Comments
 (0)