furyhawk/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-21 10:16:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: refresh prompt fence after compaction writes

Browse Source 
Fix embedded attempts falsely reporting session takeover after OpenClaw-owned auto-compaction writes a compaction entry while the prompt fence is released.

The compaction append path now publishes an owned session-file fence only when the guarded SessionManager append produced the expected compaction entry. External or interleaved session-file edits remain takeover errors.

Closes #90729
This commit is contained in:
Josh Lehman
2026-06-05 17:05:35 -07:00
committed by GitHub
parent a4f7e4cbb9
commit bbfe8ccaf6
5 changed files with 368 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/embedded-agent-runner/run/attempt.session-lock.test.ts
+270
View File
@@ -1,4 +1,5 @@
// Coverage for embedded attempt session-file ownership and write locks.
import { appendFileSync } from "node:fs";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
@@ -9,6 +10,7 @@ import {
runWithOwnedSessionTranscriptWritePublication,
withOwnedSessionTranscriptWrites,
} from "../../../config/sessions/transcript-write-context.js";
import { guardSessionManager } from "../../session-tool-result-guard-wrapper.js";
import {
SessionWriteLockStaleError,
SessionWriteLockTimeoutError,
@@ -17,6 +19,7 @@ import {
acquireSessionWriteLock,
resetSessionWriteLockStateForTest,
} from "../../session-write-lock.js";
import { SessionManager } from "../../sessions/session-manager.js";
import {
acquireEmbeddedAttemptSessionFileOwner,
createEmbeddedAttemptSessionLockController,
@@ -904,6 +907,273 @@ describe("embedded attempt session lock lifecycle", () => {
expect(controller.hasSessionTakeover()).toBe(false);
});
it("refreshes the prompt fence after an owned session manager compaction append", async () => {
const sessionFile = await createTempSessionFile();
const release = vi.fn(async () => {});
const acquireSessionWriteLockLocal2 = vi.fn(async () => ({ release }));
const controller = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: acquireSessionWriteLockLocal2,
lockOptions: { ...lockOptions, sessionFile },
});
const sessionManager = guardSessionManager(SessionManager.open(sessionFile), {
withCompactionPersistence: (append, validateAppend) =>
controller.withOwnedSessionFileWrite(append, validateAppend),
});
const firstKeptEntryId = sessionManager.appendMessage({
role: "user",
content: "old question",
timestamp: 1,
});
sessionManager.appendMessage({
role: "assistant",
content: [{ type: "text", text: "old answer" }],
api: "messages",
provider: "openclaw",
model: "session-lock-test",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: 2,
});
await controller.releaseForPrompt();
sessionManager.appendCompaction("threshold summary", firstKeptEntryId, 160_001);
await expect(controller.withSessionWriteLock(() => "finalize")).resolves.toBe("finalize");
expect(controller.hasSessionTakeover()).toBe(false);
});
it("still rejects unowned external compaction appends before the prompt stream lock is reacquired", async () => {
const sessionFile = await createTempSessionFile();
const release = vi.fn(async () => {});
const acquireSessionWriteLockLocal1 = vi.fn(async () => ({ release }));
const controller = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: acquireSessionWriteLockLocal1,
lockOptions: { ...lockOptions, sessionFile },
});
await controller.releaseForPrompt();
await fs.appendFile(
sessionFile,
JSON.stringify({
type: "compaction",
id: "external-compaction",
parentId: "session",
timestamp: new Date().toISOString(),
summary: "external summary",
firstKeptEntryId: "session",
tokensBefore: 160_001,
}) + "\n",
"utf8",
);
await expect(controller.withSessionWriteLock(() => "finalize")).rejects.toBeInstanceOf(
EmbeddedAttemptSessionTakeoverError,
);
expect(controller.hasSessionTakeover()).toBe(true);
});
it("still rejects an external edit that happens before an owned session manager compaction append", async () => {
const sessionFile = await createTempSessionFile();
const release = vi.fn(async () => {});
const acquireSessionWriteLockLocal0 = vi.fn(async () => ({ release }));
const controller = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: acquireSessionWriteLockLocal0,
lockOptions: { ...lockOptions, sessionFile },
});
const sessionManager = guardSessionManager(SessionManager.open(sessionFile), {
withCompactionPersistence: (append, validateAppend) =>
controller.withOwnedSessionFileWrite(append, validateAppend),
});
const firstKeptEntryId = sessionManager.appendMessage({
role: "user",
content: "old question",
timestamp: 1,
});
sessionManager.appendMessage({
role: "assistant",
content: [{ type: "text", text: "old answer" }],
api: "messages",
provider: "openclaw",
model: "session-lock-test",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: 2,
});
await controller.releaseForPrompt();
await fs.appendFile(sessionFile, '{"type":"message","id":"external-edit"}\n', "utf8");
sessionManager.appendCompaction("threshold summary", firstKeptEntryId, 160_001);
await expect(controller.withSessionWriteLock(() => "finalize")).rejects.toBeInstanceOf(
EmbeddedAttemptSessionTakeoverError,
);
expect(controller.hasSessionTakeover()).toBe(true);
});
it("still rejects an external edit interleaved inside an owned session manager compaction append", async () => {
const sessionFile = await createTempSessionFile();
const release = vi.fn(async () => {});
const acquireSessionWriteLockLocal30 = vi.fn(async () => ({ release }));
const controller = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: acquireSessionWriteLockLocal30,
lockOptions: { ...lockOptions, sessionFile },
});
const sessionManager = guardSessionManager(SessionManager.open(sessionFile), {
withCompactionPersistence: (append, validateAppend) =>
controller.withOwnedSessionFileWrite(() => {
appendFileSync(sessionFile, '{"type":"message","id":"external-edit"}\n', "utf8");
return append();
}, validateAppend),
});
const firstKeptEntryId = sessionManager.appendMessage({
role: "user",
content: "old question",
timestamp: 1,
});
sessionManager.appendMessage({
role: "assistant",
content: [{ type: "text", text: "old answer" }],
api: "messages",
provider: "openclaw",
model: "session-lock-test",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: 2,
});
await controller.releaseForPrompt();
sessionManager.appendCompaction("threshold summary", firstKeptEntryId, 160_001);
await expect(controller.withSessionWriteLock(() => "finalize")).rejects.toBeInstanceOf(
EmbeddedAttemptSessionTakeoverError,
);
expect(controller.hasSessionTakeover()).toBe(true);
});
it("allows owned session manager compaction after a later controller advances the prompt fence", async () => {
const sessionFile = await createTempSessionFile();
const firstController = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: vi.fn(async () => ({ release: vi.fn(async () => {}) })),
lockOptions: { ...lockOptions, sessionFile },
});
await firstController.releaseForPrompt();
await firstController.dispose();
const release = vi.fn(async () => {});
const acquireSessionWriteLockLocal29 = vi.fn(async () => ({ release }));
const secondController = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: acquireSessionWriteLockLocal29,
lockOptions: { ...lockOptions, sessionFile },
});
const sessionManager = guardSessionManager(SessionManager.open(sessionFile), {
withCompactionPersistence: (append, validateAppend) =>
secondController.withOwnedSessionFileWrite(append, validateAppend),
});
const firstKeptEntryId = await secondController.withSessionWriteLock(() => {
const entryId = sessionManager.appendMessage({
role: "user",
content: "new question",
timestamp: 1,
});
sessionManager.appendMessage({
role: "assistant",
content: [{ type: "text", text: "new answer" }],
api: "messages",
provider: "openclaw",
model: "session-lock-test",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: 2,
});
return entryId;
});
await secondController.releaseForPrompt();
sessionManager.appendCompaction("threshold summary", firstKeptEntryId, 160_001);
await expect(secondController.withSessionWriteLock(() => "finalize")).resolves.toBe("finalize");
expect(secondController.hasSessionTakeover()).toBe(false);
});
it("allows owned session manager compaction after another controller publishes an owned write", async () => {
const sessionFile = await createTempSessionFile();
const firstController = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: vi.fn(async () => ({ release: vi.fn(async () => {}) })),
lockOptions: { ...lockOptions, sessionFile },
});
const sessionManager = guardSessionManager(SessionManager.open(sessionFile), {
withCompactionPersistence: (append, validateAppend) =>
firstController.withOwnedSessionFileWrite(append, validateAppend),
});
const firstKeptEntryId = sessionManager.appendMessage({
role: "user",
content: "old question",
timestamp: 1,
});
sessionManager.appendMessage({
role: "assistant",
content: [{ type: "text", text: "old answer" }],
api: "messages",
provider: "openclaw",
model: "session-lock-test",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: 2,
});
await firstController.releaseForPrompt();
const secondController = await createEmbeddedAttemptSessionLockController({
acquireSessionWriteLock: vi.fn(async () => ({ release: vi.fn(async () => {}) })),
lockOptions: { ...lockOptions, sessionFile },
});
await secondController.releaseForPrompt();
await secondController.withSessionWriteLock(
async () => {
await fs.appendFile(sessionFile, '{"type":"message","id":"owned-other"}\n', "utf8");
},
{ publishOwnedWrite: true },
);
sessionManager.appendCompaction("threshold summary", firstKeptEntryId, 160_001);
await expect(firstController.withSessionWriteLock(() => "finalize")).resolves.toBe("finalize");
expect(firstController.hasSessionTakeover()).toBe(false);
});
it("allows post-prompt writes after the prompt context publishes an owned transcript write", async () => {
const sessionFile = await createTempSessionFile();
const releases: string[] = [];
src/agents/embedded-agent-runner/run/attempt.session-lock.ts
+60 -1
View File
@@ -2,7 +2,7 @@
* Coordinates embedded-attempt session ownership, takeover, and prompt locks.
*/
import { AsyncLocalStorage } from "node:async_hooks";
import { statSync } from "node:fs";
import { readFileSync, statSync } from "node:fs";
import fs from "node:fs/promises";
import { isDeepStrictEqual } from "node:util";
import { normalizeStringEntries } from "@openclaw/normalization-core/string-normalization";
@@ -29,6 +29,8 @@ type SessionWriteLockRunOptions = {
publishOwnedWrite?: boolean;
};
type SessionFileWriteAppendValidator<T> = (result: T, appendedText: string) => boolean;
type SessionWithAgentPrompt = {
agent?: {
streamFn?: PromptReleaseStreamFn;
@@ -575,6 +577,10 @@ export type EmbeddedAttemptSessionLockController = {
releaseForPrompt(): Promise<void>;
releaseHeldLockForAbort(): Promise<void>;
refreshAfterOwnedSessionWrite(): void;
withOwnedSessionFileWrite<T>(
run: () => T,
validateAppend?: SessionFileWriteAppendValidator<T>,
): T;
reacquireAfterPrompt(): Promise<void>;
waitForSessionEvents(session: unknown): Promise<void>;
withSessionWriteLock<T>(
@@ -787,6 +793,42 @@ export async function createEmbeddedAttemptSessionLockController(params: {
}
}
// Synchronous append paths cannot await withSessionWriteLock. Only publish
// their post-write fingerprint when the pre-write state was already trusted.
function publishOwnedSessionFileFenceSync<T>(write: {
beforeWrite: SessionFileFingerprint;
result: T;
beforeText?: string;
validateAppend?: SessionFileWriteAppendValidator<T>;
}): void {
if (takeoverDetected) {
return;
}
const fingerprint = readSessionFileFingerprintSync(params.lockOptions.sessionFile);
const beforeWriteIsTrusted =
(fenceActive && sameSessionFileFingerprint(fenceFingerprint, write.beforeWrite)) ||
isTrustedSessionFileState(sessionFileFenceKey, write.beforeWrite);
if (sameSessionFileFingerprint(write.beforeWrite, fingerprint) || !beforeWriteIsTrusted) {
return;
}
if (write.validateAppend) {
const afterText = readFileSync(params.lockOptions.sessionFile, "utf8");
if (
write.beforeText === undefined ||
!afterText.startsWith(write.beforeText) ||
!write.validateAppend(write.result, afterText.slice(write.beforeText.length))
) {
return;
}
}
const generation = recordOwnedSessionFileWrite(sessionFileFenceKey, fingerprint);
if (fenceActive) {
fenceFingerprint = fingerprint;
fenceSnapshot = { fingerprint };
fenceGeneration = generation;
}
}
const noopLock: SessionLock = { release: async () => {} };
async function releaseHeldLockWithFence(): Promise<void> {
@@ -913,6 +955,23 @@ export async function createEmbeddedAttemptSessionLockController(params: {
fenceSnapshot = { fingerprint: fenceFingerprint };
}
},
withOwnedSessionFileWrite<T>(
run: () => T,
validateAppend?: SessionFileWriteAppendValidator<T>,
): T {
const beforeWrite = readSessionFileFingerprintSync(params.lockOptions.sessionFile);
const beforeText = validateAppend
? readFileSync(params.lockOptions.sessionFile, "utf8")
: undefined;
const result = run();
publishOwnedSessionFileFenceSync({
beforeWrite,
result,
...(beforeText !== undefined ? { beforeText } : {}),
...(validateAppend ? { validateAppend } : {}),
});
return result;
},
async reacquireAfterPrompt(): Promise<void> {
await waitForHeldLockDrain();
if (takeoverDetected || heldLock) {
src/agents/embedded-agent-runner/run/attempt.ts
+2
View File
@@ -2005,6 +2005,8 @@ export async function runEmbeddedAttempt(
onMessagePersisted: () => {
sessionLockController.refreshAfterOwnedSessionWrite();
},
withCompactionPersistence: (append, validateAppend) =>
sessionLockController.withOwnedSessionFileWrite(append, validateAppend),
onUserMessagePersisted: (message) => {
params.onUserMessagePersisted?.(message);
},
src/agents/session-tool-result-guard-wrapper.ts
+5
View File
@@ -49,6 +49,10 @@ export function guardSessionManager(
message: Extract<AgentMessage, { role: "user" }>,
) => void | Promise<void>;
onMessagePersisted?: (message: AgentMessage) => void | Promise<void>;
withCompactionPersistence?: (
append: () => string,
validateAppend: (entryId: string, appendedText: string) => boolean,
) => string;
onAssistantErrorMessagePersisted?: (
message: Extract<AgentMessage, { role: "assistant" }>,
) => void | Promise<void>;
@@ -140,6 +144,7 @@ export function guardSessionManager(
suppressTranscriptOnlyAssistantPersistence: opts?.suppressTranscriptOnlyAssistantPersistence,
suppressAssistantErrorPersistence: opts?.suppressAssistantErrorPersistence,
onMessagePersisted: opts?.onMessagePersisted,
withCompactionPersistence: opts?.withCompactionPersistence,
onUserMessagePersisted: opts?.onUserMessagePersisted,
onAssistantErrorMessagePersisted: opts?.onAssistantErrorMessagePersisted,
});
src/agents/session-tool-result-guard.ts
+31
View File
@@ -58,11 +58,28 @@ function resolveMaxToolResultChars(opts?: { maxToolResultChars?: number }): numb
}
type UserAgentMessage = Extract<AgentMessage, { role: "user" }>;
type CompactionAppendValidator = (entryId: string, appendedText: string) => boolean;
function isUserAgentMessage(message: AgentMessage): message is UserAgentMessage {
return message.role === "user";
}
function isExpectedCompactionAppend(entryId: string, appendedText: string): boolean {
const lines = appendedText
.trimEnd()
.split("\n")
.filter((line) => line.length > 0);
if (lines.length !== 1) {
return false;
}
try {
const entry = JSON.parse(lines[0]) as { type?: unknown; id?: unknown };
return entry.type === "compaction" && entry.id === entryId;
} catch {
return false;
}
}
type TranscriptSeqByEntryId = Map<string, number>;
function resolveEntryTranscriptSeq(
@@ -568,6 +585,10 @@ export function installSessionToolResultGuard(
message: Extract<AgentMessage, { role: "user" }>,
) => void | Promise<void>;
onMessagePersisted?: (message: AgentMessage) => void | Promise<void>;
withCompactionPersistence?: (
append: () => string,
validateAppend: CompactionAppendValidator,
) => string;
onAssistantErrorMessagePersisted?: (
message: Extract<AgentMessage, { role: "assistant" }>,
) => void | Promise<void>;
@@ -625,6 +646,15 @@ export function installSessionToolResultGuard(
}),
};
};
const originalAppendCompaction = sessionManager.appendCompaction.bind(sessionManager);
const guardedAppendCompaction = ((
...args: Parameters<SessionManager["appendCompaction"]>
): string => {
const append = () => originalAppendCompaction(...args);
return opts?.withCompactionPersistence
? opts.withCompactionPersistence(append, isExpectedCompactionAppend)
: append();
}) as SessionManager["appendCompaction"];
/**
* Run the before_message_write hook. Returns the (possibly modified) message,
@@ -821,6 +851,7 @@ export function installSessionToolResultGuard(
// Monkey-patch appendMessage with our guarded version.
sessionManager.appendMessage = guardedAppend as SessionManager["appendMessage"];
sessionManager.appendCompaction = guardedAppendCompaction;
return {
flushPendingToolResults,