Skip to content

Commit 1e878dd

Browse files
lzyyzznlvincentkoc
lzyyzznl
and
vincentkoc
authored
#92109: [Bug]: EmbeddedAttemptSessionTakeoverError caused by Btrfs ctimeNs instability (#92123)
* fix(session-lock): remove ctimeNs from session file fingerprint comparison Btrfs background maintenance (snapshots, scrub, quota) updates ctime without any file content change. Including ctimeNs in the fingerprint causes false-positive EmbeddedAttemptSessionTakeoverError on all Btrfs filesystems, breaking cron jobs and subagent spawns with 100% failure. dev + ino + size + mtimeNs are sufficient to detect external writes — any content change will also update mtimeNs and/or size. ctimeNs only tracks metadata changes and adds no meaningful protection. Closes #92109 * test(session-lock): cover ctime-only fence drift * fix(session-lock): narrow ctime drift acceptance * fix(session-lock): trust verified ctime drift * fix(session-lock): bound ctime drift digest * fix(session-lock): skip absent ctime digest --------- Co-authored-by: Vincent Koc <[email protected]>
1 parent dbf1b74 commit 1e878dd

2 files changed

Lines changed: 260 additions & 11 deletions

File tree

‎src/agents/embedded-agent-runner/run/attempt.session-lock.test.ts‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,15 @@ async function createTempSessionFile(): Promise<string> {
5656
return sessionFile;
5757
}
5858

