fix(gateway): expose idempotencyKey in chat history metadata (#96273)

wyf027 · web-flow · commit f9dddea72a7b · 2026-06-28T11:07:51.000-07:00
* fix(gateway): surface idempotency key in history metadata

* test(gateway): avoid unsafe optional chain

* fix(gateway): preserve oversized transcript idempotency keys

* fix(gateway): preserve oversized idempotency keys

* fix(gateway): reuse transcript json helpers

---------

Co-authored-by: 吴杨帆 &lt;39647285+leno23@users.noreply.github.com&gt;
diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts
@@ -1649,13 +1649,16 @@ export function buildOversizedHistoryPlaceholder(message?: unknown): Record<stri
       : {};
   const metadataId = typeof metadata.id === "string" ? metadata.id : undefined;
   const metadataSeq = typeof metadata.seq === "number" ? metadata.seq : undefined;
+  const metadataIdempotencyKey =
+    typeof metadata.idempotencyKey === "string" ? metadata.idempotencyKey : undefined;
   return {
     role,
     timestamp,
     content: [{ type: "text", text: CHAT_HISTORY_OVERSIZED_PLACEHOLDER }],
     __openclaw: {
       ...(metadataId ? { id: metadataId } : {}),
       ...(metadataSeq !== undefined ? { seq: metadataSeq } : {}),
+      ...(metadataIdempotencyKey ? { idempotencyKey: metadataIdempotencyKey } : {}),
       truncated: true,
       reason: "oversized",
     },
diff --git a/src/gateway/server-session-events.test.ts b/src/gateway/server-session-events.test.ts
@@ -43,12 +43,15 @@ function createHandler(projectSessionActive: boolean) {
   return { broadcastToConnIds, handler };
 }
 
-async function emitAssistantTranscriptUpdate(projectSessionActive: boolean) {
+async function emitAssistantTranscriptUpdate(
+  projectSessionActive: boolean,
+  message: unknown = { role: "assistant", content: [{ type: "text", text: "Final answer" }] },
+) {
   const { broadcastToConnIds, handler } = createHandler(projectSessionActive);
   handler({
     sessionFile: "/tmp/sess-main.jsonl",
     sessionKey: "agent:main:main",
-    message: { role: "assistant", content: [{ type: "text", text: "Final answer" }] },
+    message,
     messageId: "message-1",
     messageSeq: 1,
   });
@@ -79,4 +82,22 @@ describe("createTranscriptUpdateBroadcastHandler", () => {
       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
@@ -30,6 +30,14 @@ import {
 type SessionEventSubscribers = Pick<SessionEventSubscriberRegistry, "getAll">;
 type SessionMessageSubscribers = Pick<SessionMessageSubscriberRegistry, "get">;
 
+function readMessageIdempotencyKey(message: unknown): string | undefined {
+  if (!message || typeof message !== "object" || Array.isArray(message)) {
+    return undefined;
+  }
+  const value = (message as Record<string, unknown>).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
@@ -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
@@ -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<string, unknown>).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
@@ -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<string, Record<string, unknown>>)["__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
@@ -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<string, unknown>).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,13 +375,17 @@ 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<string, unknown> = {
     ...(type ? { type } : {}),
     ...(id ? { id } : {}),
     ...(parentId !== undefined ? { parentId } : {}),
     ...(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,
     });
