Skip to content

Commit a0b16f3

Browse files
Alix-007Alix-007
authored
fix(sessions): cache validated transcripts across turns (#90412)
Avoid repeated full JSONL parsing and cloning on every embedded-agent turn by keeping a bounded, validated transcript cache and advancing repair incrementally. The final implementation preserves lock ownership and exact fingerprint validation, publishes only verified writes, handles header rewrites and unterminated JSONL safely, and adds focused regression coverage. Fixes #83943. Co-authored-by: Alix-007 <[email protected]>
1 parent cc000e0 commit a0b16f3

13 files changed

Lines changed: 2416 additions & 159 deletions

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

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,65 @@ function cloneBigIntStatWith(
6666
}
6767

6868
describe("embedded attempt session lock lifecycle", () => {
69+
it("recognizes an unchanged session file trusted by the previous prompt release", async () => {
70+
const sessionFile = await createTempSessionFile();
71+
const options = { ...lockOptions, sessionFile };
72+
const first = await createEmbeddedAttemptSessionLockController({
73+
acquireSessionWriteLock,
74+
lockOptions: options,
75+
});
76+
77+
expect(await first.readTrustedCurrentSessionFileSnapshot()).toBeUndefined();
78+
await first.releaseForPrompt();
79+
await first.dispose();
80+
81+
const second = await createEmbeddedAttemptSessionLockController({
82+
acquireSessionWriteLock,
83+
lockOptions: options,
84+
});
85+
expect(await second.readTrustedCurrentSessionFileSnapshot()).toBeDefined();
86+
await second.dispose();
87+
});
88+
89+
it("does not trust a session file changed after the previous prompt release", async () => {
90+
const sessionFile = await createTempSessionFile();
91+
const options = { ...lockOptions, sessionFile };
92+
const first = await createEmbeddedAttemptSessionLockController({
93+
acquireSessionWriteLock,
94+
lockOptions: options,
95+
});
96+
await first.releaseForPrompt();
97+
await first.dispose();
98+
await fs.appendFile(sessionFile, '{"type":"message","id":"external"}\n', "utf8");
99+
100+
const second = await createEmbeddedAttemptSessionLockController({
101+
acquireSessionWriteLock,
102+
lockOptions: options,
103+
});
104+
expect(await second.readTrustedCurrentSessionFileSnapshot()).toBeUndefined();
105+
await second.dispose();
106+
});
107+
108+
it("publishes a known owned append for the next attempt", async () => {
109+
const sessionFile = await createTempSessionFile();
110+
const options = { ...lockOptions, sessionFile };
111+
const first = await createEmbeddedAttemptSessionLockController({
112+
acquireSessionWriteLock,
113+
lockOptions: options,
114+
});
115+
await fs.appendFile(sessionFile, '{"type":"message","id":"owned"}\n', "utf8");
116+
first.refreshAfterOwnedSessionWrite();
117+
await first.releaseForPrompt();
118+
await first.dispose();
119+
120+
const second = await createEmbeddedAttemptSessionLockController({
121+
acquireSessionWriteLock,
122+
lockOptions: options,
123+
});
124+
expect(await second.readTrustedCurrentSessionFileSnapshot()).toBeDefined();
125+
await second.dispose();
126+
});
127+
69128
it("serializes embedded attempts that share a session file owner", async () => {
70129
const sessionFile = await createTempSessionFile();
71130
const firstOwner = await acquireEmbeddedAttemptSessionFileOwner({ sessionFile });
@@ -1045,6 +1104,90 @@ describe("embedded attempt session lock lifecycle", () => {
10451104
expect(release).toHaveBeenCalledTimes(3);
10461105
});
10471106

1107+
it("authorizes entry-cache advancement only under the exact trusted file lock", async () => {
1108+
const sessionFile = await createTempSessionFile();
1109+
const release = vi.fn(async () => {});
1110+
const acquireSessionWriteLockLocal = vi.fn(async () => ({ release }));
1111+
const controller = await createEmbeddedAttemptSessionLockController({
1112+
acquireSessionWriteLock: acquireSessionWriteLockLocal,
1113+
lockOptions: { ...lockOptions, sessionFile },
1114+
});
1115+
1116+
await controller.releaseForPrompt();
1117+
const stat = await fs.stat(sessionFile, { bigint: true });
1118+
const snapshot = {
1119+
dev: stat.dev,
1120+
ino: stat.ino,
1121+
size: stat.size,
1122+
mtimeNs: stat.mtimeNs,
1123+
ctimeNs: stat.ctimeNs,
1124+
};
1125+
1126+
expect(controller.canAdvanceSessionEntryCache(snapshot)).toBe(false);
1127+
await expect(
1128+
controller.withSessionWriteLock(() => {
1129+
expect(controller.canAdvanceSessionEntryCache(snapshot)).toBe(true);
1130+
return "locked";
1131+
}),
1132+
).resolves.toBe("locked");
1133+
expect(controller.canAdvanceSessionEntryCache(snapshot)).toBe(false);
1134+
});
1135+
1136+
it("publishes an exact owned snapshot only while the write lock is active", async () => {
1137+
const sessionFile = await createTempSessionFile();
1138+
const controller = await createEmbeddedAttemptSessionLockController({
1139+
acquireSessionWriteLock: vi.fn(async () => ({
1140+
release: vi.fn(async () => {}),
1141+
})),
1142+
lockOptions: { ...lockOptions, sessionFile },
1143+
});
1144+
const readSnapshot = async () => {
1145+
const stat = await fs.stat(sessionFile, { bigint: true });
1146+
return {
1147+
dev: stat.dev,
1148+
ino: stat.ino,
1149+
size: stat.size,
1150+
mtimeNs: stat.mtimeNs,
1151+
ctimeNs: stat.ctimeNs,
1152+
};
1153+
};
1154+
const initialSnapshot = await readSnapshot();
1155+
1156+
expect(controller.publishOwnedSessionFileSnapshot(initialSnapshot)).toBe(false);
1157+
await controller.withSessionWriteLock(async () => {
1158+
expect(controller.publishOwnedSessionFileSnapshot(initialSnapshot)).toBe(true);
1159+
await fs.appendFile(sessionFile, '{"type":"message","id":"owned"}\n', "utf8");
1160+
expect(controller.publishOwnedSessionFileSnapshot(initialSnapshot)).toBe(false);
1161+
expect(controller.publishOwnedSessionFileSnapshot(await readSnapshot())).toBe(true);
1162+
});
1163+
expect(controller.publishOwnedSessionFileSnapshot(await readSnapshot())).toBe(false);
1164+
await controller.dispose();
1165+
});
1166+
1167+
it("publishes only an unchanged repair-validated snapshot while retaining the lock", async () => {
1168+
const sessionFile = await createTempSessionFile();
1169+
const acquireSessionWriteLockLocal = vi.fn(async () => ({
1170+
release: vi.fn(async () => {}),
1171+
}));
1172+
const controller = await createEmbeddedAttemptSessionLockController({
1173+
acquireSessionWriteLock: acquireSessionWriteLockLocal,
1174+
lockOptions: { ...lockOptions, sessionFile },
1175+
});
1176+
const stat = await fs.stat(sessionFile, { bigint: true });
1177+
const snapshot = {
1178+
dev: stat.dev,
1179+
ino: stat.ino,
1180+
size: stat.size,
1181+
mtimeNs: stat.mtimeNs,
1182+
ctimeNs: stat.ctimeNs,
1183+
};
1184+
1185+
expect(controller.publishValidatedSessionFileSnapshot(snapshot)).toBe(true);
1186+
await fs.appendFile(sessionFile, '{"type":"message","id":"external"}\n', "utf8");
1187+
expect(controller.publishValidatedSessionFileSnapshot(snapshot)).toBe(false);
1188+
await controller.dispose();
1189+
});
1190+
10481191
it("refreshes the prompt fence after an owned session manager append", async () => {
10491192
const sessionFile = await createTempSessionFile();
10501193
const release = vi.fn(async () => {});

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

Lines changed: 82 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,10 @@ import { createReadStream, readFileSync, statSync } from "node:fs";
77
import fs from "node:fs/promises";
88
import { isDeepStrictEqual } from "node:util";
99
import { normalizeStringEntries } from "@openclaw/normalization-core/string-normalization";
10-
import { withOwnedSessionTranscriptWrites } from "../../../config/sessions/transcript-write-context.js";
10+
import {
11+
type OwnedSessionTranscriptCacheSnapshot,
12+
withOwnedSessionTranscriptWrites,
13+
} from "../../../config/sessions/transcript-write-context.js";
1114
import { resolveGlobalSingleton } from "../../../shared/global-singleton.js";
1215
import { isSessionWriteLockAcquireError } from "../../session-write-lock-error.js";
1316
import type { acquireSessionWriteLock } from "../../session-write-lock.js";
@@ -53,6 +56,8 @@ type SessionFileFingerprint =
5356
ctimeNs: bigint;
5457
};
5558

59+
export type TrustedSessionFileSnapshot = Extract<SessionFileFingerprint, { exists: true }>;
60+
5661
const TRANSCRIPT_ONLY_OPENCLAW_ASSISTANT_MODELS = new Set(["delivery-mirror", "gateway-injected"]);
5762
const MAX_BENIGN_SESSION_FENCE_ADVANCE_BYTES = 1024 * 1024;
5863
const MAX_BENIGN_SESSION_FENCE_REWRITE_BYTES = 8 * 1024 * 1024;
@@ -655,6 +660,10 @@ export class EmbeddedAttemptSessionTakeoverError extends Error {
655660
}
656661

657662
export type EmbeddedAttemptSessionLockController = {
663+
canAdvanceSessionEntryCache(snapshot: OwnedSessionTranscriptCacheSnapshot): boolean;
664+
publishOwnedSessionFileSnapshot(snapshot: OwnedSessionTranscriptCacheSnapshot): boolean;
665+
publishValidatedSessionFileSnapshot(snapshot: OwnedSessionTranscriptCacheSnapshot): boolean;
666+
readTrustedCurrentSessionFileSnapshot(): Promise<TrustedSessionFileSnapshot | undefined>;
658667
releaseForPrompt(): Promise<void>;
659668
releaseHeldLockForAbort(): Promise<void>;
660669
refreshAfterOwnedSessionWrite(): void;
@@ -1037,17 +1046,82 @@ export async function createEmbeddedAttemptSessionLockController(params: {
10371046
}
10381047

10391048
return {
1049+
canAdvanceSessionEntryCache(snapshot: OwnedSessionTranscriptCacheSnapshot): boolean {
1050+
if (takeoverDetected || activeWriteLock.getStore()?.active !== true) {
1051+
return false;
1052+
}
1053+
const fingerprint: SessionFileFingerprint = { exists: true, ...snapshot };
1054+
return (
1055+
(fenceActive && sameSessionFileFingerprint(fenceFingerprint, fingerprint)) ||
1056+
isTrustedSessionFileState(sessionFileFenceKey, fingerprint)
1057+
);
1058+
},
1059+
publishOwnedSessionFileSnapshot(snapshot: OwnedSessionTranscriptCacheSnapshot): boolean {
1060+
if (takeoverDetected || activeWriteLock.getStore()?.active !== true) {
1061+
return false;
1062+
}
1063+
const fingerprint: SessionFileFingerprint = { exists: true, ...snapshot };
1064+
const current = readSessionFileFingerprintSync(params.lockOptions.sessionFile);
1065+
if (!sameSessionFileFingerprint(fingerprint, current)) {
1066+
return false;
1067+
}
1068+
const generation = recordOwnedSessionFileWrite(sessionFileFenceKey, current);
1069+
if (fenceActive) {
1070+
fenceFingerprint = current;
1071+
fenceSnapshot = { fingerprint: current };
1072+
fenceGeneration = generation;
1073+
}
1074+
return true;
1075+
},
1076+
publishValidatedSessionFileSnapshot(snapshot: OwnedSessionTranscriptCacheSnapshot): boolean {
1077+
if (takeoverDetected || !heldLock || heldLockDraining) {
1078+
return false;
1079+
}
1080+
const fingerprint: SessionFileFingerprint = { exists: true, ...snapshot };
1081+
const current = readSessionFileFingerprintSync(params.lockOptions.sessionFile);
1082+
if (!sameSessionFileFingerprint(fingerprint, current)) {
1083+
return false;
1084+
}
1085+
fenceGeneration = recordTrustedSessionFileState(sessionFileFenceKey, current);
1086+
if (fenceActive) {
1087+
fenceFingerprint = current;
1088+
fenceSnapshot = { fingerprint: current };
1089+
}
1090+
return true;
1091+
},
1092+
async readTrustedCurrentSessionFileSnapshot(): Promise<TrustedSessionFileSnapshot | undefined> {
1093+
const fingerprint = await readSessionFileFingerprint(params.lockOptions.sessionFile);
1094+
return fingerprint.exists && isTrustedSessionFileState(sessionFileFenceKey, fingerprint)
1095+
? fingerprint
1096+
: undefined;
1097+
},
10401098
async releaseForPrompt(): Promise<void> {
10411099
await releaseHeldLockWithFence();
10421100
},
10431101
async releaseHeldLockForAbort(): Promise<void> {
10441102
await releaseHeldLockWithFence();
10451103
},
10461104
refreshAfterOwnedSessionWrite(): void {
1047-
if (fenceActive && !takeoverDetected) {
1048-
fenceFingerprint = readSessionFileFingerprintSync(params.lockOptions.sessionFile);
1049-
fenceSnapshot = { fingerprint: fenceFingerprint };
1105+
if (takeoverDetected) {
1106+
return;
10501107
}
1108+
const beforeWrite = fenceFingerprint;
1109+
const fingerprint = readSessionFileFingerprintSync(params.lockOptions.sessionFile);
1110+
if (!fenceActive) {
1111+
// User-message persistence occurs before the prompt fence activates.
1112+
// The retained session lock owns that write, so publish its exact state
1113+
// for the next attempt before release establishes the active fence.
1114+
fenceGeneration = recordTrustedSessionFileState(sessionFileFenceKey, fingerprint);
1115+
return;
1116+
}
1117+
if (
1118+
!sameSessionFileFingerprint(beforeWrite, fingerprint) &&
1119+
isTrustedSessionFileState(sessionFileFenceKey, beforeWrite ?? { exists: false })
1120+
) {
1121+
fenceGeneration = recordOwnedSessionFileWrite(sessionFileFenceKey, fingerprint);
1122+
}
1123+
fenceFingerprint = fingerprint;
1124+
fenceSnapshot = { fingerprint };
10511125
},
10521126
withOwnedSessionFileWrite<T>(
10531127
run: () => T,
@@ -1175,6 +1249,8 @@ export function installPromptSubmissionLockRelease(params: {
11751249
run: () => Promise<T> | T,
11761250
options?: SessionWriteLockRunOptions,
11771251
) => Promise<T>;
1252+
canAdvanceSessionEntryCache?: (snapshot: OwnedSessionTranscriptCacheSnapshot) => boolean;
1253+
publishSessionFileSnapshot?: (snapshot: OwnedSessionTranscriptCacheSnapshot) => boolean;
11781254
}): void {
11791255
const agent = (params.session as SessionWithAgentPrompt).agent;
11801256
if (typeof agent?.streamFn !== "function") {
@@ -1195,6 +1271,8 @@ export function installPromptSubmissionLockRelease(params: {
11951271
sessionFile: params.sessionFile,
11961272
sessionKey: params.sessionKey,
11971273
withSessionWriteLock: params.withSessionWriteLock,
1274+
canAdvanceSessionEntryCache: params.canAdvanceSessionEntryCache,
1275+
publishSessionFileSnapshot: params.publishSessionFileSnapshot,
11981276
},
11991277
async () => await originalStreamFn(...args),
12001278
);

‎src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -465,7 +465,7 @@ vi.mock("../tool-schema-runtime.js", () => ({
465465
}));
466466

467467
vi.mock("../../session-file-repair.js", () => ({
468-
repairSessionFileIfNeeded: async () => {},
468+
repairSessionFileIfNeeded: async () => ({ repaired: false, droppedLines: 0 }),
469469
}));
470470

471471
vi.mock("../session-manager-cache.js", () => ({

0 commit comments

Comments
 (0)