Skip to content

Commit 61fc7d2

Browse files
jalehmanjalehman
committed
fix: allow same-session reply initialization drift
1 parent a66f164 commit 61fc7d2

3 files changed

Lines changed: 341 additions & 31 deletions

File tree

‎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+
});

‎src/config/sessions/session-accessor.test.ts‎

Lines changed: 73 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -534,7 +534,7 @@ describe("session accessor file-backed seam", () => {
534534
});
535535
});
536536

537-
it("rejects stale reply session initialization when decision metadata changes", async () => {
537+
it("commits reply session initialization despite active-turn metadata changes", async () => {
538538
const sessionKey = "agent:main:main";
539539
await upsertSessionEntry(
540540
{ sessionKey, storePath },
@@ -544,16 +544,16 @@ describe("session accessor file-backed seam", () => {
544544
},
545545
);
546546
const snapshot = loadReplySessionInitializationSnapshot({ sessionKey, storePath });
547-
let concurrentUpdatedAt = 0;
548547
await sessionStore.updateSessionStore(storePath, (store) => {
549548
const current = store[sessionKey];
550549
if (!current) {
551550
throw new Error("expected existing session entry");
552551
}
553-
concurrentUpdatedAt = current.updatedAt + 1;
554552
store[sessionKey] = {
555553
...current,
556-
updatedAt: concurrentUpdatedAt,
554+
compactionCount: 1,
555+
totalTokensFresh: false,
556+
updatedAt: current.updatedAt + 1,
557557
};
558558
});
559559

@@ -570,13 +570,21 @@ describe("session accessor file-backed seam", () => {
570570
storePath,
571571
});
572572

573-
expect(committed).toMatchObject({
574-
ok: false,
575-
reason: "stale-snapshot",
573+
expect(committed.ok).toBe(true);
574+
if (!committed.ok) {
575+
throw new Error("expected reply session initialization to commit");
576+
}
577+
expect(committed.sessionEntry).toMatchObject({
578+
compactionCount: 1,
579+
sessionId: "existing-session",
580+
totalTokensFresh: false,
581+
updatedAt: 30,
576582
});
577583
expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({
584+
compactionCount: 1,
578585
sessionId: "existing-session",
579-
updatedAt: concurrentUpdatedAt,
586+
totalTokensFresh: false,
587+
updatedAt: 30,
580588
});
581589
});
582590

@@ -643,6 +651,63 @@ describe("session accessor file-backed seam", () => {
643651
});
644652
});
645653

654+
it("preserves concurrent optional additions when prepared fields are undefined", async () => {
655+
const sessionKey = "agent:main:main";
656+
await upsertSessionEntry(
657+
{ sessionKey, storePath },
658+
{
659+
sessionId: "existing-session",
660+
updatedAt: 10,
661+
},
662+
);
663+
664+
const snapshot = loadReplySessionInitializationSnapshot({ sessionKey, storePath });
665+
666+
await sessionStore.updateSessionStore(storePath, (store) => {
667+
const current = store[sessionKey];
668+
if (!current) {
669+
throw new Error("expected existing session entry");
670+
}
671+
store[sessionKey] = {
672+
...current,
673+
modelOverride: "channel-model",
674+
modelOverrideSource: "user",
675+
};
676+
});
677+
678+
const committed = await commitReplySessionInitialization({
679+
activeSessionKey: sessionKey,
680+
agentId: "main",
681+
expectedRevision: snapshot.revision,
682+
sessionEntry: {
683+
sessionId: "existing-session",
684+
updatedAt: 30,
685+
modelOverride: undefined,
686+
modelOverrideSource: undefined,
687+
},
688+
sessionKey,
689+
snapshotEntry: snapshot.currentEntry,
690+
storePath,
691+
});
692+
693+
expect(committed.ok).toBe(true);
694+
if (!committed.ok) {
695+
throw new Error("expected reply session initialization to commit");
696+
}
697+
expect(committed.sessionEntry).toMatchObject({
698+
modelOverride: "channel-model",
699+
modelOverrideSource: "user",
700+
sessionId: "existing-session",
701+
updatedAt: 30,
702+
});
703+
expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({
704+
modelOverride: "channel-model",
705+
modelOverrideSource: "user",
706+
sessionId: "existing-session",
707+
updatedAt: 30,
708+
});
709+
});
710+
646711
it("does not restore pending final delivery metadata cleared after the snapshot", async () => {
647712
const sessionKey = "agent:main:main";
648713
await upsertSessionEntry(

0 commit comments

Comments
 (0)