Skip to content

Commit 741d952

Browse files
zhangguiping-xydtsteipete
zhangguiping-xydt
and
steipete
authored
fix: outbound recovery can replay already sent replies (#99600)
* fix(outbound): avoid replaying sent recovery deliveries * fix(outbound): persist recovery send state before ack Co-authored-by: zhang-guiping <[email protected]> * fix(outbound): preserve partial send evidence across adapters * test(line): cover multi-send delivery progress * fix(outbound): preserve repeated receipt result multiplicity * test(heartbeat): expect delivery progress callback --------- Co-authored-by: Peter Steinberger <[email protected]>
1 parent 2abad57 commit 741d952

56 files changed

Lines changed: 2640 additions & 355 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎extensions/discord/src/outbound-adapter.test.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -309,6 +309,16 @@ describe("discordOutbound", () => {
309309
});
310310

311311
it("routes audioAsVoice payloads through the Discord voice send helper", async () => {
312+
const onDeliveryResult = vi.fn();
313+
hoisted.sendMessageDiscordMock.mockImplementation(
314+
async (_to: unknown, _text: unknown, options: unknown) => {
315+
const deliveryResult = { messageId: "msg-1", channelId: "ch-1" };
316+
const onProgress = (options as { onDeliveryResult?: (result: unknown) => Promise<void> })
317+
.onDeliveryResult;
318+
await onProgress?.(deliveryResult);
319+
return deliveryResult;
320+
},
321+
);
312322
const result = await discordOutbound.sendPayload?.({
313323
cfg: {},
314324
to: "channel:123456",
@@ -321,6 +331,7 @@ describe("discordOutbound", () => {
321331
accountId: "default",
322332
replyToId: "reply-1",
323333
replyToMode: "first",
334+
onDeliveryResult,
324335
});
325336

326337
const voiceCall = mockCall(hoisted.sendVoiceMessageDiscordMock, "sendVoiceMessageDiscord");
@@ -359,6 +370,11 @@ describe("discordOutbound", () => {
359370
messageId: "msg-1",
360371
channelId: "ch-1",
361372
});
373+
expect(onDeliveryResult.mock.calls.map((call) => call[0]?.messageId)).toEqual([
374+
"voice-1",
375+
"msg-1",
376+
"msg-1",
377+
]);
362378
});
363379

364380
it("keeps captured replyTo on audioAsVoice sends when replyToMode is batched", async () => {

‎extensions/discord/src/outbound-adapter.ts‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
import type { OutboundIdentity } from "openclaw/plugin-sdk/channel-outbound";
33
import { resolveOutboundSendDep } from "openclaw/plugin-sdk/channel-outbound";
44
import {
5+
attachChannelToResult,
56
type ChannelOutboundAdapter,
67
createAttachedChannelResultAdapter,
78
} from "openclaw/plugin-sdk/channel-send-result";
@@ -174,6 +175,7 @@ export const discordOutbound: ChannelOutboundAdapter = {
174175
identity,
175176
silent,
176177
formatting,
178+
onDeliveryResult,
177179
}) => {
178180
if (!silent) {
179181
const webhookResult = await maybeSendDiscordWebhookText({
@@ -202,6 +204,11 @@ export const discordOutbound: ChannelOutboundAdapter = {
202204
silent: silent ?? undefined,
203205
cfg,
204206
...resolveDiscordFormattingOptions({ formatting }),
207+
onDeliveryResult: onDeliveryResult
208+
? async (result) => {
209+
await onDeliveryResult(attachChannelToResult("discord", result));
210+
}
211+
: undefined,
205212
}),
206213
});
207214
},
@@ -220,6 +227,7 @@ export const discordOutbound: ChannelOutboundAdapter = {
220227
threadId,
221228
silent,
222229
formatting,
230+
onDeliveryResult,
223231
}) => {
224232
const send =
225233
resolveOutboundSendDep<DiscordSendFn>(deps, "discord") ??
@@ -254,6 +262,11 @@ export const discordOutbound: ChannelOutboundAdapter = {
254262
silent: silent ?? undefined,
255263
cfg,
256264
...formattingOptions,
265+
onDeliveryResult: onDeliveryResult
266+
? async (result) => {
267+
await onDeliveryResult(attachChannelToResult("discord", result));
268+
}
269+
: undefined,
257270
}),
258271
});
259272
return await withDiscordDeliveryRetry({
@@ -270,6 +283,11 @@ export const discordOutbound: ChannelOutboundAdapter = {
270283
silent: silent ?? undefined,
271284
cfg,
272285
...formattingOptions,
286+
onDeliveryResult: onDeliveryResult
287+
? async (result) => {
288+
await onDeliveryResult(attachChannelToResult("discord", result));
289+
}
290+
: undefined,
273291
}),
274292
});
275293
}
@@ -288,6 +306,11 @@ export const discordOutbound: ChannelOutboundAdapter = {
288306
silent: silent ?? undefined,
289307
cfg,
290308
...formattingOptions,
309+
onDeliveryResult: onDeliveryResult
310+
? async (result) => {
311+
await onDeliveryResult(attachChannelToResult("discord", result));
312+
}
313+
: undefined,
291314
}),
292315
});
293316
},

