diff --git a/packages/agent-core/src/harness/session/session-context.test.ts b/packages/agent-core/src/harness/session/session-context.test.ts index 62af8f63c0cf..2891df8f7d0a 100644 --- a/packages/agent-core/src/harness/session/session-context.test.ts +++ b/packages/agent-core/src/harness/session/session-context.test.ts @@ -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", () => { diff --git a/packages/agent-core/src/harness/session/session.ts b/packages/agent-core/src/harness/session/session.ts index 3bce804a524e..e07ba76c1bae 100644 --- a/packages/agent-core/src/harness/session/session.ts +++ b/packages/agent-core/src/harness/session/session.ts @@ -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 }); } } } diff --git a/packages/ai/src/transports/openai-responses-compaction-replay.test.ts b/packages/ai/src/transports/openai-responses-compaction-replay.test.ts index 97dd51732b79..a8d3ac8190eb 100644 --- a/packages/ai/src/transports/openai-responses-compaction-replay.test.ts +++ b/packages/ai/src/transports/openai-responses-compaction-replay.test.ts @@ -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(); diff --git a/packages/ai/src/transports/openai-responses-compaction-replay.ts b/packages/ai/src/transports/openai-responses-compaction-replay.ts index 7fc205b93eaf..9e6cf36bd0a9 100644 --- a/packages/ai/src/transports/openai-responses-compaction-replay.ts +++ b/packages/ai/src/transports/openai-responses-compaction-replay.ts @@ -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; diff --git a/packages/ai/src/transports/openai-responses-replay.ts b/packages/ai/src/transports/openai-responses-replay.ts index 06aee280ad30..e4ffb6f01715 100644 --- a/packages/ai/src/transports/openai-responses-replay.ts +++ b/packages/ai/src/transports/openai-responses-replay.ts @@ -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; diff --git a/src/agents/embedded-agent-runner/tool-result-truncation.test.ts b/src/agents/embedded-agent-runner/tool-result-truncation.test.ts index 05ee1275e59d..3802c3d21f5c 100644 --- a/src/agents/embedded-agent-runner/tool-result-truncation.test.ts +++ b/src/agents/embedded-agent-runner/tool-result-truncation.test.ts @@ -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; + 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; + 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 () => { diff --git a/src/agents/embedded-agent-runner/transcript-rewrite.ts b/src/agents/embedded-agent-runner/transcript-rewrite.ts index fb549e445471..1e9cf1c12b2e 100644 --- a/src/agents/embedded-agent-runner/transcript-rewrite.ts +++ b/src/agents/embedded-agent-runner/transcript-rewrite.ts @@ -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; type SessionBranchEntry = ReturnType[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[0]); + return appendMessage( + stripStalePrefixReplay(entry.message) as Parameters[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(); + // 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[0]); + : appendMessage( + stripStalePrefixReplay(replacement) as Parameters< + typeof params.sessionManager.appendMessage + >[0], + ); rewrittenEntryIds.set(entry.id, newEntryId); }