59+
function cloneBigIntStatWith(
60+
stat: Awaited<ReturnType<typeof fs.stat>>,
61+
fields: Partial<Awaited<ReturnType<typeof fs.stat>>>,
62+
): Awaited<ReturnType<typeof fs.stat>> {
63+
return Object.assign(Object.create(Object.getPrototypeOf(stat)), stat, fields) as Awaited<
64+
ReturnType<typeof fs.stat>
65+
>;
66+
}
67+
5968
describe("embedded attempt session lock lifecycle", () => {
6069
it("serializes embedded attempts that share a session file owner", async () => {
6170
const sessionFile = await createTempSessionFile();
@@ -834,6 +843,152 @@ describe("embedded attempt session lock lifecycle", () => {
834843
expect(release).toHaveBeenCalledTimes(2);
835844
});
836845

846+
it("allows ctime-only fingerprint drift while the prompt lock is released", async () => {
847+
const sessionFile = await createTempSessionFile();
848+
const release = vi.fn(async () => {});
849+
const acquireSessionWriteLockLocalCtimeDrift = vi.fn(async () => ({ release }));
850+
const controller = await createEmbeddedAttemptSessionLockController({
851+
acquireSessionWriteLock: acquireSessionWriteLockLocalCtimeDrift,
852+
lockOptions: { ...lockOptions, sessionFile },
853+
});
854+
855+
await controller.releaseForPrompt();
856+
857+
const stableStat = await fs.stat(sessionFile, { bigint: true });
858+
const driftedStat = cloneBigIntStatWith(stableStat, {
859+
ctimeNs: stableStat.ctimeNs + 1_000_000n,
860+
});
861+
const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (target, options) => {
862+
if (target === sessionFile && options?.bigint === true) {
863+
return driftedStat;
864+
}
865+
throw new Error(`unexpected stat call for ${String(target)}`);
866+
});
867+
868+
try {
869+
await expect(controller.withSessionWriteLock(() => "finalize")).resolves.toBe("finalize");
870+
} finally {
871+
statSpy.mockRestore();
872+
}
873+
expect(controller.hasSessionTakeover()).toBe(false);
874+
expect(acquireSessionWriteLockLocalCtimeDrift).toHaveBeenCalledTimes(2);
875+
expect(release).toHaveBeenCalledTimes(2);
876+
});
877+
878+
it("trusts owned writes after accepting ctime-only fingerprint drift", async () => {
879+
const sessionFile = await createTempSessionFile();
880+
const release = vi.fn(async () => {});
881+
const acquireSessionWriteLockLocalOwnedAfterDrift = vi.fn(async () => ({ release }));
882+
const controller = await createEmbeddedAttemptSessionLockController({
883+
acquireSessionWriteLock: acquireSessionWriteLockLocalOwnedAfterDrift,
884+
lockOptions: { ...lockOptions, sessionFile },
885+
});
886+
887+
await controller.releaseForPrompt();
888+
889+
const stableStat = await fs.stat(sessionFile, { bigint: true });
890+
const driftedStat = cloneBigIntStatWith(stableStat, {
891+
ctimeNs: stableStat.ctimeNs + 1_000_000n,
892+
});
893+
const appendedText = '{"type":"message","id":"owned-after-drift"}\n';
894+
const changedStat = cloneBigIntStatWith(stableStat, {
895+
ctimeNs: stableStat.ctimeNs + 2_000_000n,
896+
mtimeNs: stableStat.mtimeNs + 1_000_000n,
897+
size: stableStat.size + BigInt(Buffer.byteLength(appendedText)),
898+
});
899+
let currentStat = driftedStat;
900+
const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (target, options) => {
901+
if (target === sessionFile && options?.bigint === true) {
902+
return currentStat;
903+
}
904+
throw new Error(`unexpected stat call for ${String(target)}`);
905+
});
906+
907+
try {
908+
await expect(
909+
controller.withSessionWriteLock(
910+
async () => {
911+
currentStat = changedStat;
912+
await fs.appendFile(sessionFile, appendedText, "utf8");
913+
},
914+
{ publishOwnedWrite: true },
915+
),
916+
).resolves.toBeUndefined();
917+
await expect(controller.withSessionWriteLock(() => "finalize")).resolves.toBe("finalize");
918+
} finally {
919+
statSpy.mockRestore();
920+
}
921+
expect(controller.hasSessionTakeover()).toBe(false);
922+
expect(acquireSessionWriteLockLocalOwnedAfterDrift).toHaveBeenCalledTimes(3);
923+
expect(release).toHaveBeenCalledTimes(3);
924+
});
925+
926+
it("allows ctime-only fingerprint drift for large transcript snapshots", async () => {
927+
const sessionFile = await createTempSessionFile();
928+
await fs.writeFile(sessionFile, Buffer.alloc(8 * 1024 * 1024 + 1, "x"));
929+
const release = vi.fn(async () => {});
930+
const acquireSessionWriteLockLocalLargeCtimeDrift = vi.fn(async () => ({ release }));
931+
const controller = await createEmbeddedAttemptSessionLockController({
932+
acquireSessionWriteLock: acquireSessionWriteLockLocalLargeCtimeDrift,
933+
lockOptions: { ...lockOptions, sessionFile },
934+
});
935+
936+
await controller.releaseForPrompt();
937+
938+
const stableStat = await fs.stat(sessionFile, { bigint: true });
939+
const driftedStat = cloneBigIntStatWith(stableStat, {
940+
ctimeNs: stableStat.ctimeNs + 1_000_000n,
941+
});
942+
const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (target, options) => {
943+
if (target === sessionFile && options?.bigint === true) {
944+
return driftedStat;
945+
}
946+
throw new Error(`unexpected stat call for ${String(target)}`);
947+
});
948+
949+
try {
950+
await expect(controller.withSessionWriteLock(() => "finalize")).resolves.toBe("finalize");
951+
} finally {
952+
statSpy.mockRestore();
953+
}
954+
expect(controller.hasSessionTakeover()).toBe(false);
955+
});
956+
957+
it("rejects same-size transcript rewrites with restored mtime", async () => {
958+
const sessionFile = await createTempSessionFile();
959+
const release = vi.fn(async () => {});
960+
const acquireSessionWriteLockLocalSameSizeRewrite = vi.fn(async () => ({ release }));
961+
const controller = await createEmbeddedAttemptSessionLockController({
962+
acquireSessionWriteLock: acquireSessionWriteLockLocalSameSizeRewrite,
963+
lockOptions: { ...lockOptions, sessionFile },
964+
});
965+
966+
await controller.releaseForPrompt();
967+
968+
const stableStat = await fs.stat(sessionFile, { bigint: true });
969+
await fs.writeFile(sessionFile, '{"type":"sessioN"}\n', "utf8");
970+
const driftedStat = cloneBigIntStatWith(stableStat, {
971+
ctimeNs: stableStat.ctimeNs + 1_000_000n,
972+
});
973+
const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (target, options) => {
974+
if (target === sessionFile && options?.bigint === true) {
975+
return driftedStat;
976+
}
977+
throw new Error(`unexpected stat call for ${String(target)}`);
978+
});
979+
980+
try {
981+
await expect(controller.withSessionWriteLock(() => "finalize")).rejects.toBeInstanceOf(
982+
EmbeddedAttemptSessionTakeoverError,
983+
);
984+
} finally {
985+
statSpy.mockRestore();
986+
}
987+
expect(controller.hasSessionTakeover()).toBe(true);
988+
expect(acquireSessionWriteLockLocalSameSizeRewrite).toHaveBeenCalledTimes(2);
989+
expect(release).toHaveBeenCalledTimes(2);
990+
});
991+
837992
it("still rejects external edits after the prompt stream lock is reacquired", async () => {
838993
const sessionFile = await createTempSessionFile();
839994
const release = vi.fn(async () => {});

‎src/agents/embedded-agent-runner/run/attempt.session-lock.ts‎

Lines changed: 105 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@
22
* Coordinates embedded-attempt session ownership, takeover, and prompt locks.
33
*/
44
import { AsyncLocalStorage } from "node:async_hooks";
5-
import { readFileSync, statSync } from "node:fs";
5+
import { createHash } from "node:crypto";
6+
import { createReadStream, readFileSync, statSync } from "node:fs";
67
import fs from "node:fs/promises";
78
import { isDeepStrictEqual } from "node:util";
89
import { normalizeStringEntries } from "@openclaw/normalization-core/string-normalization";
@@ -57,10 +58,12 @@ const MAX_BENIGN_SESSION_FENCE_ADVANCE_BYTES = 1024 * 1024;
5758
const MAX_BENIGN_SESSION_FENCE_REWRITE_BYTES = 8 * 1024 * 1024;
5859
const MAX_BENIGN_SESSION_FENCE_REWRITE_RESULT_BYTES =
5960
MAX_BENIGN_SESSION_FENCE_REWRITE_BYTES + MAX_BENIGN_SESSION_FENCE_ADVANCE_BYTES;
61+
const MAX_BENIGN_SESSION_FENCE_CTIME_DIGEST_BYTES = 32 * 1024 * 1024;
6062
const MAX_SAFE_FILE_OFFSET = BigInt(Number.MAX_SAFE_INTEGER);
6163

6264
type SessionFileFenceSnapshot = {
6365
fingerprint: SessionFileFingerprint;
66+
digest?: string;
6467
text?: string;
6568
};
6669

@@ -90,6 +93,20 @@ function sameSessionFileIdentity(
9093
return Boolean(left?.exists && right.exists && left.dev === right.dev && left.ino === right.ino);
9194
}
9295

96+
function sameSessionFileContentMetadata(
97+
left: SessionFileFingerprint | undefined,
98+
right: SessionFileFingerprint,
99+
): boolean {
100+
return Boolean(
101+
left?.exists &&
102+
right.exists &&
103+
left.dev === right.dev &&
104+
left.ino === right.ino &&
105+
left.size === right.size &&
106+
left.mtimeNs === right.mtimeNs,
107+
);
108+
}
109+
93110
function splitSessionFileLines(text: string): string[] {
94111
return normalizeStringEntries(text.split(/\r?\n/));
95112
}
@@ -218,21 +235,45 @@ async function readSessionFileFenceSnapshot(
218235
sessionFile: string,
219236
): Promise<SessionFileFenceSnapshot> {
220237
const fingerprint = await readSessionFileFingerprint(sessionFile);
238+
if (!fingerprint.exists) {
239+
return { fingerprint };
240+
}
221241
if (
222-
!fingerprint.exists ||
223-
fingerprint.size > BigInt(MAX_BENIGN_SESSION_FENCE_REWRITE_BYTES) ||
224-
fingerprint.size > MAX_SAFE_FILE_OFFSET
242+
fingerprint.size <= BigInt(MAX_BENIGN_SESSION_FENCE_REWRITE_BYTES) &&
243+
fingerprint.size <= MAX_SAFE_FILE_OFFSET
225244
) {
226-
return { fingerprint };
245+
try {
246+
return {
247+
fingerprint,
248+
text: await fs.readFile(sessionFile, "utf8"),
249+
};
250+
} catch {
251+
return { fingerprint };
252+
}
227253
}
228-
try {
229-
return {
230-
fingerprint,
231-
text: await fs.readFile(sessionFile, "utf8"),
232-
};
233-
} catch {
254+
if (fingerprint.size > BigInt(MAX_BENIGN_SESSION_FENCE_CTIME_DIGEST_BYTES)) {
234255
return { fingerprint };
235256
}
257+
return {
258+
fingerprint,
259+
digest: await readSessionFileDigest(sessionFile),
260+
};
261+
}
262+
263+
async function readSessionFileDigest(sessionFile: string): Promise<string | undefined> {
264+
const hash = createHash("sha256");
265+
return await new Promise<string | undefined>((resolve) => {
266+
const stream = createReadStream(sessionFile);
267+
stream.on("data", (chunk) => {
268+
hash.update(chunk);
269+
});
270+
stream.on("error", () => {
271+
resolve(undefined);
272+
});
273+
stream.on("end", () => {
274+
resolve(hash.digest("hex"));
275+
});
276+
});
236277
}
237278

238279
async function sessionFenceAdvanceIsBenign(params: {
@@ -259,6 +300,33 @@ async function sessionFenceAdvanceIsBenign(params: {
259300
return lines.length > 0 && lines.every(isTranscriptOnlyOpenClawAssistantLine);
260301
}
261302

303+
async function sessionFenceCtimeDriftIsBenign(params: {
304+
sessionFile: string;
305+
previous: SessionFileFenceSnapshot | undefined;
306+
current: SessionFileFingerprint;
307+
}): Promise<boolean> {
308+
if (
309+
!sameSessionFileContentMetadata(params.previous?.fingerprint, params.current) ||
310+
params.previous?.fingerprint.exists !== true ||
311+
!params.current.exists ||
312+
params.previous.fingerprint.ctimeNs === params.current.ctimeNs
313+
) {
314+
return false;
315+
}
316+
if (params.previous.text === undefined) {
317+
if (params.previous.digest === undefined) {
318+
return false;
319+
}
320+
const currentDigest = await readSessionFileDigest(params.sessionFile);
321+
return currentDigest !== undefined && currentDigest === params.previous.digest;
322+
}
323+
try {
324+
return (await fs.readFile(params.sessionFile, "utf8")) === params.previous.text;
325+
} catch {
326+
return false;
327+
}
328+
}
329+
262330
async function sessionFenceRewriteIsBenign(params: {
263331
sessionFile: string;
264332
previous: SessionFileFenceSnapshot | undefined;
@@ -500,6 +568,19 @@ function recordOwnedSessionFileWrite(
500568
return ownedSessionFileWriteGeneration;
501569
}
502570

571+
function recordTrustedSessionFileState(
572+
sessionFileKey: string,
573+
fingerprint: SessionFileFingerprint,
574+
): number {
575+
ownedSessionFileWriteGeneration += 1;
576+
const state = {
577+
generation: ownedSessionFileWriteGeneration,
578+
fingerprint,
579+
};
580+
trustedSessionFileStates.set(sessionFileKey, state);
581+
return ownedSessionFileWriteGeneration;
582+
}
583+
503584
function trustSessionFileState(
504585
sessionFileKey: string,
505586
fingerprint: SessionFileFingerprint,
@@ -731,6 +812,19 @@ export async function createEmbeddedAttemptSessionLockController(params: {
731812
return;
732813
}
733814

815+
if (
816+
await sessionFenceCtimeDriftIsBenign({
817+
sessionFile: params.lockOptions.sessionFile,
818+
previous: fenceSnapshot,
819+
current,
820+
})
821+
) {
822+
fenceSnapshot = await readSessionFileFenceSnapshot(params.lockOptions.sessionFile);
823+
fenceFingerprint = fenceSnapshot.fingerprint;
824+
fenceGeneration = recordTrustedSessionFileState(sessionFileFenceKey, current);
825+
return;
826+
}
827+
734828
if (
735829
(await sessionFenceAdvanceIsBenign({
736830
sessionFile: params.lockOptions.sessionFile,

0 commit comments

Comments
 (0)