diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 113829a8b6f0..62567ec6e6e3 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -1649,6 +1649,8 @@ export function buildOversizedHistoryPlaceholder(message?: unknown): Record { session: { hasActiveRun: false }, }); }); + + it("broadcasts user idempotency keys in session.message metadata", async () => { + await expect( + emitAssistantTranscriptUpdate(false, { + role: "user", + content: [{ type: "text", text: "Optimistic turn" }], + idempotencyKey: "client-turn-3", + }), + ).resolves.toMatchObject({ + message: { + __openclaw: { + id: "message-1", + idempotencyKey: "client-turn-3", + seq: 1, + }, + }, + }); + }); }); diff --git a/src/gateway/server-session-events.ts b/src/gateway/server-session-events.ts index b18379299e84..e9e89096e3f9 100644 --- a/src/gateway/server-session-events.ts +++ b/src/gateway/server-session-events.ts @@ -30,6 +30,14 @@ import { type SessionEventSubscribers = Pick; type SessionMessageSubscribers = Pick; +function readMessageIdempotencyKey(message: unknown): string | undefined { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return undefined; + } + const value = (message as Record).idempotencyKey; + return typeof value === "string" && value.trim() ? value : undefined; +} + function resolveSessionMessageBroadcastKeys(sessionKey: string, agentId?: string): string[] { // Global sessions can be subscribed through either the raw global key or the // default-agent scoped key; non-default agent global sessions stay scoped. @@ -173,8 +181,10 @@ async function handleTranscriptUpdateBroadcast( includeSession: true, hasActiveRun, }); + const idempotencyKey = readMessageIdempotencyKey(update.message); const rawMessage = attachOpenClawTranscriptMeta(update.message, { ...(typeof update.messageId === "string" ? { id: update.messageId } : {}), + ...(idempotencyKey ? { idempotencyKey } : {}), ...(messageSeq !== undefined ? { seq: messageSeq } : {}), }); const message = projectChatDisplayMessage(rawMessage); diff --git a/src/gateway/session-history-state.test.ts b/src/gateway/session-history-state.test.ts index 005b8a891f3d..3e0d0b7c6676 100644 --- a/src/gateway/session-history-state.test.ts +++ b/src/gateway/session-history-state.test.ts @@ -126,6 +126,34 @@ describe("SessionHistorySseState", () => { } }); + test("carries inline user idempotency keys into history metadata", () => { + const state = newState([]); + + const appended = state.appendInlineMessage({ + message: { + role: "user", + content: [{ type: "text", text: "optimistic turn" }], + idempotencyKey: "client-turn-2", + }, + messageId: "message-user-2", + messageSeq: 2, + }); + + expect(appended).toBeDefined(); + expect(appended?.messageSeq).toBe(2); + expect( + ( + appended!.message as { + __openclaw?: { id?: string; idempotencyKey?: string; seq?: number }; + } + )["__openclaw"], + ).toMatchObject({ + id: "message-user-2", + idempotencyKey: "client-turn-2", + seq: 2, + }); + }); + test("reuses one canonical array for items and messages", () => { const snapshot = buildSessionHistorySnapshot({ rawMessages: [assistantTextMessage("first", 1), assistantTextMessage("second", 2)], diff --git a/src/gateway/session-history-state.ts b/src/gateway/session-history-state.ts index 3f7ce9ce082c..92156fb1e656 100644 --- a/src/gateway/session-history-state.ts +++ b/src/gateway/session-history-state.ts @@ -16,6 +16,7 @@ import { // raw messages are projected for display, paginated by transcript seq, then // incrementally updated until cursor/window semantics require a full refresh. type SessionHistoryTranscriptMeta = { + idempotencyKey?: string; seq?: number; }; @@ -56,6 +57,14 @@ type SessionHistoryRawSnapshot = { transcriptPath?: string; }; +function readMessageIdempotencyKey(message: unknown): string | undefined { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return undefined; + } + const value = (message as Record).idempotencyKey; + return typeof value === "string" && value.trim() ? value : undefined; +} + /** Computes an oversized raw transcript tail window for projected chat history. */ export function resolveSessionHistoryTailReadOptions(limit: number): { maxMessages: number; @@ -258,8 +267,10 @@ export class SessionHistorySseState { } else { this.rawTranscriptSeq += 1; } + const idempotencyKey = readMessageIdempotencyKey(update.message); const nextMessage = attachOpenClawTranscriptMeta(update.message, { ...(typeof update.messageId === "string" ? { id: update.messageId } : {}), + ...(idempotencyKey ? { idempotencyKey } : {}), seq: this.rawTranscriptSeq, }); // Projection can split, drop, or rewrite raw transcript messages. When one diff --git a/src/gateway/session-utils.fs.test.ts b/src/gateway/session-utils.fs.test.ts index d48d6aefd3eb..2d6558107313 100644 --- a/src/gateway/session-utils.fs.test.ts +++ b/src/gateway/session-utils.fs.test.ts @@ -691,6 +691,32 @@ describe("readSessionMessages", () => { }); }); + test("surfaces persisted user idempotency keys in __openclaw metadata (#79844)", async () => { + const sessionId = "test-session-idempotency-key"; + writeTranscript(tmpDir, sessionId, [ + { type: "session", version: 1, id: sessionId }, + { + id: "entry-user-1", + message: { + role: "user", + content: "pending optimistic turn", + idempotencyKey: "client-turn-1", + }, + }, + ]); + + const result = await readRecentSessionMessagesAsync(sessionId, storePath, undefined, { + maxMessages: 5, + maxBytes: 2048, + }); + + expect(result).toHaveLength(1); + expectMessageFields(result[0], { + content: "pending optimistic turn", + openclaw: { id: "entry-user-1", idempotencyKey: "client-turn-1" }, + }); + }); + test("honors byte caps for async recent-message reads", async () => { const sessionId = "test-session-recent-async-byte-cap"; const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); @@ -2651,7 +2677,11 @@ describe("oversized transcript line guards", () => { timestamp, id: "oversized-child", parentId: "root-msg", - message: { role: "assistant", content: oversizedContent }, + message: { + role: "assistant", + content: oversizedContent, + idempotencyKey: "oversized-key", + }, }), ]; fs.writeFileSync(transcriptPath, `${lines.join("\n")}\n`, "utf-8"); @@ -2670,6 +2700,7 @@ describe("oversized transcript line guards", () => { // id is preserved in __openclaw transcript metadata const meta = (oversized as Record>)["__openclaw"]; expect(meta?.id).toBe("oversized-child"); + expect(meta?.idempotencyKey).toBe("oversized-key"); expect(meta?.recordTimestampMs).toBe(Date.parse(timestamp)); // parentId extraction is proven by the record being included: // if parentId was not extracted, the tree would orphan this node. diff --git a/src/gateway/session-utils.fs.ts b/src/gateway/session-utils.fs.ts index 7d94d3b7e30a..22ba456d8e61 100644 --- a/src/gateway/session-utils.fs.ts +++ b/src/gateway/session-utils.fs.ts @@ -31,6 +31,7 @@ import { extractJsonNullableStringFieldPrefix, extractJsonNumberFieldPrefix, extractJsonStringFieldPrefix, + normalizeOptionalString, } from "./session-transcript-json.js"; import type { SessionPreviewItem } from "./session-utils.types.js"; @@ -158,6 +159,14 @@ export function attachOpenClawTranscriptMeta( }; } +function readTranscriptMessageIdempotencyKey(message: unknown): string | undefined { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return undefined; + } + const value = (message as Record).idempotencyKey; + return typeof value === "string" && value.trim() ? value : undefined; +} + /** Read all visible transcript messages for a session from the first existing candidate file. */ export function readSessionMessages( sessionId: string, @@ -301,12 +310,60 @@ async function readRecentTranscriptTailLinesAsync( const MAX_TRANSCRIPT_PARSE_LINE_BYTES = 256 * 1024; const OVERSIZED_TRANSCRIPT_METADATA_PREFIX_CHARS = 64 * 1024; +const OVERSIZED_TRANSCRIPT_METADATA_SUFFIX_CHARS = 64 * 1024; const TRANSCRIPT_OVERSIZED_MESSAGE_PLACEHOLDER = "[chat.history omitted: message too large]"; function isOversizedTranscriptLine(line: string): boolean { return Buffer.byteLength(line, "utf8") > MAX_TRANSCRIPT_PARSE_LINE_BYTES; } +function isJsonObjectFieldToken(source: string, tokenIndex: number): boolean { + for (let index = tokenIndex - 1; index >= 0; index--) { + const char = source[index]; + if (/\s/.test(char)) { + continue; + } + return char === "{" || char === ","; + } + return true; +} + +function extractJsonStringFieldWindow( + source: string, + field: string, + startIndex = 0, + endIndex = source.length, +): string | undefined { + const fieldToken = JSON.stringify(field); + let searchIndex = startIndex; + while (searchIndex < endIndex) { + const tokenIndex = source.indexOf(fieldToken, searchIndex); + if (tokenIndex < 0 || tokenIndex >= endIndex) { + return undefined; + } + searchIndex = tokenIndex + fieldToken.length; + if (!isJsonObjectFieldToken(source, tokenIndex)) { + continue; + } + const match = /^\s*:\s*"((?:\\.|[^"\\])*)"/.exec(source.slice(searchIndex, endIndex)); + if (!match) { + continue; + } + try { + const decoded = JSON.parse(`"${match[1]}"`) as unknown; + return normalizeOptionalString(decoded); + } catch { + return undefined; + } + } + return undefined; +} + +function extractJsonStringFieldSuffix(source: string, field: string): string | undefined { + const startIndex = Math.max(0, source.length - OVERSIZED_TRANSCRIPT_METADATA_SUFFIX_CHARS); + return extractJsonStringFieldWindow(source, field, startIndex); +} + function buildOversizedTranscriptRecord(line: string): TailTranscriptRecord { const prefix = line.slice(0, OVERSIZED_TRANSCRIPT_METADATA_PREFIX_CHARS); const messageMatch = /"message"\s*:/.exec(prefix); @@ -318,6 +375,9 @@ function buildOversizedTranscriptRecord(line: string): TailTranscriptRecord { extractJsonStringFieldPrefix(recordPrefix, "timestamp") ?? extractJsonNumberFieldPrefix(recordPrefix, "timestamp"); const role = extractJsonStringFieldPrefix(prefix, "role") ?? "assistant"; + const idempotencyKey = + extractJsonStringFieldPrefix(prefix, "idempotencyKey") ?? + extractJsonStringFieldSuffix(line, "idempotencyKey"); const record: Record = { ...(type ? { type } : {}), ...(id ? { id } : {}), @@ -325,6 +385,7 @@ function buildOversizedTranscriptRecord(line: string): TailTranscriptRecord { ...(timestamp !== undefined ? { timestamp } : {}), message: { role, + ...(idempotencyKey ? { idempotencyKey } : {}), content: [{ type: "text", text: TRANSCRIPT_OVERSIZED_MESSAGE_PLACEHOLDER }], __openclaw: { truncated: true, reason: "oversized" }, }, @@ -838,8 +899,10 @@ function parsedSessionEntryToMessage(parsed: unknown, seq: number): unknown { : typeof entry.timestamp === "number" ? entry.timestamp : Number.NaN; + const idempotencyKey = readTranscriptMessageIdempotencyKey(entry.message); return attachOpenClawTranscriptMeta(entry.message, { ...(typeof entry.id === "string" ? { id: entry.id } : {}), + ...(idempotencyKey ? { idempotencyKey } : {}), ...(Number.isFinite(recordTimestampMs) ? { recordTimestampMs } : {}), seq, });