fix(discord): revalidate delivery authority per send (#128357)

* fix(discord): fence media caption follow-ups

* fix(discord): revalidate delivery custody per post

* test(discord): prove delivery fence over HTTP

Amp-Thread-ID: https://ampcode.com/threads/T-01a021f4-b547-7788-a916-d4a94cbd3e3b

* fix(discord): fence each retry attempt

Amp-Thread-ID: https://ampcode.com/threads/T-01a021f4-b547-7788-a916-d4a94cbd3e3b

* fix(discord): fence remaining retry paths

Amp-Thread-ID: https://ampcode.com/threads/T-01a021f4-b547-7788-a916-d4a94cbd3e3b

---------

Co-authored-by: Amp <amp@ampcode.com>

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();

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),
     });
   });
 

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?: <T extends ReplyPayload>(payload: T) => T;
+  onPlatformSendDispatch: () => Promise<void>;
+};
+
 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?: <T extends ReplyPayload>(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;
     }

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");

extensions/discord/src/monitor/reply-delivery.ts

@@ -230,6 +230,7 @@ export async function deliverDiscordReply(params: {
   allowedMentions?: DiscordAllowedMentions;
   kind: "tool" | "block" | "final";
   bindPendingFinalDelivery?: <T extends ReplyPayload>(payload: T) => T;
+  onPlatformSendDispatch?: () => Promise<void>;
 }) {
   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,

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({

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 };

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 };

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 });

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) {

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) ??

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({

extensions/discord/src/voice-message.ts

@@ -405,6 +405,7 @@ export async function sendDiscordVoiceMessage(
   request: DiscordRetryRunner,
   silent?: boolean,
   token?: string,
+  onPlatformSendDispatch?: () => Promise<void>,
 ): 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,

src/channels/turn/direct-delivery-custody.ts

@@ -28,23 +28,29 @@ export function createDirectPendingFinalCustody(
     return undefined;
   }
   const { kind: _kind, ...identity } = completion;
-  let admission: Promise<void> | 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;
     },
   };

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" },