mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-09 08:10:32 +00:00
fix(gateway): expose idempotencyKey in chat history metadata (#96273)
* 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: 吴杨帆 <39647285+leno23@users.noreply.github.com>
This commit is contained in:
@@ -1649,6 +1649,8 @@ 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,
|
||||
@@ -1656,6 +1658,7 @@ export function buildOversizedHistoryPlaceholder(message?: unknown): Record<stri
|
||||
__openclaw: {
|
||||
...(metadataId ? { id: metadataId } : {}),
|
||||
...(metadataSeq !== undefined ? { seq: metadataSeq } : {}),
|
||||
...(metadataIdempotencyKey ? { idempotencyKey: metadataIdempotencyKey } : {}),
|
||||
truncated: true,
|
||||
reason: "oversized",
|
||||
},
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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)],
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,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<string, unknown> = {
|
||||
...(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,
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user