diff --git a/src/agents/cache-trace.ts b/src/agents/cache-trace.ts index a66f929726c2..54d7359e5124 100644 --- a/src/agents/cache-trace.ts +++ b/src/agents/cache-trace.ts @@ -119,7 +119,7 @@ function digest(value: unknown): string { return crypto.createHash("sha256").update(serialized).digest("hex"); } -function summarizeMessages(messages: AgentMessage[]): { +export function summarizeMessages(messages: AgentMessage[]): { messageCount: number; messageRoles: Array; messageFingerprints: string[]; diff --git a/src/agents/embedded-agent-runner/run/attempt.llm-boundary.cache-stability.test.ts b/src/agents/embedded-agent-runner/run/attempt.llm-boundary.cache-stability.test.ts index a9443b63fd5c..b6e83fdc150f 100644 --- a/src/agents/embedded-agent-runner/run/attempt.llm-boundary.cache-stability.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt.llm-boundary.cache-stability.test.ts @@ -1,3 +1,6 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; import { streamOpenAICompletions, streamOpenAIResponses } from "@openclaw/ai/internal/openai"; /** * Cache-stability gate for the prompt-cache bust fix (issue #3658). @@ -22,6 +25,13 @@ import { describe, expect, it } from "vitest"; import { stripInboundMetadata } from "../../../auto-reply/reply/strip-inbound-meta.js"; import { buildTimestampPrefix } from "../../../gateway/server-methods/agent-timestamp.js"; import type { Context, Model } from "../../../llm/types.js"; +import { + appendUserTurnTranscriptMessage, + createUserTurnTranscriptRecorder, + mergePreparedUserTurnMessageForRuntime, + type UserTurnInput, +} from "../../../sessions/user-turn-transcript.js"; +import { summarizeMessages } from "../../cache-trace.js"; import { OPENCLAW_RUNTIME_CONTEXT_CUSTOM_TYPE, relocateCurrentRuntimeContextCarrierToTail, @@ -358,6 +368,95 @@ describe("prompt-cache byte-identity (issue #3658)", () => { }); }); +describe("append-only late media (issue #99495)", () => { + it("keeps every sent fingerprint stable and appends one late-media turn", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-99495-boundary-")); + const transcriptPath = path.join(dir, "session.jsonl"); + const admittedInput = { + text: "describe this", + timestamp: TS_TURN1, + idempotencyKey: "cache-99495:user", + }; + let resolveMedia!: (input: UserTurnInput) => void; + let markResolverStarted!: () => void; + const resolverStarted = new Promise((resolve) => { + markResolverStarted = resolve; + }); + const mediaInput = new Promise((resolve) => { + resolveMedia = resolve; + }); + try { + const recorder = createUserTurnTranscriptRecorder({ + input: admittedInput, + resolveInput: async () => { + markResolverStarted(); + return await mediaInput; + }, + target: { transcriptPath, sessionId: "session-99495", sessionKey: "main", cwd: dir }, + }); + const persistence = recorder.persistFallback(); + await resolverStarted; + await appendUserTurnTranscriptMessage({ + transcriptPath, + input: admittedInput, + sessionId: "session-99495", + sessionKey: "main", + cwd: dir, + }); + recorder.markRuntimePersisted(recorder.message); + const admittedRuntimeMessage = mergePreparedUserTurnMessageForRuntime({ + runtimeMessage: currentUserMsg(admittedInput.text, admittedInput.timestamp), + preparedMessage: recorder.message, + }); + const sent = normalizeMessagesForLlmBoundary([admittedRuntimeMessage], { timezone: TZ }); + recorder.markSentToProvider?.(); + resolveMedia({ + ...admittedInput, + media: [{ path: path.join(dir, "image.png"), contentType: "image/png" }], + }); + await persistence; + const persisted = fs + .readFileSync(transcriptPath, "utf8") + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { message?: AgentMsg }) + .flatMap((entry) => (entry.message ? [entry.message] : [])); + const next = normalizeMessagesForLlmBoundary(persisted, { timezone: TZ }); + const sentSummary = summarizeMessages(sent); + const nextSummary = summarizeMessages(next); + + expect(nextSummary.messageCount).toBe(sentSummary.messageCount + 1); + expect(nextSummary.messageFingerprints.slice(0, sentSummary.messageCount)).toEqual( + sentSummary.messageFingerprints, + ); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + + it("keeps media inline when resolution finishes before serialization", async () => { + const prepared = createUserTurnTranscriptRecorder({ + input: { text: "describe this", timestamp: TS_TURN1 }, + resolveInput: async () => ({ + text: "describe this", + timestamp: TS_TURN1, + media: [{ path: "media://inbound/image.jpg", contentType: "image/jpeg" }], + }), + target: { transcriptPath: "/unused/session.jsonl" }, + }); + const resolved = await prepared.resolveMessage(); + const merged = mergePreparedUserTurnMessageForRuntime({ + runtimeMessage: currentUserMsg("describe this", TS_TURN1), + preparedMessage: resolved, + }); + prepared.markSentToProvider?.(); + const summary = summarizeMessages(normalizeMessagesForLlmBoundary([merged], { timezone: TZ })); + + expect(summary.messageCount).toBe(1); + expect(merged).toMatchObject({ MediaPath: "media://inbound/image.jpg" }); + }); +}); + function runtimeCarrier(content: string, timestamp: number): AgentMsg { return { role: "custom", diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 3e34c109a1d8..22482fc51d11 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -320,6 +320,12 @@ import { } from "../sandbox-skills.js"; import { prewarmSessionFile, trackSessionManagerAccess } from "../session-manager-cache.js"; import { prepareSessionManagerForRun } from "../session-manager-init.js"; +import { + cloneToolResultPromptProjectionState, + getEmbeddedSessionPromptState, + hasSessionUserTurnBeenSent, + markSessionUserTurnsSent, +} from "../session-prompt-state.js"; import { describeEmbeddedAgentStreamStrategy, resetEmbeddedAgentBaseStreamFnCacheForTest, @@ -347,9 +353,7 @@ import { import { resolveLiveToolResultMaxChars, resolveLiveToolResultAggregateMaxChars, - createToolResultPromptProjectionState, truncateOversizedToolResultsInMessages, - type ToolResultPromptProjectionState, truncateOversizedToolResultsInSessionManager, } from "../tool-result-truncation.js"; import { splitSdkTools } from "../tool-split.js"; @@ -2656,8 +2660,10 @@ export async function runEmbeddedAttempt( ); } let prePromptMessageCount = activeSession.messages.length; - const toolResultPromptProjectionState: ToolResultPromptProjectionState = - createToolResultPromptProjectionState(); + // Session-owned projections survive attempt teardown so already-sent tool results + // cannot rewrite the provider prompt-cache tail between turns (#99495). + const sessionPromptState = getEmbeddedSessionPromptState(params.sessionId); + const toolResultPromptProjectionState = sessionPromptState.toolResults; let contextEngineAfterTurnCheckpoint: number | null = null; let unwindowedContextEngineMessagesForPrecheck: AgentMessage[] | undefined; let contextEnginePromptAuthority: NonNullable = @@ -4373,7 +4379,7 @@ export async function runEmbeddedAttempt( contextTokenBudget, promptToolResultMaxChars, promptToolResultAggregateMaxChars, - toolResultPromptProjectionState, + cloneToolResultPromptProjectionState(toolResultPromptProjectionState), ); const promptHistoryChanged = promptToolResultTruncation.messages !== activeSession.messages; @@ -4999,9 +5005,21 @@ export async function runEmbeddedAttempt( promptToolResultAggregateMaxChars, toolResultPromptProjectionState, ); - return providerPromptHistoryTruncation.messages !== messages - ? providerPromptHistoryTruncation.messages - : messages; + const providerMessages = + providerPromptHistoryTruncation.messages !== messages + ? providerPromptHistoryTruncation.messages + : messages; + // This provider-dispatch transform marks the current turn sent so late + // media appends instead of rewriting its prompt-cache slot (#99495). + markSessionUserTurnsSent(sessionPromptState, providerMessages); + const recorder = params.userTurnTranscriptRecorder; + if ( + recorder && + hasSessionUserTurnBeenSent(sessionPromptState, recorder.message) !== false + ) { + recorder.markSentToProvider?.(); + } + return providerMessages; }, ); activeSession.agent.streamFn = providerPromptStreamFn; diff --git a/src/agents/embedded-agent-runner/session-prompt-state.ts b/src/agents/embedded-agent-runner/session-prompt-state.ts new file mode 100644 index 000000000000..cd17b63ba43d --- /dev/null +++ b/src/agents/embedded-agent-runner/session-prompt-state.ts @@ -0,0 +1,107 @@ +/** Process-local prompt projection state owned by an embedded session lifecycle. */ +import { resolveGlobalSingleton } from "../../shared/global-singleton.js"; +import type { AgentMessage } from "../runtime/index.js"; + +export type ToolResultPromptProjectionState = { + replacements: Map; + frozen: Set; + ambiguousBaseKeys: Set; + sourceTextByKey: Map; +}; + +export type EmbeddedSessionPromptState = { + toolResults: ToolResultPromptProjectionState; + sentUserTurnIds: Set; +}; + +const MAX_SESSION_PROMPT_STATES = 64; +const SESSION_PROMPT_STATES_KEY = Symbol.for("openclaw.embeddedSessionPromptStates"); +const sessionPromptStates = resolveGlobalSingleton( + SESSION_PROMPT_STATES_KEY, + () => new Map(), +); + +function createSessionPromptState(): EmbeddedSessionPromptState { + return { + toolResults: { + replacements: new Map(), + frozen: new Set(), + ambiguousBaseKeys: new Set(), + sourceTextByKey: new Map(), + }, + sentUserTurnIds: new Set(), + }; +} + +export function cloneToolResultPromptProjectionState( + state: ToolResultPromptProjectionState, +): ToolResultPromptProjectionState { + return { + replacements: new Map(state.replacements), + frozen: new Set(state.frozen), + ambiguousBaseKeys: new Set(state.ambiguousBaseKeys), + sourceTextByKey: new Map(state.sourceTextByKey), + }; +} + +export function getEmbeddedSessionPromptState(sessionId: string): EmbeddedSessionPromptState { + const existing = sessionPromptStates.get(sessionId); + if (existing) { + sessionPromptStates.delete(sessionId); + sessionPromptStates.set(sessionId, existing); + return existing; + } + const created = createSessionPromptState(); + sessionPromptStates.set(sessionId, created); + while (sessionPromptStates.size > MAX_SESSION_PROMPT_STATES) { + const oldest = sessionPromptStates.keys().next().value; + if (typeof oldest !== "string") { + break; + } + sessionPromptStates.delete(oldest); + } + return created; +} + +export function clearEmbeddedSessionPromptStates(sessionIds: Iterable): void { + for (const sessionId of sessionIds) { + const normalized = sessionId?.trim(); + if (normalized) { + sessionPromptStates.delete(normalized); + } + } +} + +export function markSessionUserTurnsSent( + state: EmbeddedSessionPromptState, + messages: AgentMessage[], +): void { + for (const message of messages) { + if (message.role !== "user") { + continue; + } + const idempotencyKey = (message as { idempotencyKey?: unknown }).idempotencyKey; + if (typeof idempotencyKey === "string" && idempotencyKey.length > 0) { + state.sentUserTurnIds.add(idempotencyKey); + } + } +} + +export function hasSessionUserTurnBeenSent( + state: EmbeddedSessionPromptState, + message: AgentMessage | undefined, +): boolean | undefined { + if (!message || message.role !== "user") { + return undefined; + } + const idempotencyKey = (message as { idempotencyKey?: unknown }).idempotencyKey; + return typeof idempotencyKey === "string" && idempotencyKey.length > 0 + ? state.sentUserTurnIds.has(idempotencyKey) + : undefined; +} + +export const testing = { + reset() { + sessionPromptStates.clear(); + }, +}; 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 c4e11cbb16c3..38c9346495f4 100644 --- a/src/agents/embedded-agent-runner/tool-result-truncation.test.ts +++ b/src/agents/embedded-agent-runner/tool-result-truncation.test.ts @@ -10,6 +10,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { onSessionTranscriptUpdate } from "../../sessions/transcript-events.js"; import { formatFullOutputFooter } from "../sessions/tools/tool-contracts.js"; import { makeAgentAssistantMessage } from "../test-helpers/agent-message-fixtures.js"; +import { + getEmbeddedSessionPromptState, + testing as sessionPromptStateTesting, +} from "./session-prompt-state.js"; let truncateToolResultText: typeof import("./tool-result-truncation.js").truncateToolResultText; let truncateToolResultMessage: typeof import("./tool-result-truncation.js").truncateToolResultMessage; @@ -57,6 +61,7 @@ beforeEach(async () => { }); afterEach(async () => { + sessionPromptStateTesting.reset(); if (tmpDir) { await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}); tmpDir = undefined; @@ -644,6 +649,43 @@ describe("truncateOversizedToolResultsInMessages", () => { expect(lastText && getToolResultTextLength(lastText)).toBeLessThanOrEqual(12_000); }); + it("keeps #99495 historical bytes stable across attempts sharing session state", () => { + const state = getEmbeddedSessionPromptState("session-99495").toolResults; + const history = [ + makeToolResult("a".repeat(4_000), "history_1"), + makeToolResult("b".repeat(4_000), "history_2"), + ]; + const first = truncateOversizedToolResultsInMessages(history, 128_000, 5_000, 20_000, state); + const secondAttemptState = getEmbeddedSessionPromptState("session-99495").toolResults; + const second = truncateOversizedToolResultsInMessages( + [...history, makeToolResult("c".repeat(12_000), "current")], + 128_000, + 5_000, + 20_000, + secondAttemptState, + ); + + expect(secondAttemptState).toBe(state); + expect(second.messages.slice(0, history.length)).toEqual(first.messages); + }); + + it("shrinks #99495 frozen bytes monotonically only under a tighter hard cap", () => { + const state = getEmbeddedSessionPromptState("session-99495-shrink").toolResults; + const history = [ + makeToolResult("a".repeat(8_000), "history_1"), + makeToolResult("b".repeat(8_000), "history_2"), + ]; + const first = truncateOversizedToolResultsInMessages(history, 128_000, 6_000, 20_000, state); + const shrunk = truncateOversizedToolResultsInMessages(history, 128_000, 3_000, 20_000, state); + const relaxed = truncateOversizedToolResultsInMessages(history, 128_000, 7_000, 20_000, state); + const lengths = (messages: AgentMessage[]) => messages.map(getToolResultTextLength); + + expect( + lengths(shrunk.messages).every((length, index) => length <= lengths(first.messages)[index]!), + ).toBe(true); + expect(relaxed.messages).toEqual(shrunk.messages); + }); + it("preserves fresh trailing tool results when aggregate history is already saturated", () => { const projectionState = createToolResultPromptProjectionState(); const history: AgentMessage[] = []; @@ -978,7 +1020,7 @@ describe("truncateOversizedToolResultsInMessages", () => { ).toBeLessThan(15_000); }); - it("does not reuse ambiguous projections across filtered history", () => { + it("freezes #99495 ambiguous-key projections across filtered history", () => { const projectionState = createToolResultPromptProjectionState(); const duplicate = (text: string) => ({ role: "toolResult" as const, @@ -1004,7 +1046,7 @@ describe("truncateOversizedToolResultsInMessages", () => { ); expect(first.messages[0]).not.toEqual(first.messages[1]); - expect(filtered.messages[0]).toEqual(duplicate("b".repeat(100))); + expect(filtered.messages[0]).toEqual(first.messages[1]); }); }); diff --git a/src/agents/embedded-agent-runner/tool-result-truncation.ts b/src/agents/embedded-agent-runner/tool-result-truncation.ts index 4c6d2e2ecc51..f9892123c46c 100644 --- a/src/agents/embedded-agent-runner/tool-result-truncation.ts +++ b/src/agents/embedded-agent-runner/tool-result-truncation.ts @@ -1,6 +1,7 @@ /** * Truncates oversized tool-result content in messages and transcripts. */ +import { createHash } from "node:crypto"; import { existsSync } from "node:fs"; import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce"; import { sliceUtf16Safe, truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; @@ -19,6 +20,7 @@ import { SessionManager } from "../sessions/index.js"; import { formatFullOutputFooter } from "../sessions/tools/tool-contracts.js"; import { formatContextLimitTruncationNotice } from "./context-truncation-notice.js"; import { log } from "./logger.js"; +import type { ToolResultPromptProjectionState } from "./session-prompt-state.js"; import { persistTranscriptStateMutation, readTranscriptFileState, @@ -666,12 +668,7 @@ type ToolResultReplacement = { message: AgentMessage; }; -export type ToolResultPromptProjectionState = { - replacements: Map; - frozen: Set; - ambiguousBaseKeys: Set; - sourceTextByKey: Map; -}; +export type { ToolResultPromptProjectionState } from "./session-prompt-state.js"; export function createToolResultPromptProjectionState(): ToolResultPromptProjectionState { return { @@ -712,16 +709,27 @@ function getToolResultProjectionKeys( } } const occurrences = new Map(); - return baseKeys.map((baseKey) => { - if (!baseKey) { + return baseKeys.map((baseKey, index) => { + if (baseKey && !projectionState.ambiguousBaseKeys.has(baseKey)) { + return baseKey; + } + const message = messages[index]; + if (!message || message.role !== "toolResult") { return undefined; } - if (projectionState.ambiguousBaseKeys.has(baseKey)) { - return undefined; - } - const occurrence = occurrences.get(baseKey) ?? 0; - occurrences.set(baseKey, occurrence + 1); - return `${baseKey}:${occurrence}`; + // Ambiguous/missing tool ids still need a stable frozen identity; otherwise + // each request rewrites their prompt-cache tail projection (#99495). + const messageId = (message as { id?: unknown }).id; + const sourceIdentity = + typeof messageId === "string" && messageId.length > 0 + ? `id:${messageId}` + : `text:${createHash("sha256") + .update(JSON.stringify(getToolResultTextBlocks(message))) + .digest("base64url")}`; + const fallbackBase = `fallback:${baseKey ?? "tool"}:${sourceIdentity}`; + const occurrence = occurrences.get(fallbackBase) ?? 0; + occurrences.set(fallbackBase, occurrence + 1); + return `${fallbackBase}:${occurrence}`; }); } @@ -846,9 +854,9 @@ function buildAggregateToolResultReplacements(params: { }); const recoveryCandidates = [ ...aggregateRecoveryCandidates.filter((item) => item.aggregateEligible), - ...(protectedEntryIds.size > 0 - ? aggregateRecoveryCandidates.filter((item) => !item.aggregateEligible) - : []), + // Frozen bytes move only after fresh output is exhausted and the hard aggregate + // guard still overflows; starting from the frozen projection makes this shrink-only. + ...aggregateRecoveryCandidates.filter((item) => !item.aggregateEligible), ]; // Spend aggregate reduction on older entries first so fresh tool output stays intact. diff --git a/src/auto-reply/reply/session-reset-cleanup.test.ts b/src/auto-reply/reply/session-reset-cleanup.test.ts index 8ff3c06a2342..1846029406e6 100644 --- a/src/auto-reply/reply/session-reset-cleanup.test.ts +++ b/src/auto-reply/reply/session-reset-cleanup.test.ts @@ -1,5 +1,9 @@ // Tests session reset cleanup for stale files and persisted state. import { afterEach, describe, expect, it, vi } from "vitest"; +import { + getEmbeddedSessionPromptState, + testing as sessionPromptStateTesting, +} from "../../agents/embedded-agent-runner/session-prompt-state.js"; import { enqueueSystemEvent, peekSystemEvents, @@ -14,12 +18,22 @@ import { import { clearSessionResetRuntimeState } from "./session-reset-cleanup.js"; afterEach(() => { + sessionPromptStateTesting.reset(); replyRunTesting.resetReplyRunRegistry(); resetDiagnosticRunActivityForTest(); resetSystemEventsForTest(); }); describe("clearSessionResetRuntimeState", () => { + it("disposes prompt projections with the archived session", () => { + const state = getEmbeddedSessionPromptState("old-session"); + state.sentUserTurnIds.add("sent-user-turn"); + + clearSessionResetRuntimeState(["old-session"]); + + expect(getEmbeddedSessionPromptState("old-session")).not.toBe(state); + }); + it("clears reset queues and drains system events for normalized keys", () => { enqueueSystemEvent("stale alpha", { sessionKey: "alpha" }); enqueueSystemEvent("stale beta", { sessionKey: "beta" }); diff --git a/src/auto-reply/reply/session-reset-cleanup.ts b/src/auto-reply/reply/session-reset-cleanup.ts index c21c72e27f3b..c4970265367f 100644 --- a/src/auto-reply/reply/session-reset-cleanup.ts +++ b/src/auto-reply/reply/session-reset-cleanup.ts @@ -1,4 +1,5 @@ /** Clears reset-related queues and system events for session keys. */ +import { clearEmbeddedSessionPromptStates } from "../../agents/embedded-agent-runner/session-prompt-state.js"; import { drainSystemEventEntries } from "../../infra/system-events.js"; import { clearSessionQueues, type ClearSessionQueueResult } from "./queue/cleanup.js"; import { clearReplyRunForResetBySessionId } from "./reply-run-registry.js"; @@ -13,6 +14,7 @@ export function clearSessionResetRuntimeState( keys: Array, opts?: { activeReplySessionId?: string }, ): ClearSessionResetRuntimeStateResult { + clearEmbeddedSessionPromptStates(keys); const cleared = clearSessionQueues(keys); let systemEventsCleared = 0; diff --git a/src/sessions/user-turn-transcript.test.ts b/src/sessions/user-turn-transcript.test.ts index 009d0abd88ba..4d39fbc412cf 100644 --- a/src/sessions/user-turn-transcript.test.ts +++ b/src/sessions/user-turn-transcript.test.ts @@ -17,6 +17,7 @@ import { mergePreparedUserTurnMessageForRuntime, persistUserTurnTranscript, resolvePersistedUserTurnText, + type UserTurnInput, } from "./user-turn-transcript.js"; describe("user turn transcript persistence", () => { @@ -717,6 +718,101 @@ describe("user turn transcript persistence", () => { ]); }); + it("appends #99495 media that resolves after the admitted turn reached the provider", async () => { + const dir = createTempDir("openclaw-user-turn-recorder-late-media-"); + const transcriptPath = path.join(dir, "session.jsonl"); + const admittedInput = { + text: "describe this", + timestamp: 123, + idempotencyKey: "chat-run-late:user", + }; + let resolveMedia!: (input: UserTurnInput) => void; + let markResolverStarted!: () => void; + const resolverStarted = new Promise((resolve) => { + markResolverStarted = resolve; + }); + const mediaInput = new Promise((resolve) => { + resolveMedia = resolve; + }); + const recorder = createUserTurnTranscriptRecorder({ + input: admittedInput, + resolveInput: async () => { + markResolverStarted(); + return await mediaInput; + }, + target: { + transcriptPath, + sessionId: "session-1", + sessionKey: "main", + cwd: dir, + }, + }); + const persistence = recorder.persistFallback(); + await resolverStarted; + await appendUserTurnTranscriptMessage({ + transcriptPath, + input: admittedInput, + sessionId: "session-1", + sessionKey: "main", + cwd: dir, + }); + recorder.markRuntimePersisted(recorder.message); + recorder.markSentToProvider?.(); + resolveMedia({ + ...admittedInput, + media: [{ path: path.join(dir, "image.png"), contentType: "image/png" }], + }); + + await persistence; + + expect(readTranscriptMessages(transcriptPath)).toEqual([ + expect.objectContaining({ + content: "describe this", + idempotencyKey: "chat-run-late:user", + }), + expect.objectContaining({ + content: `[media attached: ${path.join(dir, "image.png")}]`, + idempotencyKey: "chat-run-late:user:late-media", + MediaPath: path.join(dir, "image.png"), + }), + ]); + }); + + it("keeps #99495 media inline when it resolves before first serialization", async () => { + const dir = createTempDir("openclaw-user-turn-recorder-early-media-"); + const transcriptPath = path.join(dir, "session.jsonl"); + const recorder = createUserTurnTranscriptRecorder({ + input: { + text: "describe this", + timestamp: 123, + idempotencyKey: "chat-run-early:user", + }, + resolveInput: async () => ({ + text: "describe this", + timestamp: 123, + idempotencyKey: "chat-run-early:user", + media: [{ path: path.join(dir, "image.png"), contentType: "image/png" }], + }), + target: { + transcriptPath, + sessionId: "session-1", + sessionKey: "main", + cwd: dir, + }, + }); + + await recorder.persistFallback(); + recorder.markSentToProvider?.(); + + expect(readTranscriptMessages(transcriptPath)).toEqual([ + expect.objectContaining({ + content: "describe this", + idempotencyKey: "chat-run-early:user", + MediaPath: path.join(dir, "image.png"), + }), + ]); + }); + it("falls back to the admitted text message when lazy media resolution fails", async () => { const dir = createTempDir("openclaw-user-turn-recorder-lazy-failed-"); const transcriptPath = path.join(dir, "session.jsonl"); diff --git a/src/sessions/user-turn-transcript.ts b/src/sessions/user-turn-transcript.ts index 9d156a75cc24..34755cc6a319 100644 --- a/src/sessions/user-turn-transcript.ts +++ b/src/sessions/user-turn-transcript.ts @@ -301,6 +301,54 @@ function isUserMessage(message: AgentMessage): message is PersistedUserTurnMessa return (message as { role?: unknown }).role === "user"; } +function buildLateResolvedMediaMessage(params: { + admittedMessage?: PersistedUserTurnMessage; + resolvedMessage: PersistedUserTurnMessage; +}): PersistedUserTurnMessage | undefined { + const admittedMedia = buildPersistedUserTurnMediaInputsFromFields(params.admittedMessage); + const resolvedMedia = buildPersistedUserTurnMediaInputsFromFields(params.resolvedMessage); + if ( + resolvedMedia.length === 0 || + JSON.stringify(resolvedMedia) === JSON.stringify(admittedMedia) + ) { + return undefined; + } + const resolved = params.resolvedMessage as unknown as Record; + const admittedContent = params.admittedMessage?.content; + const resolvedContent = params.resolvedMessage.content; + const mediaOnlyText = resolvedMedia + .map((media) => media.path ?? media.url) + .filter((value): value is string => Boolean(value)) + .map((value) => `[media attached: ${value}]`) + .join("\n"); + const content = + typeof resolvedContent === "string" && resolvedContent === admittedContent + ? mediaOnlyText + : Array.isArray(resolvedContent) && typeof admittedContent === "string" + ? (() => { + const mediaContent = resolvedContent.filter( + (block) => + !block || + typeof block !== "object" || + (block as { type?: unknown; text?: unknown }).type !== "text" || + (block as { text?: unknown }).text !== admittedContent, + ); + return mediaContent.length > 0 + ? mediaContent + : [{ type: "text" as const, text: mediaOnlyText }]; + })() + : resolvedContent; + const idempotencyKey = + typeof resolved.idempotencyKey === "string" && resolved.idempotencyKey.length > 0 + ? `${resolved.idempotencyKey}:late-media` + : `late-media:${typeof resolved.timestamp === "number" ? resolved.timestamp : Date.now()}`; + return { + ...resolved, + content, + idempotencyKey, + } as unknown as PersistedUserTurnMessage; +} + function isBeforeAgentRunBlockedMessage(message: AgentMessage): boolean { const marker = (message as { __openclaw?: { beforeAgentRunBlocked?: unknown } })["__openclaw"] ?.beforeAgentRunBlocked; @@ -567,6 +615,9 @@ export function createUserTurnTranscriptRecorder( let selfPersistencePromise: Promise | undefined; let resolvedMessagePromise: Promise | undefined; let persistedMessageNotified = false; + let runtimePersistedMessage: PersistedUserTurnMessage | undefined; + let sentToProvider = false; + let resolvedBeforeProvider = false; const handlePersistenceError = (error: unknown) => { if (params.onPersistenceError) { @@ -593,12 +644,13 @@ export function createUserTurnTranscriptRecorder( resolvedMessagePromise = (async () => { try { const resolvedInput = await params.resolveInput?.(); - return ( + const resolvedMessage = resolvePersistedUserTurnMessage({ message: params.message, input: resolvedInput ?? params.input, - }) ?? message - ); + }) ?? message; + resolvedBeforeProvider = !sentToProvider; + return resolvedMessage; } catch (error) { handlePersistenceError(error); return message; @@ -640,9 +692,6 @@ export function createUserTurnTranscriptRecorder( target?: UserTurnTranscriptTargetResolver; updateMode?: UserTurnTranscriptUpdateMode; }): Promise => { - if (persisted) { - return persistedResult; - } if (options.skipWhenBlocked && blocked) { return undefined; } @@ -653,9 +702,6 @@ export function createUserTurnTranscriptRecorder( // Approved persistence waits for runtime-owned writes first to avoid // duplicate user turns when the harness already persisted the message. await waitForRuntimePersistence(); - if (persisted) { - return persistedResult; - } } if (selfPersistencePromise) { return await selfPersistencePromise; @@ -670,19 +716,54 @@ export function createUserTurnTranscriptRecorder( return undefined; } const updateMode = options.updateMode ?? params.updateMode ?? "inline"; - const result = isUserTurnTranscriptFileTarget(target) - ? await appendFileTargetUserTurnTranscript({ - target, - message: resolvedMessage, - updateMode, - beforeMessageWrite: params.beforeMessageWrite, - }) - : await persistUserTurnTranscript({ - ...target, - message: resolvedMessage, - updateMode, - ...(params.beforeMessageWrite ? { beforeMessageWrite: params.beforeMessageWrite } : {}), - }); + const persistMessage = async ( + candidate: PersistedUserTurnMessage, + candidateUpdateMode: UserTurnTranscriptUpdateMode, + ) => + isUserTurnTranscriptFileTarget(target) + ? await appendFileTargetUserTurnTranscript({ + target, + message: candidate, + updateMode: candidateUpdateMode, + beforeMessageWrite: params.beforeMessageWrite, + }) + : await persistUserTurnTranscript({ + ...target, + message: candidate, + updateMode: candidateUpdateMode, + ...(params.beforeMessageWrite + ? { beforeMessageWrite: params.beforeMessageWrite } + : {}), + }); + const lateMediaMessage = + sentToProvider && !resolvedBeforeProvider + ? buildLateResolvedMediaMessage({ + admittedMessage: runtimePersistedMessage ?? message, + resolvedMessage, + }) + : undefined; + if (lateMediaMessage) { + // The admitted bytes already crossed the LLM boundary. Persisting media as a + // second turn preserves that prefix; inline replacement would thrash cache tail (#99495). + if (!persisted && message) { + const admittedResult = await persistMessage(message, updateMode); + if (admittedResult) { + persisted = true; + persistedResult = admittedResult; + notifyMessagePersisted(admittedResult.message); + } + } + const appendedMedia = await persistMessage(lateMediaMessage, "none"); + if (appendedMedia) { + persisted = true; + persistedResult = appendedMedia; + } + return appendedMedia; + } + if (persisted) { + return persistedResult; + } + const result = await persistMessage(resolvedMessage, updateMode); if (result) { persisted = true; persistedResult = result; @@ -701,11 +782,15 @@ export function createUserTurnTranscriptRecorder( return { message, resolveMessage: resolveMessageForPersistence, + markSentToProvider: () => { + sentToProvider = true; + }, markRuntimePersistencePending: (pending) => { runtimePersistencePromise = pending; }, markRuntimePersisted: (persistedMessage) => { persisted = true; + runtimePersistedMessage = persistedMessage; if (persistedMessage && persistedResult) { persistedResult = { ...persistedResult, diff --git a/src/sessions/user-turn-transcript.types.ts b/src/sessions/user-turn-transcript.types.ts index bf956e896797..98ec205cc0d3 100644 --- a/src/sessions/user-turn-transcript.types.ts +++ b/src/sessions/user-turn-transcript.types.ts @@ -78,6 +78,7 @@ export type UserTurnTranscriptTargetResolver = export type UserTurnTranscriptRecorder = { readonly message: PersistedUserTurnMessage | undefined; resolveMessage: () => Promise; + markSentToProvider?: () => void; markRuntimePersistencePending: (pending: Promise) => void; markRuntimePersisted: (message?: PersistedUserTurnMessage) => void; markBlocked: () => void;