ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(ai): invalidate stale compaction replay (#120786)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-08 17:29:17 -07:00
committed by GitHub
parent 2a7782674e
commit 56cdad5055
7 changed files with 201 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/agent-core/src/harness/session/session-context.test.ts
+91 -25
View File
@@ -1,3 +1,4 @@
import type { AssistantMessage, ProviderReplayState } from "@openclaw/llm-core";
import { describe, expect, it } from "vitest";
import type { SessionTreeEntry } from "../types.js";
import { buildSessionContext } from "./session.js";
@@ -14,15 +15,72 @@ function userEntry(id: string, parentId: string | null, content: string): Sessio
};
}
function assistantEntry(
id: string,
parentId: string | null,
content: string,
providerReplay?: ProviderReplayState,
): SessionTreeEntry {
return {
type: "message",
id,
parentId,
timestamp,
message: {
role: "assistant",
api: "openai-responses",
content: [{ type: "text", text: content }],
provider: "test-provider",
model: "test-model",
usage: {
input: 1,
output: 1,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 2,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.parse(timestamp),
...(providerReplay ? { providerReplay } : {}),
},
};
}
function replayState(type: string, data: string): ProviderReplayState {
return {
v: 1,
type,
data,
provider: "openai",
api: "openai-responses",
model: "gpt-5.6-luna",
baseUrlHash: "route-a",
};
}
describe("buildSessionContext", () => {
it("replays only the retained tail and newer entries after compaction", () => {
const retainedCheckpoint = replayState("openai-responses-compaction", "retained-checkpoint");
const retainedSuppression = replayState("openai-responses-compaction-suppression", "rejected");
const postBoundaryCheckpoint = replayState(
"openai-responses-compaction",
"post-boundary-checkpoint",
);
const entries: SessionTreeEntry[] = [
userEntry("old", null, "discarded"),
userEntry("kept", "old", "retained"),
assistantEntry("retained-checkpoint", "kept", "retained checkpoint", retainedCheckpoint),
assistantEntry(
"retained-suppression",
"retained-checkpoint",
"retained suppression",
retainedSuppression,
),
{
type: "model_change",
id: "model",
parentId: "kept",
parentId: "retained-suppression",
timestamp,
provider: "test-provider",
modelId: "test-model",
@@ -36,7 +94,13 @@ describe("buildSessionContext", () => {
firstKeptEntryId: "kept",
tokensBefore: 123,
},
userEntry("new", "compaction", "new turn"),
assistantEntry(
"post-checkpoint",
"compaction",
"post-boundary checkpoint",
postBoundaryCheckpoint,
),
userEntry("new", "post-checkpoint", "new turn"),
];
const context = buildSessionContext(entries);
@@ -48,16 +112,29 @@ describe("buildSessionContext", () => {
expect(context.messages.map((message) => message.role)).toEqual([
"compactionSummary",
"user",
"assistant",
"assistant",
"assistant",
"user",
]);
expect(context.messages).toMatchObject([
{ summary: "older context" },
{ content: "retained" },
{ content: [{ text: "retained checkpoint" }] },
{ content: [{ text: "retained suppression" }] },
{ content: [{ text: "post-boundary checkpoint" }] },
{ content: "new turn" },
]);
const assistants = context.messages.filter(
(message): message is AssistantMessage => message.role === "assistant",
);
expect(assistants[0]).not.toHaveProperty("providerReplay");
expect(assistants[1]?.providerReplay).toEqual(retainedSuppression);
expect(assistants[2]?.providerReplay).toEqual(postBoundaryCheckpoint);
});
it("treats the latest reset as a hard cut with a user/assistant-only kept tail", () => {
const retainedCheckpoint = replayState("openai-responses-compaction", "reset-checkpoint");
const entries: SessionTreeEntry[] = [
userEntry("discarded", null, "discarded"),
userEntry("kept-user", "discarded", "kept question"),
@@ -75,29 +152,7 @@ describe("buildSessionContext", () => {
timestamp: Date.parse(timestamp),
},
},
{
type: "message",
id: "kept-assistant",
parentId: "kept-tool",
timestamp,
message: {
role: "assistant",
api: "openai-responses",
content: [{ type: "text", text: "kept answer" }],
provider: "test-provider",
model: "test-model",
usage: {
input: 1,
output: 1,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 2,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.parse(timestamp),
},
},
assistantEntry("kept-assistant", "kept-tool", "kept answer", retainedCheckpoint),
{
type: "reset",
id: "reset",
@@ -117,6 +172,17 @@ describe("buildSessionContext", () => {
expect(JSON.stringify(context.messages)).toContain("new turn");
expect(JSON.stringify(context.messages)).not.toContain("discarded");
expect(JSON.stringify(context.messages)).not.toContain("hidden tool result");
const keptAssistant = context.messages.find(
(message): message is AssistantMessage => message.role === "assistant",
);
expect(keptAssistant).not.toHaveProperty("providerReplay");
expect(
(
keptAssistant as AssistantMessage & {
[key: symbol]: true | undefined;
}
)[Symbol.for("openclaw.sessionHistoryPrelude")],
).toBe(true);
});
it("lets the latest compaction shadow an earlier reset boundary", () => {
packages/agent-core/src/harness/session/session.ts
+19 -5
View File
@@ -1,3 +1,4 @@
import { stripOpenAIResponsesCompactionReplayCheckpoint } from "@openclaw/ai/transports";
import type { AgentMessage } from "../../types.js";
import {
asAgentMessage,
@@ -38,13 +39,23 @@ export function projectSessionEntryMessage(entry: SessionTreeEntry): AgentMessag
}
}
function appendContextMessage(messages: AgentMessage[], entry: SessionTreeEntry): void {
function stripStalePrefixReplay(message: AgentMessage): AgentMessage {
return message.role === "assistant"
? stripOpenAIResponsesCompactionReplayCheckpoint(message)
: message;
}
function appendContextMessage(
messages: AgentMessage[],
entry: SessionTreeEntry,
options?: { prefixWasRewritten?: boolean },
): void {
if (entry.type === "compaction" || (entry.type === "branch_summary" && !entry.summary)) {
return;
}
const message = projectSessionEntryMessage(entry);
if (message) {
messages.push(message);
messages.push(options?.prefixWasRewritten ? stripStalePrefixReplay(message) : message);
}
}
@@ -53,7 +64,9 @@ function appendResetKeptMessage(messages: AgentMessage[], entry: SessionTreeEntr
entry.type === "message" &&
(entry.message.role === "user" || entry.message.role === "assistant")
) {
const message = { ...entry.message } as AgentMessage & { [SESSION_HISTORY_PRELUDE]?: true };
const message = { ...stripStalePrefixReplay(entry.message) } as AgentMessage & {
[SESSION_HISTORY_PRELUDE]?: true;
};
Object.defineProperty(message, SESSION_HISTORY_PRELUDE, {
configurable: true,
enumerable: false,
@@ -91,7 +104,8 @@ export function buildSessionContext(pathEntries: SessionTreeEntry[]): SessionCon
}
const boundaryIdx = pathEntries.findIndex((entry) => entry.id === boundary.id);
// A reset kept tail mirrors the old cross-log replay contract: only user/assistant
// rows survive. Compaction keeps its existing richer retained-tail behavior.
// rows survive. Both retained-tail forms now follow rewritten prefixes, so
// prefix-bound checkpoints are stale.
let foundFirstKept = false;
for (const entry of pathEntries.slice(0, boundaryIdx)) {
if (entry.id === boundary.firstKeptEntryId) {
@@ -101,7 +115,7 @@ export function buildSessionContext(pathEntries: SessionTreeEntry[]): SessionCon
if (boundary.type === "reset") {
appendResetKeptMessage(messages, entry);
} else {
appendContextMessage(messages, entry);
appendContextMessage(messages, entry, { prefixWasRewritten: true });
}
}
}
packages/ai/src/transports/openai-responses-compaction-replay.test.ts
+23
View File
@@ -13,6 +13,7 @@ import {
} from "./openai-responses-compaction-replay.js";
import { stringifyRedactedEvent, stringifyRedactedPayload } from "./openai-responses-debug.js";
import { convertResponsesMessages } from "./openai-responses-replay-internal.js";
import { stripOpenAIResponsesCompactionReplayCheckpoint } from "./openai-responses-replay.js";
import {
processResponsesStream,
type OpenAIResponsesStreamEvent,
@@ -136,6 +137,28 @@ function responseMessage(id: string, text: string) {
}
describe("OpenAI Responses compaction replay", () => {
it("strips only exact compaction checkpoints with structural sharing", () => {
const unchanged = createOutput();
expect(stripOpenAIResponsesCompactionReplayCheckpoint(unchanged)).toBe(unchanged);
const checkpoint = createAssistant(
[{ type: "text", text: "checkpoint owner" }],
compactionState(),
);
const stripped = stripOpenAIResponsesCompactionReplayCheckpoint(checkpoint);
expect(stripped).not.toBe(checkpoint);
expect(stripped.content).toBe(checkpoint.content);
expect(stripped).not.toHaveProperty("providerReplay");
expect(checkpoint.providerReplay).toEqual(compactionState());
const suppression = createOutput();
suppressOpenAIResponsesCompaction(suppression, model, replayIdentity);
expect(stripOpenAIResponsesCompactionReplayCheckpoint(suppression)).toBe(suppression);
const unrelated = createAssistant([], compactionState(model, { type: "future-replay" }));
expect(stripOpenAIResponsesCompactionReplayCheckpoint(unrelated)).toBe(unrelated);
});
it("persists a streamed compaction output item as opaque provider replay state", async () => {
const output = createOutput();
packages/ai/src/transports/openai-responses-compaction-replay.ts
+12
View File
@@ -22,6 +22,18 @@ type OpenAIResponsesCompactionSuppressionState = ProviderReplayState & {
baseUrlHash: string;
};
/** Removes prefix-bound checkpoint state while preserving route-scoped suppression state. */
export function stripOpenAIResponsesCompactionReplayCheckpoint(
message: AssistantMessage,
): AssistantMessage {
if (message.providerReplay?.type !== OPENAI_RESPONSES_COMPACTION_REPLAY_TYPE) {
return message;
}
const replaySafeMessage = { ...message };
delete replaySafeMessage.providerReplay;
return replaySafeMessage;
}
function hashOptionalReplayContextValue(value: string | undefined): string | undefined {
const normalized = value?.trim();
return normalized ? shortHash(normalized) : undefined;
packages/ai/src/transports/openai-responses-replay.ts
+2
View File
@@ -1,3 +1,5 @@
export { stripOpenAIResponsesCompactionReplayCheckpoint } from "./openai-responses-compaction-replay.js";
/** Resolves the assistant message id that can be replayed to OpenAI Responses. */
export function resolveReplayableResponsesMessageId(params: {
replayResponsesItemIds: boolean;
src/agents/embedded-agent-runner/tool-result-truncation.test.ts
+38 -5
View File
@@ -1438,6 +1438,18 @@ describe("truncateOversizedToolResultsInSession", () => {
await appendTranscriptMessage(scope, {
message: makeUserMessage("run tools"),
});
const staleCheckpointReplay = {
v: 1,
type: "openai-responses-compaction",
data: "stale-checkpoint",
provider: "openai",
api: "openai-responses",
model: "gpt-5.2",
baseUrlHash: "ozhevd1smnk8s",
} satisfies NonNullable<AssistantMessage["providerReplay"]>;
const preBoundaryCheckpointOwner = makeAssistantMessage("pre-boundary checkpoint owner");
preBoundaryCheckpointOwner.providerReplay = staleCheckpointReplay;
await appendTranscriptMessage(scope, { message: preBoundaryCheckpointOwner });
const medium = "alpha beta gamma delta epsilon ".repeat(600);
const firstToolResult = await appendTranscriptMessage(scope, {
message: makeToolResult(medium, "call_1"),
@@ -1448,6 +1460,17 @@ describe("truncateOversizedToolResultsInSession", () => {
const thirdToolResult = await appendTranscriptMessage(scope, {
message: makeToolResult(medium, "call_3"),
});
const staleCheckpointOwner = makeAssistantMessage("stale checkpoint owner");
staleCheckpointOwner.providerReplay = staleCheckpointReplay;
await appendTranscriptMessage(scope, { message: staleCheckpointOwner });
const suppressionReplay = {
...staleCheckpointReplay,
type: "openai-responses-compaction-suppression",
data: "rejected",
} satisfies NonNullable<AssistantMessage["providerReplay"]>;
const suppressionOwner = makeAssistantMessage("suppression owner");
suppressionOwner.providerReplay = suppressionReplay;
await appendTranscriptMessage(scope, { message: suppressionOwner });
const listener = vi.fn();
const cleanup = onInternalSessionTranscriptUpdate(listener);
@@ -1489,16 +1512,26 @@ describe("truncateOversizedToolResultsInSession", () => {
.map(getFirstToolResultText);
expect(originalToolResultTexts).toEqual([medium, medium, medium]);
const toolResultTexts = SessionManager.open(scope)
const activeMessages = SessionManager.open(scope)
.getBranch()
.flatMap((entry) =>
entry.type === "message" && entry.message.role === "toolResult"
? [getFirstToolResultText(entry.message as ToolResultMessage)]
: [],
.flatMap((entry) => (entry.type === "message" ? [entry.message] : []));
const toolResultTexts = activeMessages.flatMap((message) =>
message.role === "toolResult" ? [getFirstToolResultText(message as ToolResultMessage)] : [],
);
const findAssistant = (text: string) =>
activeMessages.find(
(message): message is AssistantMessage =>
message.role === "assistant" &&
message.content.some((block) => block.type === "text" && block.text === text),
);
expect(toolResultTexts.some((text) => text.includes("truncated"))).toBe(true);
expect(toolResultTexts.join("").length).toBeLessThan(medium.length * 3);
expect(findAssistant("pre-boundary checkpoint owner")?.providerReplay).toEqual(
staleCheckpointReplay,
);
expect(findAssistant("stale checkpoint owner")?.providerReplay).toBeUndefined();
expect(findAssistant("suppression owner")?.providerReplay).toEqual(suppressionReplay);
});
it("reuses frozen provider projection bytes on the recovery branch", async () => {
src/agents/embedded-agent-runner/transcript-rewrite.ts
+16 -2
View File
@@ -1,4 +1,5 @@
/** Rewrites transcript entries by branching and re-appending the active suffix. */
import { stripOpenAIResponsesCompactionReplayCheckpoint } from "@openclaw/ai/transports";
import type {
TranscriptRewriteReplacement,
TranscriptRewriteResult,
@@ -10,6 +11,12 @@ import { SessionManager } from "../sessions/index.js";
type SessionManagerLike = ReturnType<typeof SessionManager.open>;
type SessionBranchEntry = ReturnType<SessionManagerLike["getBranch"]>[number];
function stripStalePrefixReplay(message: AgentMessage): AgentMessage {
return message.role === "assistant"
? stripOpenAIResponsesCompactionReplayCheckpoint(message)
: message;
}
function estimateMessageBytes(message: AgentMessage): number {
return Buffer.byteLength(JSON.stringify(message), "utf8");
}
@@ -56,7 +63,9 @@ function appendBranchEntry(params: {
}): string {
const { sessionManager, entry, rewrittenEntryIds, appendMessage } = params;
if (entry.type === "message") {
return appendMessage(entry.message as Parameters<typeof sessionManager.appendMessage>[0]);
return appendMessage(
stripStalePrefixReplay(entry.message) as Parameters<typeof sessionManager.appendMessage>[0],
);
}
if (entry.type === "compaction") {
return sessionManager.appendCompaction(
@@ -178,6 +187,7 @@ export function rewriteTranscriptEntriesInSessionManager(params: {
// re-running persistence hooks or size truncation on replayed messages.
const appendMessage = getRawSessionAppendMessage(params.sessionManager);
const rewrittenEntryIds = new Map<string, string>();
// Every re-appended message follows the rewritten prefix, so its prefix-bound checkpoint is stale.
for (const entry of branch.slice(firstMatchedIndex)) {
const replacement = entry.type === "message" ? replacementsById.get(entry.id) : undefined;
const newEntryId =
@@ -188,7 +198,11 @@ export function rewriteTranscriptEntriesInSessionManager(params: {
rewrittenEntryIds,
appendMessage,
})
: appendMessage(replacement as Parameters<typeof params.sessionManager.appendMessage>[0]);
: appendMessage(
stripStalePrefixReplay(replacement) as Parameters<
typeof params.sessionManager.appendMessage
>[0],
);
rewrittenEntryIds.set(entry.id, newEntryId);
}