diff --git a/extensions/discord/src/monitor/agent-components.dispatch.ts b/extensions/discord/src/monitor/agent-components.dispatch.ts index c88fbb932d58..b4edf00e08fd 100644 --- a/extensions/discord/src/monitor/agent-components.dispatch.ts +++ b/extensions/discord/src/monitor/agent-components.dispatch.ts @@ -335,6 +335,7 @@ export async function dispatchDiscordComponentEvent(params: { mediaLocalRoots, kind: info.kind, bindPendingFinalDelivery: info.bindPendingFinalDelivery, + onPlatformSendDispatch: info.onPlatformSendDispatch, }); if (result.visibleReplySent) { replyReference.markSent(); diff --git a/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts b/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts index f16c5ef7d806..a694c6854a6b 100644 --- a/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts +++ b/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts @@ -236,6 +236,7 @@ describe("processDiscordMessage draft streaming final delivery", () => { expect(editMessageDiscord).not.toHaveBeenCalled(); expect(firstMockArg(deliverDiscordReply, "deliverDiscordReply")).toMatchObject({ allowedMentions: { parse: ["users", "roles"] }, + onPlatformSendDispatch: expect.any(Function), }); }); diff --git a/extensions/discord/src/monitor/message-handler.process.ts b/extensions/discord/src/monitor/message-handler.process.ts index f9abbb0d61f2..a4b56e5b2fe7 100644 --- a/extensions/discord/src/monitor/message-handler.process.ts +++ b/extensions/discord/src/monitor/message-handler.process.ts @@ -72,6 +72,12 @@ type DiscordMessageProcessObserver = { onReplyPlanResolved?: (params: { createdThreadId?: string; sessionKey?: string }) => void; }; +type DiscordProviderDeliveryInfo = { + kind: ReplyDispatchKind; + bindPendingFinalDelivery?: (payload: T) => T; + onPlatformSendDispatch: () => Promise; +}; + export async function processDiscordMessage( ctx: DiscordMessagePreflightContext, observer?: DiscordMessageProcessObserver, @@ -217,7 +223,7 @@ async function processDiscordMessageInner( let userFacingFinalDelivered = false; let userFacingFinalDeliveryFailed = false; let pendingToolWarningFinal: - | { payload: ReplyPayload; info: { kind: ReplyDispatchKind } } + | { payload: ReplyPayload; info: DiscordProviderDeliveryInfo } | undefined; const markFinalReplyDelivered = (isError = false) => { draftPreview.markFinalReplyDelivered(isError); @@ -265,10 +271,7 @@ async function processDiscordMessageInner( const deliverDiscordPayload = async ( payload: ReplyPayload, - info: { - kind: ReplyDispatchKind; - bindPendingFinalDelivery?: (payload: T) => T; - }, + info: DiscordProviderDeliveryInfo, options?: { allowFallbackOnlyToolWarning?: boolean; allowProgressBlock?: boolean; @@ -326,6 +329,7 @@ async function processDiscordMessageInner( mediaLocalRoots, kind: "block", bindPendingFinalDelivery: info.bindPendingFinalDelivery, + onPlatformSendDispatch: info.onPlatformSendDispatch, }); if (result.visibleReplySent) { replyReference.markSent(); @@ -481,6 +485,7 @@ async function processDiscordMessageInner( allowedMentions, kind: info.kind, bindPendingFinalDelivery: info.bindPendingFinalDelivery, + onPlatformSendDispatch: info.onPlatformSendDispatch, }); return deliveryResult.visibleReplySent; }, @@ -530,6 +535,7 @@ async function processDiscordMessageInner( mediaLocalRoots, kind: info.kind, bindPendingFinalDelivery: info.bindPendingFinalDelivery, + onPlatformSendDispatch: info.onPlatformSendDispatch, }); if (!result.visibleReplySent) { return result; @@ -683,7 +689,16 @@ async function processDiscordMessageInner( return; } dispatchError = true; - if (await completeDiscordSessionConflict(err, deliverDiscordPayload, onDiscordDeliveryError)) { + const conflictCompleted = await completeDiscordSessionConflict( + err, + (payload, info) => + deliverDiscordPayload(payload, { + ...info, + onPlatformSendDispatch: () => Promise.resolve(), + }), + onDiscordDeliveryError, + ); + if (conflictCompleted) { // The visible terminal notice owns this event, so replay can commit. return; } diff --git a/extensions/discord/src/monitor/reply-delivery.test.ts b/extensions/discord/src/monitor/reply-delivery.test.ts index c86cd0a29641..00289e979ba1 100644 --- a/extensions/discord/src/monitor/reply-delivery.test.ts +++ b/extensions/discord/src/monitor/reply-delivery.test.ts @@ -119,6 +119,7 @@ describe("deliverDiscordReply", () => { it("bridges regular replies to shared outbound with Discord package deps", async () => { const rest = {} as RequestClient; const replies = [{ text: "shared path" }]; + const onPlatformSendDispatch = vi.fn(async () => undefined); await deliverDiscordReply({ replies, @@ -133,11 +134,13 @@ describe("deliverDiscordReply", () => { replyToMode: "all", allowedMentions: { parse: [] }, kind: "final", + onPlatformSendDispatch, }); const params = firstDeliverParams(); expect(params.channel).toBe("discord"); expect(params.to).toBe("channel:101"); + expect(params.onPlatformSendDispatch).toBe(onPlatformSendDispatch); expect(params.accountId).toBe("default"); expect(params.payloads).toEqual(replies); expect(params.replyToId).toBe("reply-1"); diff --git a/extensions/discord/src/monitor/reply-delivery.ts b/extensions/discord/src/monitor/reply-delivery.ts index 052726677435..705bfd20325d 100644 --- a/extensions/discord/src/monitor/reply-delivery.ts +++ b/extensions/discord/src/monitor/reply-delivery.ts @@ -230,6 +230,7 @@ export async function deliverDiscordReply(params: { allowedMentions?: DiscordAllowedMentions; kind: "tool" | "block" | "final"; bindPendingFinalDelivery?: (payload: T) => T; + onPlatformSendDispatch?: () => Promise; }) { void params.runtime; @@ -257,6 +258,7 @@ export async function deliverDiscordReply(params: { formatting: delivery.formatting, threadId: delivery.threadId, identity: delivery.identity, + onPlatformSendDispatch: params.onPlatformSendDispatch, deps: createDiscordDeliveryDeps({ cfg: params.cfg, token: params.token, diff --git a/extensions/discord/src/send.components.test.ts b/extensions/discord/src/send.components.test.ts index 15ab16c464d9..d4c8cb060d3f 100644 --- a/extensions/discord/src/send.components.test.ts +++ b/extensions/discord/src/send.components.test.ts @@ -185,6 +185,46 @@ describe("sendDiscordComponentMessage", () => { expect(onDeliveryResult.mock.calls[0]?.[0]?.messageId).toBe("msg-progress"); }); + it("rechecks delivery authority before each retried component post", async () => { + let authorityActive = true; + const loopback = await createDiscordLoopbackRest({ + status: (request) => { + if (request.method === "POST") { + authorityActive = false; + return 503; + } + return 200; + }, + }); + try { + const authorityRevoked = new Error("delivery authority revoked"); + const onPlatformSendDispatch = vi.fn(async () => { + if (!authorityActive) { + throw authorityRevoked; + } + }); + + await expect( + sendDiscordComponentMessage( + "channel:789", + { blocks: [{ type: "actions", buttons: [{ label: "Open" }] }] }, + { + cfg: DISCORD_TEST_CFG, + rest: loopback.rest, + token: "test-token", + onPlatformSendDispatch, + }, + ), + ).rejects.toBe(authorityRevoked); + + expect(onPlatformSendDispatch).toHaveBeenCalledTimes(2); + const messageRequests = loopback.requests.filter((request) => request.method === "POST"); + expect(messageRequests).toHaveLength(1); + } finally { + await loopback.close(); + } + }); + it("edits component messages and refreshes component registry entries", async () => { const { rest, patchMock, getMock } = makeDiscordRest(); getMock.mockResolvedValueOnce({ diff --git a/extensions/discord/src/send.components.ts b/extensions/discord/src/send.components.ts index 84e8dd4442ab..8a6b1ba4acc6 100644 --- a/extensions/discord/src/send.components.ts +++ b/extensions/discord/src/send.components.ts @@ -323,12 +323,13 @@ export async function sendDiscordComponentMessage( let result: { id: string; channel_id: string }; try { - await opts.onPlatformSendDispatch?.(); result = (await request( - () => - createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { + async () => { + await opts.onPlatformSendDispatch?.(); + return createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { body, - }), + }); + }, "components", { safety: "nonce-protected-create" }, )) as { id: string; channel_id: string }; diff --git a/extensions/discord/src/send.outbound.ts b/extensions/discord/src/send.outbound.ts index e884fb665266..749f8886124e 100644 --- a/extensions/discord/src/send.outbound.ts +++ b/extensions/discord/src/send.outbound.ts @@ -242,10 +242,10 @@ export async function sendMessageDiscord( }); let threadRes: { id: string; message?: { id: string; channel_id: string } }; try { - await opts.onPlatformSendDispatch?.(); threadRes = (await request( - () => - createThread<{ id: string; message?: { id: string; channel_id: string } }>( + async () => { + await opts.onPlatformSendDispatch?.(); + return createThread<{ id: string; message?: { id: string; channel_id: string } }>( rest, channelId, { @@ -259,7 +259,8 @@ export async function sendMessageDiscord( message: starterBody, }, }, - ), + ); + }, "forum-thread", { safety: "non-idempotent-create" }, )) as { id: string; message?: { id: string; channel_id: string } }; @@ -506,9 +507,13 @@ async function resolveDiscordStructuredSendContext( : undefined; return { send: async (kind, body) => { - await opts.onPlatformSendDispatch?.(); const result = (await request( - () => createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { body }), + async () => { + await opts.onPlatformSendDispatch?.(); + return createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { + body, + }); + }, kind, { safety: "nonce-protected-create" }, )) as { id: string; channel_id: string }; diff --git a/extensions/discord/src/send.sends-basic-channel-messages.test.ts b/extensions/discord/src/send.sends-basic-channel-messages.test.ts index 29a1d77e7ae8..a25b2d4bcd39 100644 --- a/extensions/discord/src/send.sends-basic-channel-messages.test.ts +++ b/extensions/discord/src/send.sends-basic-channel-messages.test.ts @@ -502,6 +502,79 @@ describe("sendMessageDiscord", () => { expect(onDeliveryResult.mock.calls.map((call) => call[0]?.messageId)).toEqual(["msg1"]); }); + it("rechecks delivery authority before media caption follow-up chunks", async () => { + const loopback = await createDiscordLoopbackRest(); + try { + const authorityRevoked = new Error("delivery authority revoked"); + let authorityActive = true; + const onPlatformSendDispatch = vi.fn(async () => { + if (!authorityActive) { + throw authorityRevoked; + } + }); + const onDeliveryResult = vi.fn(async () => { + authorityActive = false; + }); + + await expect( + sendMessageDiscord("channel:789", "a".repeat(2_500), { + rest: loopback.rest, + token: "test-token", + cfg: DISCORD_TEST_CFG, + mediaUrl: "file:///tmp/photo.jpg", + onDeliveryResult, + onPlatformSendDispatch, + }), + ).rejects.toBe(authorityRevoked); + + expect(onDeliveryResult).toHaveBeenCalledOnce(); + expect(onPlatformSendDispatch).toHaveBeenCalledTimes(2); + const messageRequests = loopback.requests.filter((request) => request.method === "POST"); + expect(messageRequests).toHaveLength(1); + expect(messageRequests[0]?.path).toContain("/channels/789/messages"); + expect(messageRequests[0]?.contentType).toMatch(/^multipart\/form-data; boundary=/); + } finally { + await loopback.close(); + } + }); + + it("rechecks delivery authority before each retried text post", async () => { + let authorityActive = true; + const loopback = await createDiscordLoopbackRest({ + status: (request) => { + if (request.method === "POST") { + authorityActive = false; + return 503; + } + return 200; + }, + }); + try { + const authorityRevoked = new Error("delivery authority revoked"); + const onPlatformSendDispatch = vi.fn(async () => { + if (!authorityActive) { + throw authorityRevoked; + } + }); + + await expect( + sendMessageDiscord("channel:789", "retry once", { + rest: loopback.rest, + token: "test-token", + cfg: DISCORD_TEST_CFG, + retry: { attempts: 2, minDelayMs: 0, maxDelayMs: 0, jitter: 0 }, + onPlatformSendDispatch, + }), + ).rejects.toBe(authorityRevoked); + + expect(onPlatformSendDispatch).toHaveBeenCalledTimes(2); + const messageRequests = loopback.requests.filter((request) => request.method === "POST"); + expect(messageRequests).toHaveLength(1); + } finally { + await loopback.close(); + } + }); + it("allows Discord link embeds when suppressEmbeds is disabled", async () => { const { rest, postMock, getMock } = makeDiscordRest(); getMock.mockResolvedValueOnce({ type: ChannelType.GuildText }); diff --git a/extensions/discord/src/send.shared.ts b/extensions/discord/src/send.shared.ts index ef86ca1165c2..ef3040fb7088 100644 --- a/extensions/discord/src/send.shared.ts +++ b/extensions/discord/src/send.shared.ts @@ -374,9 +374,11 @@ async function sendDiscordText(params: DiscordTextSendParams) { flags, replyTo: chunkReplyTo, }); - await onPlatformSendDispatch?.(); const result = (await request( - () => createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { body }), + async () => { + await onPlatformSendDispatch?.(); + return createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { body }); + }, "text", { safety: "nonce-protected-create" }, )) as { id: string; channel_id: string }; @@ -479,9 +481,11 @@ async function sendDiscordMedia(params: DiscordMediaSendParams) { }); let res: { id: string; channel_id: string }; try { - await onPlatformSendDispatch?.(); res = (await request( - () => createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { body }), + async () => { + await onPlatformSendDispatch?.(); + return createChannelMessage<{ id: string; channel_id: string }>(rest, channelId, { body }); + }, "media", { safety: "nonce-protected-create" }, )) as { id: string; channel_id: string }; @@ -527,6 +531,7 @@ async function sendDiscordMedia(params: DiscordMediaSendParams) { allowedMentions, maxChars, onResult, + onPlatformSendDispatch, }); for (const id of followup.platformMessageIds) { if (id) { diff --git a/extensions/discord/src/send.test-harness.ts b/extensions/discord/src/send.test-harness.ts index 7384792ee2af..281d4b5d64df 100644 --- a/extensions/discord/src/send.test-harness.ts +++ b/extensions/discord/src/send.test-harness.ts @@ -59,6 +59,7 @@ export function timerDelayAt(source: MockCallSource, callIndex = 0) { export async function createDiscordLoopbackRest(options?: { respond?: (request: DiscordLoopbackRequest) => unknown; + status?: (request: DiscordLoopbackRequest) => number; }): Promise<{ rest: RequestClient; requests: DiscordLoopbackRequest[]; @@ -77,7 +78,9 @@ export async function createDiscordLoopbackRest(options?: { path: request.url, }; requests.push(received); - response.writeHead(200, { "Content-Type": "application/json" }); + response.writeHead(options?.status?.(received) ?? 200, { + "Content-Type": "application/json", + }); response.end( JSON.stringify( options?.respond?.(received) ?? diff --git a/extensions/discord/src/send.voice.ts b/extensions/discord/src/send.voice.ts index 7cc39f8df827..e0e8583ee7f9 100644 --- a/extensions/discord/src/send.voice.ts +++ b/extensions/discord/src/send.voice.ts @@ -119,7 +119,6 @@ export async function sendVoiceMessageDiscord( const metadata = await getVoiceMessageMetadata(oggPath); const audioBuffer = await fs.readFile(oggPath); - await opts.onPlatformSendDispatch?.(); const result = await sendDiscordVoiceMessage( rest, channelId, @@ -129,6 +128,7 @@ export async function sendVoiceMessageDiscord( request, opts.silent, token, + opts.onPlatformSendDispatch, ); recordChannelActivity({ diff --git a/extensions/discord/src/voice-message.ts b/extensions/discord/src/voice-message.ts index cc928ae1de61..ed7db5d7521e 100644 --- a/extensions/discord/src/voice-message.ts +++ b/extensions/discord/src/voice-message.ts @@ -405,6 +405,7 @@ export async function sendDiscordVoiceMessage( request: DiscordRetryRunner, silent?: boolean, token?: string, + onPlatformSendDispatch?: () => Promise, ): Promise<{ id: string; channel_id: string }> { const filename = "voice-message.ogg"; const fileSize = audioBuffer.byteLength; @@ -480,6 +481,7 @@ export async function sendDiscordVoiceMessage( try { return (await request( async () => { + await onPlatformSendDispatch?.(); try { return (await rest.post(`/channels/${channelId}/messages`, { body: messagePayload, diff --git a/src/channels/turn/direct-delivery-custody.ts b/src/channels/turn/direct-delivery-custody.ts index 47e1d9cec217..fe61d365efea 100644 --- a/src/channels/turn/direct-delivery-custody.ts +++ b/src/channels/turn/direct-delivery-custody.ts @@ -28,23 +28,29 @@ export function createDirectPendingFinalCustody( return undefined; } const { kind: _kind, ...identity } = completion; - let admission: Promise | undefined; + let firstDispatch = true; + let admissionTail = Promise.resolve(); return { bindPendingFinalDelivery: (nextPayload) => setReplyPayloadMetadata(nextPayload, { pendingFinalDeliveryCompletion: identity, }), onPlatformSendDispatch: () => { - admission ??= settlePendingFinalDelivery(completion, "unknown", ["prepared", "queued"]).then( - (result) => { - if (result.state !== "unknown") { - throw new PlatformMessageNotDispatchedError( - "Pending final delivery ownership changed before platform dispatch", - { cause: new Error(`pending final delivery is ${result.state}`) }, - ); - } - }, - ); + const expectedStates = firstDispatch + ? (["prepared", "queued"] as const) + : (["unknown"] as const); + firstDispatch = false; + const admission = admissionTail.then(async () => { + const result = await settlePendingFinalDelivery(completion, "unknown", expectedStates); + if (result.state !== "unknown") { + throw new PlatformMessageNotDispatchedError( + "Pending final delivery ownership changed before platform dispatch", + { cause: new Error(`pending final delivery is ${result.state}`) }, + ); + } + }); + // Every physical post must observe the state left by the prior post's check. + admissionTail = admission.catch(() => undefined); return admission; }, }; diff --git a/src/channels/turn/run-channel-turn.custody.test.ts b/src/channels/turn/run-channel-turn.custody.test.ts index d77ae2b4042a..0920b18d53e0 100644 --- a/src/channels/turn/run-channel-turn.custody.test.ts +++ b/src/channels/turn/run-channel-turn.custody.test.ts @@ -4,6 +4,7 @@ import type { DispatchReplyWithDispatcher } from "../../auto-reply/reply/provide import type { FinalizedMsgContext } from "../../auto-reply/templating.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { PlatformMessageNotDispatchedError } from "../../infra/outbound/deliver-types.js"; +import { createDirectPendingFinalCustody } from "./direct-delivery-custody.js"; import { dispatchRoutedChannelTurn } from "./lifecycle.js"; const dispatchReplyWithRoutedChannelDispatcherCore = vi.hoisted(() => vi.fn()); @@ -69,6 +70,50 @@ describe("channel turn failed-send custody", () => { })); }); + it("serializes and revalidates pending-final custody before every provider post", async () => { + const payload = setReplyPayloadMetadata( + { text: "reply" }, + { pendingFinalDeliveryCompletion: completion }, + ); + const custody = createDirectPendingFinalCustody(payload); + if (!custody) { + throw new Error("expected pending-final custody"); + } + let resolveFirstCheck: ((result: { state: "unknown" }) => void) | undefined; + const firstCheck = new Promise<{ state: "unknown" }>((resolve) => { + resolveFirstCheck = resolve; + }); + let checkCount = 0; + settlePendingFinalDelivery.mockImplementation(async () => { + if (checkCount++ === 0) { + return firstCheck; + } + return { state: "suppressed" }; + }); + + const firstDispatch = custody.onPlatformSendDispatch(); + const secondDispatch = custody.onPlatformSendDispatch(); + await Promise.resolve(); + expect(settlePendingFinalDelivery).toHaveBeenCalledOnce(); + resolveFirstCheck?.({ state: "unknown" }); + + await expect(firstDispatch).resolves.toBeUndefined(); + await expect(secondDispatch).rejects.toBeInstanceOf(PlatformMessageNotDispatchedError); + + expect(settlePendingFinalDelivery).toHaveBeenNthCalledWith( + 1, + { kind: "pending-final", ...completion }, + "unknown", + ["prepared", "queued"], + ); + expect(settlePendingFinalDelivery).toHaveBeenNthCalledWith( + 2, + { kind: "pending-final", ...completion }, + "unknown", + ["unknown"], + ); + }); + const run = (error: Error) => { const sourcePayload = setReplyPayloadMetadata( { text: "reply" },