Skip to content

Commit 329d170

Browse files
vincentkocopenclaw-clownfish[bot]
authored and
giodl73-repo
committed
fix(reply): dedupe duplicate non-streaming final replies (openclaw#100828)
* fix(reply): dedupe duplicate non-streaming final replies * fix(reply): preserve distinct final delivery identities * chore(reply): drop release-owned changelog entry * chore(reply): drop release-owned changelog entry --------- Co-authored-by: openclaw-clownfish[bot] <280122609+openclaw-clownfish[bot]@users.noreply.github.com>
1 parent 8456e79 commit 329d170

2 files changed

Lines changed: 149 additions & 0 deletions

File tree

‎src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
33
import { clearAgentHarnesses } from "../../agents/harness/registry.js";
44
import type { PluginHookReplyDispatchResult } from "../../plugins/hooks.js";
55
import { createInternalHookEventPayload } from "../../test-utils/internal-hook-event-payload.js";
6+
import { setReplyPayloadMetadata, type ReplyPayload } from "../types.js";
67
import {
78
acpManagerRuntimeMocks,
89
acpMocks,
@@ -350,6 +351,109 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => {
350351
}
351352
});
352353

354+
it("dedupes byte-identical non-streaming final payload entries for one turn", async () => {
355+
hookMocks.runner.hasHooks.mockReturnValue(false);
356+
const dispatcher = createDispatcher();
357+
const replyPayload = {
358+
text: "repeat once",
359+
mediaUrls: ["file:///tmp/repeat.png"],
360+
channelData: { telegram: { parseMode: "MarkdownV2" } },
361+
} satisfies ReplyPayload;
362+
363+
const result = await dispatchReplyFromConfig({
364+
ctx: createHookCtx(),
365+
cfg: emptyConfig,
366+
dispatcher,
367+
replyResolver: async () => [replyPayload, { ...replyPayload }],
368+
});
369+
370+
expect(result.queuedFinal).toBe(true);
371+
expect(dispatcher.sendFinalReply).toHaveBeenCalledOnce();
372+
expect(dispatcher.sendFinalReply).toHaveBeenCalledWith(replyPayload);
373+
});
374+
375+
it("preserves same-content final payloads with distinct route metadata", async () => {
376+
hookMocks.runner.hasHooks.mockReturnValue(false);
377+
const dispatcher = createDispatcher();
378+
const firstReply = setReplyPayloadMetadata(
379+
{ text: "same visible reply" } satisfies ReplyPayload,
380+
{
381+
replyDelivery: { chatType: "channel", replyToMode: "off" },
382+
replyDeliverySource: { channel: "slack", accountId: "primary" },
383+
},
384+
);
385+
const secondReply = setReplyPayloadMetadata(
386+
{ text: "same visible reply" } satisfies ReplyPayload,
387+
{
388+
replyDelivery: { chatType: "channel", replyToMode: "off" },
389+
replyDeliverySource: { channel: "slack", accountId: "secondary" },
390+
},
391+
);
392+
393+
const result = await dispatchReplyFromConfig({
394+
ctx: createHookCtx(),
395+
cfg: emptyConfig,
396+
dispatcher,
397+
replyResolver: async () => [firstReply, secondReply],
398+
});
399+
400+
expect(result.queuedFinal).toBe(true);
401+
expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(2);
402+
expect(dispatcher.sendFinalReply).toHaveBeenNthCalledWith(1, firstReply);
403+
expect(dispatcher.sendFinalReply).toHaveBeenNthCalledWith(2, secondReply);
404+
});
405+
406+
it("preserves same-content final payloads with distinct reply-threading identity", async () => {
407+
hookMocks.runner.hasHooks.mockReturnValue(false);
408+
const dispatcher = createDispatcher();
409+
const implicitReply = {
410+
text: "same threaded reply",
411+
replyToId: "message-1",
412+
} satisfies ReplyPayload;
413+
const explicitReply = setReplyPayloadMetadata(
414+
{
415+
text: "same threaded reply",
416+
replyToId: "message-1",
417+
} satisfies ReplyPayload,
418+
{ replyToIdExplicit: true },
419+
);
420+
421+
await dispatchReplyFromConfig({
422+
ctx: createHookCtx(),
423+
cfg: emptyConfig,
424+
dispatcher,
425+
replyResolver: async () => [implicitReply, explicitReply],
426+
});
427+
428+
expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(2);
429+
expect(dispatcher.sendFinalReply).toHaveBeenNthCalledWith(1, implicitReply);
430+
expect(dispatcher.sendFinalReply).toHaveBeenNthCalledWith(2, explicitReply);
431+
});
432+
433+
it("preserves same-content final payloads from distinct assistant messages", async () => {
434+
hookMocks.runner.hasHooks.mockReturnValue(false);
435+
const dispatcher = createDispatcher();
436+
const firstReply = setReplyPayloadMetadata(
437+
{ text: "intentional repeat" } satisfies ReplyPayload,
438+
{ assistantMessageIndex: 1 },
439+
);
440+
const secondReply = setReplyPayloadMetadata(
441+
{ text: "intentional repeat" } satisfies ReplyPayload,
442+
{ assistantMessageIndex: 2 },
443+
);
444+
445+
await dispatchReplyFromConfig({
446+
ctx: createHookCtx(),
447+
cfg: emptyConfig,
448+
dispatcher,
449+
replyResolver: async () => [firstReply, secondReply],
450+
});
451+
452+
expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(2);
453+
expect(dispatcher.sendFinalReply).toHaveBeenNthCalledWith(1, firstReply);
454+
expect(dispatcher.sendFinalReply).toHaveBeenNthCalledWith(2, secondReply);
455+
});
456+
353457
it("clears the reply lane but defers follow-up admission until final delivery settles", async () => {
354458
const deliveryOrder: string[] = [];
355459
let startDelivery: () => void = () => {};

‎src/auto-reply/reply/dispatch-from-config.ts‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -206,6 +206,45 @@ type InternalReplyResolverOptions = {
206206
onSessionPrepared?: (binding: ReplySessionBinding) => void;
207207
};
208208

209+
function createFinalDispatchPayloadDedupeKey(payload: ReplyPayload): string {
210+
const metadata = getReplyPayloadMetadata(payload);
211+
return JSON.stringify({
212+
payload: {
213+
text: payload.text,
214+
mediaUrl: payload.mediaUrl,
215+
mediaUrls: payload.mediaUrls,
216+
trustedLocalMedia: payload.trustedLocalMedia,
217+
sensitiveMedia: payload.sensitiveMedia,
218+
presentation: payload.presentation,
219+
delivery: payload.delivery,
220+
interactive: payload.interactive,
221+
btw: payload.btw,
222+
replyToId: payload.replyToId,
223+
replyToTag: payload.replyToTag,
224+
replyToCurrent: payload.replyToCurrent,
225+
audioAsVoice: payload.audioAsVoice,
226+
spokenText: payload.spokenText,
227+
ttsSupplement: payload.ttsSupplement,
228+
isError: payload.isError,
229+
isReasoning: payload.isReasoning,
230+
isCommentary: payload.isCommentary,
231+
isReasoningSnapshot: payload.isReasoningSnapshot,
232+
isCompactionNotice: payload.isCompactionNotice,
233+
isFallbackNotice: payload.isFallbackNotice,
234+
isStatusNotice: payload.isStatusNotice,
235+
channelData: payload.channelData,
236+
},
237+
identity: {
238+
assistantMessageIndex: metadata?.assistantMessageIndex,
239+
assistantTranscriptOwned: metadata?.assistantTranscriptOwned,
240+
replyToIdExplicit: metadata?.replyToIdExplicit,
241+
replyDelivery: metadata?.replyDelivery,
242+
replyDeliverySource: metadata?.replyDeliverySource,
243+
sourceReplyTranscriptMirror: metadata?.sourceReplyTranscriptMirror,
244+
},
245+
});
246+
}
247+
209248
class DispatchReplyOperationAbortedError extends Error {
210249
constructor() {
211250
super("Dispatch reply operation aborted");
@@ -3865,6 +3904,7 @@ export async function dispatchReplyFromConfig(
38653904
!sendPolicyDenied &&
38663905
getReplyPayloadMetadata(reply)?.deliverDespiteSourceReplySuppression === true &&
38673906
(ctx.InboundEventKind !== "room_event" || explicitCommandTurnCtx);
3907+
const sentFinalPayloadDedupeKeys = new Set<string>();
38683908
for (const [replyIndex, reply] of replies.entries()) {
38693909
throwIfDispatchOperationAborted();
38703910
// Durable reasoning is a channel-owned lane; generic channels keep the
@@ -3892,6 +3932,11 @@ export async function dispatchReplyFromConfig(
38923932
}
38933933
continue;
38943934
}
3935+
const finalPayloadDedupeKey = createFinalDispatchPayloadDedupeKey(reply);
3936+
if (sentFinalPayloadDedupeKeys.has(finalPayloadDedupeKey)) {
3937+
continue;
3938+
}
3939+
sentFinalPayloadDedupeKeys.add(finalPayloadDedupeKey);
38953940
attemptedFinalDelivery = true;
38963941
const finalReply = await sendFinalPayload(reply, { deliveryId: String(replyIndex) });
38973942
queuedFinal = finalReply.queuedFinal || queuedFinal;

0 commit comments

Comments
 (0)