‎extensions/discord/src/outbound-payload.ts‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,14 @@ type DiscordOutboundPayloadContext = Parameters<
2323
>[0];
2424
type DiscordPayloadSendContext = Awaited<ReturnType<typeof createDiscordPayloadSendContext>>;
2525

26+
function resolveDiscordDeliveryProgress(ctx: DiscordOutboundPayloadContext) {
27+
return ctx.onDeliveryResult
28+
? async (result: Awaited<ReturnType<DiscordPayloadSendContext["send"]>>) => {
29+
await ctx.onDeliveryResult?.(attachChannelToResult("discord", result));
30+
}
31+
: undefined;
32+
}
33+
2634
function createDiscordUnknownPayloadResult(target: string) {
2735
return {
2836
messageId: "",
@@ -94,13 +102,15 @@ export async function sendDiscordOutboundPayload(params: {
94102
replyTo: voiceReplyTo,
95103
}),
96104
);
105+
await ctx.onDeliveryResult?.(attachChannelToResult("discord", lastResult));
97106
if (payload.text?.trim()) {
98107
lastResult = await sendContext.withRetry(
99108
async () =>
100109
await sendContext.send(sendContext.target, payload.text, {
101110
verbose: false,
102111
...resolveDiscordFormattedDeliveryOptions(ctx, sendContext),
103112
replyTo: voiceReplyTo,
113+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
104114
}),
105115
);
106116
}
@@ -111,6 +121,7 @@ export async function sendDiscordOutboundPayload(params: {
111121
verbose: false,
112122
...resolveDiscordMediaDeliveryOptions(ctx, sendContext, mediaUrl),
113123
replyTo: voiceReplyTo,
124+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
114125
}),
115126
);
116127
}
@@ -146,6 +157,7 @@ export async function sendDiscordOutboundPayload(params: {
146157
embeds,
147158
filename,
148159
...resolveDiscordFormattedDeliveryOptions(ctx, sendContext),
160+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
149161
}),
150162
),
151163
send: async ({ text, mediaUrl, isFirst }) =>
@@ -157,6 +169,7 @@ export async function sendDiscordOutboundPayload(params: {
157169
components: isFirst ? nativeComponents : undefined,
158170
embeds: isFirst ? embeds : undefined,
159171
filename: isFirst ? filename : undefined,
172+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
160173
}),
161174
),
162175
});
@@ -176,19 +189,22 @@ export async function sendDiscordOutboundPayload(params: {
176189
text: payload.text ?? "",
177190
mediaUrls,
178191
fallbackResult: createDiscordUnknownPayloadResult(sendContext.target),
179-
sendNoMedia: async () =>
180-
await sendContext.withRetry(
192+
sendNoMedia: async () => {
193+
return await sendContext.withRetry(
181194
async () =>
182195
await sendDiscordComponentMessageLazy(sendContext.target, componentSpec, {
183196
...resolveDiscordFormattedDeliveryOptions(ctx, sendContext),
197+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
184198
}),
185-
),
199+
);
200+
},
186201
send: async ({ text, mediaUrl, isFirst }) => {
187202
if (isFirst) {
188203
return await sendContext.withRetry(
189204
async () =>
190205
await sendDiscordComponentMessageLazy(sendContext.target, componentSpec, {
191206
...resolveDiscordMediaDeliveryOptions(ctx, sendContext, mediaUrl),
207+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
192208
}),
193209
);
194210
}
@@ -197,6 +213,7 @@ export async function sendDiscordOutboundPayload(params: {
197213
await sendContext.send(sendContext.target, text, {
198214
verbose: false,
199215
...resolveDiscordMediaDeliveryOptions(ctx, sendContext, mediaUrl),
216+
onDeliveryResult: resolveDiscordDeliveryProgress(ctx),
200217
}),
201218
);
202219
},

‎extensions/discord/src/send.components.test.ts‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,27 @@ describe("sendDiscordComponentMessage", () => {
130130
);
131131
});
132132

133+
it("reports the platform send before component registry bookkeeping", async () => {
134+
const { rest, postMock, getMock } = makeDiscordRest();
135+
getMock.mockResolvedValueOnce({ type: ChannelType.GuildText, id: "chan-1" });
136+
postMock.mockResolvedValueOnce({ id: "msg-progress", channel_id: "chan-1" });
137+
registerMock.mockImplementationOnce(() => {
138+
throw new Error("registry write failed");
139+
});
140+
const onDeliveryResult = vi.fn();
141+
142+
await expect(
143+
sendDiscordComponentMessage(
144+
"channel:chan-1",
145+
{ blocks: [{ type: "actions", buttons: [{ label: "Tap" }] }] },
146+
{ cfg: DISCORD_TEST_CFG, rest, token: "t", onDeliveryResult },
147+
),
148+
).rejects.toThrow("registry write failed");
149+
150+
expect(onDeliveryResult).toHaveBeenCalledOnce();
151+
expect(onDeliveryResult.mock.calls[0]?.[0]?.messageId).toBe("msg-progress");
152+
});
153+
133154
it("edits component messages and refreshes component registry entries", async () => {
134155
const { rest, patchMock, getMock } = makeDiscordRest();
135156
getMock.mockResolvedValueOnce({
@@ -263,6 +284,7 @@ describe("sendDiscordComponentMessage classic message downgrade", () => {
263284
it("forwards mediaReadFile and mediaAccess to sendMessageDiscord", async () => {
264285
const readFileMock = vi.fn().mockResolvedValue(Buffer.from("pdf"));
265286
const mediaAccess = { localRoots: ["/tmp"], readFile: readFileMock };
287+
const onDeliveryResult = vi.fn();
266288

267289
await sendDiscordComponentMessage(
268290
"channel:chan-1",
@@ -273,6 +295,7 @@ describe("sendDiscordComponentMessage classic message downgrade", () => {
273295
mediaUrl: "https://example.com/report.pdf",
274296
mediaReadFile: readFileMock,
275297
mediaAccess,
298+
onDeliveryResult,
276299
},
277300
);
278301

@@ -296,6 +319,7 @@ describe("sendDiscordComponentMessage classic message downgrade", () => {
296319
maxLinesPerMessage: undefined,
297320
tableMode: undefined,
298321
chunkMode: undefined,
322+
onDeliveryResult,
299323
},
300324
]);
301325
});

‎extensions/discord/src/send.components.ts‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,8 @@ type DiscordComponentSendOpts = {
167167
tableMode?: MarkdownTableMode;
168168
chunkMode?: ChunkMode;
169169
suppressEmbeds?: boolean;
170+
/** Persist the concrete platform send before component bookkeeping can fail. */
171+
onDeliveryResult?: (result: DiscordSendResult) => Promise<void> | void;
170172
};
171173

172174
export function registerBuiltDiscordComponentMessage(params: {
@@ -285,6 +287,7 @@ export async function sendDiscordComponentMessage(
285287
maxLinesPerMessage: opts.maxLinesPerMessage,
286288
tableMode: opts.tableMode,
287289
chunkMode: opts.chunkMode,
290+
onDeliveryResult: opts.onDeliveryResult,
288291
...(opts.suppressEmbeds === undefined ? {} : { suppressEmbeds: opts.suppressEmbeds }),
289292
});
290293
}
@@ -326,6 +329,14 @@ export async function sendDiscordComponentMessage(
326329
});
327330
}
328331

332+
const deliveryResult = createDiscordSendResult({
333+
result,
334+
fallbackChannelId: channelId,
335+
kind: "card",
336+
...(opts.replyTo ? { replyToId: opts.replyTo } : {}),
337+
});
338+
await opts.onDeliveryResult?.(deliveryResult);
339+
329340
registerBuiltDiscordComponentMessage({
330341
buildResult,
331342
messageId: result.id,
@@ -338,12 +349,7 @@ export async function sendDiscordComponentMessage(
338349
direction: "outbound",
339350
});
340351

341-
return createDiscordSendResult({
342-
result,
343-
fallbackChannelId: channelId,
344-
kind: "card",
345-
...(opts.replyTo ? { replyToId: opts.replyTo } : {}),
346-
});
352+
return deliveryResult;
347353
}
348354

349355
export async function editDiscordComponentMessage(

0 commit comments

Comments
 (0)