fix(msteams): finalize sent replies after dispatch (#82354)

Co-authored-by: Vincent Koc <vincentkoc@ieee.org>

extensions/msteams/src/messenger.test.ts

@@ -1,6 +1,7 @@
 // Msteams tests cover messenger plugin behavior.
 import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import path from "node:path";
+import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime";
 import { SILENT_REPLY_TOKEN } from "openclaw/plugin-sdk/reply-chunking";
 import type { PluginRuntime } from "openclaw/plugin-sdk/runtime-store";
 import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
@@ -381,6 +382,76 @@ describe("msteams messenger", () => {
       }
     });
 
+    it("marks local activity preparation failures as never dispatched", async () => {
+      const sendActivity = vi.fn(async () => ({ id: "should-not-send" }));
+      const missingPath = path.join(resolvePreferredOpenClawTmpDir(), "missing-msteams-file.txt");
+
+      await expect(
+        sendMSTeamsMessages({
+          replyStyle: "thread",
+          app: createMockApp(),
+          appId: "app123",
+          conversationRef: baseRef,
+          context: { sendActivity },
+          messages: [{ mediaUrl: missingPath }],
+        }),
+      ).rejects.toBeInstanceOf(PlatformMessageNotDispatchedError);
+      expect(sendActivity).not.toHaveBeenCalled();
+    });
+
+    it("does not claim no dispatch after an earlier batch message was sent", async () => {
+      const sendActivity = vi.fn(async () => ({ id: "sent-first" }));
+      const missingPath = path.join(resolvePreferredOpenClawTmpDir(), "missing-second-file.txt");
+
+      const error = await sendMSTeamsMessages({
+        replyStyle: "thread",
+        app: createMockApp(),
+        appId: "app123",
+        conversationRef: baseRef,
+        context: { sendActivity },
+        messages: [{ text: "first" }, { mediaUrl: missingPath }],
+      }).catch((cause: unknown) => cause);
+
+      expect(sendActivity).toHaveBeenCalledTimes(1);
+      expect(error).toBeInstanceOf(Error);
+      expect(error).not.toBeInstanceOf(PlatformMessageNotDispatchedError);
+    });
+
+    it("does not claim no dispatch when proactive fallback preparation fails after a send", async () => {
+      const threadSent: string[] = [];
+      const error = await sendMSTeamsMessages({
+        replyStyle: "thread",
+        app: createMockApp(),
+        appId: "app123",
+        conversationRef: {
+          ...baseRef,
+          user: undefined,
+        },
+        context: createRevokedThreadContext({ failAfterAttempt: 2, sent: threadSent }),
+        messages: [{ text: "first" }, { text: "second" }],
+      }).catch((cause: unknown) => cause);
+
+      expect(threadSent).toEqual(["first"]);
+      expect(error).toBeInstanceOf(Error);
+      expect(error).not.toBeInstanceOf(PlatformMessageNotDispatchedError);
+      expect((error as Error).message).toContain("missing user.id");
+    });
+
+    it("marks invalid proactive conversation references as never dispatched", async () => {
+      await expect(
+        sendMSTeamsMessages({
+          replyStyle: "top-level",
+          app: createMockApp(),
+          appId: "app123",
+          conversationRef: {
+            ...baseRef,
+            conversation: { id: "" },
+          },
+          messages: [{ text: "hello" }],
+        }),
+      ).rejects.toBeInstanceOf(PlatformMessageNotDispatchedError);
+    });
+
     it("retries thread sends on throttling (429)", async () => {
       const attempts: string[] = [];
       const retryEvents: Array<{ nextAttempt: number; delayMs: number }> = [];

extensions/msteams/src/messenger.ts

@@ -1,3 +1,4 @@
+import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime";
 // Msteams plugin module implements messenger behavior.
 import {
   isSilentReplyText,
@@ -459,39 +460,52 @@ export async function sendMSTeamsMessages(params: {
     throw new Error("unreachable Teams send retry loop exit");
   };
 
+  let providerDispatchStarted = false;
   const sendMessageInContext = async (
     sendFn: (activity: MSTeamsActivityLike) => Promise<unknown>,
     message: MSTeamsRenderedMessage,
     messageIndex: number,
   ): Promise<string> => {
     let pendingUploadId: string | undefined;
-    const response = await sendWithRetry(
-      async () => {
-        const activity = await buildActivity(
-          message,
-          params.conversationRef,
-          params.tokenProvider,
-          params.sharePointSiteId,
-          params.mediaMaxBytes,
-          { feedbackLoopEnabled: params.feedbackLoopEnabled },
+    let response: unknown;
+    try {
+      response = await sendWithRetry(
+        async () => {
+          const activity = await buildActivity(
+            message,
+            params.conversationRef,
+            params.tokenProvider,
+            params.sharePointSiteId,
+            params.mediaMaxBytes,
+            { feedbackLoopEnabled: params.feedbackLoopEnabled },
+          );
+
+          // Extract and strip the internal-only pending upload tag before sending.
+          pendingUploadId =
+            typeof activity["_pendingUploadId"] === "string"
+              ? activity["_pendingUploadId"]
+              : undefined;
+          if (pendingUploadId) {
+            delete activity["_pendingUploadId"];
+          }
+
+          providerDispatchStarted = true;
+          return await sendFn(activity);
+        },
+        {
+          messageIndex,
+          messageCount: messages.length,
+        },
+      );
+    } catch (error) {
+      if (!providerDispatchStarted) {
+        throw new PlatformMessageNotDispatchedError(
+          error instanceof Error ? error.message : "Teams activity preparation failed",
+          { cause: error },
         );
-
-        // Extract and strip the internal-only pending upload tag before sending.
-        pendingUploadId =
-          typeof activity["_pendingUploadId"] === "string"
-            ? activity["_pendingUploadId"]
-            : undefined;
-        if (pendingUploadId) {
-          delete activity["_pendingUploadId"];
-        }
-
-        return await sendFn(activity);
-      },
-      {
-        messageIndex,
-        messageCount: messages.length,
-      },
-    );
+      }
+      throw error;
+    }
     const messageId = extractMessageId(response) ?? "unknown";
 
     // Store the activity ID so the accept handler can replace the consent card in-place
@@ -519,7 +533,18 @@ export async function sendMSTeamsMessages(params: {
     startIndex: number,
     threadActivityId?: string,
   ): Promise<string[]> => {
-    const baseRef = buildConversationReference(params.conversationRef);
+    let baseRef: MSTeamsConversationReference;
+    try {
+      baseRef = buildConversationReference(params.conversationRef);
+    } catch (error) {
+      if (providerDispatchStarted) {
+        throw error;
+      }
+      throw new PlatformMessageNotDispatchedError(
+        error instanceof Error ? error.message : "Teams conversation preparation failed",
+        { cause: error },
+      );
+    }
     const isChannel = params.conversationRef.conversation?.conversationType === "channel";
     const sendFn = (activity: MSTeamsActivityLike) =>
       sendMSTeamsActivityWithReference(params.app, baseRef, activity, {

extensions/msteams/src/reply-dispatcher.test.ts

@@ -1,4 +1,5 @@
 // Msteams tests cover reply dispatcher plugin behavior.
+import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 const createChannelMessageReplyPipelineMock = vi.hoisted(() => vi.fn());
@@ -74,6 +75,8 @@ describe("createMSTeamsReplyDispatcher", () => {
 
   beforeEach(() => {
     vi.clearAllMocks();
+    sendMSTeamsMessagesMock.mockReset().mockResolvedValue([]);
+    renderReplyPayloadsToMessagesMock.mockReset().mockReturnValue([]);
     getGlobalHookRunnerMock.mockReturnValue(undefined);
     lastStreamMock = undefined;
 
@@ -166,7 +169,9 @@ describe("createMSTeamsReplyDispatcher", () => {
 
   type DispatcherOptions = {
     onReplyStart?: () => Promise<void> | void;
-    deliver: (payload: { text: string }) => Promise<void> | void;
+    deliver: (payload: {
+      text: string;
+    }) => ReturnType<ReturnType<typeof createMSTeamsReplyDispatcher>["delivery"]["deliver"]>;
   };
 
   type PipelineArgs = {
@@ -184,9 +189,7 @@ describe("createMSTeamsReplyDispatcher", () => {
     }
     return {
       onReplyStart: created.dispatcherOptions.onReplyStart,
-      deliver: async (payload) => {
-        await created.delivery.deliver(payload, { kind: "final" });
-      },
+      deliver: (payload) => created.delivery.deliver(payload, { kind: "final" }),
     };
   }
 
@@ -459,8 +462,13 @@ describe("createMSTeamsReplyDispatcher", () => {
     getStreamMock().close.mockResolvedValueOnce(undefined);
 
     dispatcher.replyOptions.onPartialReply?.({ text: "streamed" });
-    await options.deliver({ text: "streamed final" });
+    const result = await options.deliver({ text: "streamed final" });
     await dispatcher.dispatcherOptions.onSettled?.();
+    await expect(result?.finalization).resolves.toEqual({
+      visibleReplySent: true,
+      messageIds: ["fallback-id"],
+      content: "streamed final",
+    });
 
     expect(renderReplyPayloadsToMessagesMock).toHaveBeenCalledWith(
       [{ text: "streamed final" }],
@@ -672,7 +680,6 @@ describe("createMSTeamsReplyDispatcher", () => {
       { content: "two" },
     ] as never);
     sendMSTeamsMessagesMock
-      .mockRejectedValueOnce(Object.assign(new Error("gateway timeout"), { statusCode: 502 }))
       .mockResolvedValueOnce(["id-1"] as never)
       .mockRejectedValueOnce(Object.assign(new Error("gateway timeout"), { statusCode: 502 }));
 
@@ -683,8 +690,17 @@ describe("createMSTeamsReplyDispatcher", () => {
     );
     const options = dispatcherOptions();
 
-    await options.deliver({ text: "block content" });
+    const result = await options.deliver({ text: "block content" });
+    const finalization = expect(result?.finalization).rejects.toMatchObject({
+      code: "CHANNEL_PARTIAL_DELIVERY",
+      deliveryResult: {
+        visibleReplySent: true,
+        messageIds: ["id-1"],
+        content: "block content",
+      },
+    });
     await dispatcher.dispatcherOptions.onSettled?.();
+    await finalization;
 
     expect(onSentMessageIds).toHaveBeenCalledWith(["id-1"]);
     expect(enqueueSystemEventMock).toHaveBeenCalledTimes(1);
@@ -711,4 +727,102 @@ describe("createMSTeamsReplyDispatcher", () => {
 
     expect(enqueueSystemEventMock).not.toHaveBeenCalled();
   });
+
+  it("returns queued delivery identity only after the provider send runs", async () => {
+    renderReplyPayloadsToMessagesMock.mockReturnValue([{ text: "hello" }] as never);
+    sendMSTeamsMessagesMock.mockResolvedValue(["id-1"] as never);
+    const dispatcher = createDispatcher("groupchat", {
+      streaming: { block: { enabled: false } },
+    });
+
+    expect(dispatcher.delivery.observeMessageSent).toBe(true);
+    const result = await dispatcher.delivery.deliver({ text: "hello" }, { kind: "final" });
+    let settled = false;
+    void result?.finalization?.then(() => {
+      settled = true;
+    });
+    await Promise.resolve();
+    expect(settled).toBe(false);
+    expect(sendMSTeamsMessagesMock).not.toHaveBeenCalled();
+
+    await dispatcher.dispatcherOptions.onSettled?.();
+    await expect(result?.finalization).resolves.toEqual({
+      visibleReplySent: true,
+      messageIds: ["id-1"],
+      content: "hello",
+    });
+  });
+
+  it("keeps suppressed queued sends non-visible", async () => {
+    renderReplyPayloadsToMessagesMock.mockReturnValue([{ text: "hello" }] as never);
+    sendMSTeamsMessagesMock.mockResolvedValue([]);
+    const onSentMessageIds = vi.fn();
+    const dispatcher = createDispatcher(
+      "groupchat",
+      { streaming: { block: { enabled: false } } },
+      { onSentMessageIds },
+    );
+
+    const result = await dispatcher.delivery.deliver({ text: "hello" }, { kind: "final" });
+    await dispatcher.dispatcherOptions.onSettled?.();
+
+    await expect(result?.finalization).resolves.toEqual({
+      visibleReplySent: false,
+    });
+    expect(onSentMessageIds).not.toHaveBeenCalled();
+  });
+
+  it("returns native stream identity and final content after close", async () => {
+    const dispatcher = createDispatcher("personal");
+    dispatcher.replyOptions.onPartialReply?.({ text: "streamed" });
+
+    const result = await dispatcher.delivery.deliver({ text: "streamed final" }, { kind: "final" });
+    expect(getStreamMock().close).not.toHaveBeenCalled();
+
+    await dispatcher.dispatcherOptions.onSettled?.();
+    await expect(result?.finalization).resolves.toEqual({
+      visibleReplySent: true,
+      messageIds: ["stream-final"],
+      content: "streamed final",
+    });
+  });
+
+  it("settles delivery when sent-message ID observation throws", async () => {
+    renderReplyPayloadsToMessagesMock.mockReturnValue([{ text: "hello" }] as never);
+    sendMSTeamsMessagesMock.mockResolvedValue(["id-1"] as never);
+    const dispatcher = createDispatcher(
+      "groupchat",
+      { streaming: { block: { enabled: false } } },
+      {
+        onSentMessageIds: () => {
+          throw new Error("observer failed");
+        },
+      },
+    );
+
+    const result = await dispatcher.delivery.deliver({ text: "hello" }, { kind: "final" });
+    await dispatcher.dispatcherOptions.onSettled?.();
+
+    await expect(result?.finalization).resolves.toEqual({
+      visibleReplySent: true,
+      messageIds: ["id-1"],
+      content: "hello",
+    });
+  });
+
+  it("preserves a never-dispatched queued failure for core event suppression", async () => {
+    const failure = new PlatformMessageNotDispatchedError("local media load failed", {
+      cause: new Error("missing file"),
+    });
+    renderReplyPayloadsToMessagesMock.mockReturnValue([{ mediaUrl: "/missing/file" }] as never);
+    sendMSTeamsMessagesMock.mockRejectedValue(failure);
+    const dispatcher = createDispatcher("groupchat", {
+      streaming: { block: { enabled: false } },
+    });
+
+    const result = await dispatcher.delivery.deliver({ text: "attachment" }, { kind: "final" });
+    const finalization = expect(result?.finalization).rejects.toBe(failure);
+    await dispatcher.dispatcherOptions.onSettled?.();
+    await finalization;
+  });
 });

extensions/msteams/src/reply-dispatcher.ts

@@ -1,5 +1,8 @@
 import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime";
-import type { ChannelInboundTurnPlan } from "openclaw/plugin-sdk/channel-inbound";
+import {
+  createChannelPartialDeliveryError,
+  type ChannelInboundTurnPlan,
+} from "openclaw/plugin-sdk/channel-inbound";
 // Msteams plugin module implements reply dispatcher behavior.
 import {
   buildChannelProgressDraftLine,
@@ -10,6 +13,8 @@ import {
   resolveChannelStreamingPreviewToolProgress,
   resolveChannelStreamingSuppressDefaultToolProgressMessages,
 } from "openclaw/plugin-sdk/channel-outbound";
+import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime";
+import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
 import { getGlobalHookRunner } from "openclaw/plugin-sdk/plugin-runtime";
 import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime";
 import {
@@ -200,7 +205,27 @@ export function createMSTeamsReplyDispatcher(params: {
   const typingIndicatorEnabled =
     typeof msteamsCfg?.typingIndicator === "boolean" ? msteamsCfg.typingIndicator : true;
 
-  const pendingMessages: MSTeamsRenderedMessage[] = [];
+  type DeliveryOutcome = {
+    messageIds?: string[];
+    visibleReplySent: boolean;
+    content?: string;
+  };
+
+  type PendingDelivery = {
+    messages: MSTeamsRenderedMessage[];
+    finalization: ReturnType<typeof createDeferred<DeliveryOutcome>>;
+    content?: string;
+    native: boolean;
+    nativeSettled: boolean;
+    blockSettled: boolean;
+    settled: boolean;
+    visibleReplySent: boolean;
+    nativeMessageId?: string;
+    messageIds: string[];
+    errors: unknown[];
+  };
+
+  const pendingDeliveries: PendingDelivery[] = [];
 
   const sendMessages = async (messages: MSTeamsRenderedMessage[]): Promise<string[]> => {
     return sendMSTeamsMessages({
@@ -251,37 +276,102 @@ export function createMSTeamsReplyDispatcher(params: {
     });
   };
 
-  const queueReplyPayload = (payload: ReplyPayload) => {
-    const messages = renderReplyPayloadsToMessages([payload], {
+  const renderReplyPayload = (payload: ReplyPayload) => {
+    return renderReplyPayloadsToMessages([payload], {
       textChunkLimit: params.textLimit,
       chunkText: true,
       mediaMode: "split",
       tableMode,
       chunkMode,
     });
-    pendingMessages.push(...messages);
+  };
+
+  const deliveryOutcome = (delivery: PendingDelivery): DeliveryOutcome => {
+    const messageIds = [
+      ...(delivery.nativeMessageId ? [delivery.nativeMessageId] : []),
+      ...delivery.messageIds,
+    ];
+    return {
+      visibleReplySent: delivery.visibleReplySent,
+      ...(messageIds.length > 0 ? { messageIds } : {}),
+      ...(delivery.visibleReplySent && delivery.content !== undefined
+        ? { content: delivery.content }
+        : {}),
+    };
+  };
+
+  const settlePendingDelivery = (delivery: PendingDelivery) => {
+    if (
+      delivery.settled ||
+      !delivery.blockSettled ||
+      (delivery.native && !delivery.nativeSettled)
+    ) {
+      return;
+    }
+    delivery.settled = true;
+    const outcome = deliveryOutcome(delivery);
+    if (delivery.errors.length === 0) {
+      delivery.finalization.resolve(outcome);
+      return;
+    }
+
+    const error =
+      delivery.errors.find(
+        (candidate) => !(candidate instanceof PlatformMessageNotDispatchedError),
+      ) ?? delivery.errors[0];
+    delivery.finalization.reject(
+      delivery.visibleReplySent
+        ? createChannelPartialDeliveryError(error, {
+            ...outcome,
+            visibleReplySent: true,
+          })
+        : error,
+    );
+  };
+
+  const queueReplyPayload = (
+    payload: ReplyPayload,
+    messages: MSTeamsRenderedMessage[],
+    native: boolean,
+  ): PendingDelivery => {
+    const finalization = createDeferred<DeliveryOutcome>();
+    const delivery: PendingDelivery = {
+      messages,
+      finalization,
+      content: payload.text,
+      native,
+      nativeSettled: !native,
+      blockSettled: messages.length === 0,
+      settled: false,
+      visibleReplySent: false,
+      messageIds: [],
+      errors: [],
+    };
+    pendingDeliveries.push(delivery);
+    return delivery;
   };
 
   const flushPendingMessages = async () => {
-    if (pendingMessages.length === 0) {
-      return;
-    }
-    const toSend = pendingMessages.splice(0);
-    const total = toSend.length;
-    let ids: string[];
-    try {
-      ids = await sendMessages(toSend);
-    } catch (batchError) {
-      ids = [];
+    for (const delivery of pendingDeliveries) {
+      if (delivery.blockSettled) {
+        continue;
+      }
+      const toSend = delivery.messages.splice(0);
+      const total = toSend.length;
       let failed = 0;
-      let lastFailedError: unknown = batchError;
+      let lastFailedError: unknown;
+      const sentIds: string[] = [];
       for (const msg of toSend) {
         try {
           const msgIds = await sendMessages([msg]);
-          ids.push(...msgIds);
+          delivery.visibleReplySent ||= msgIds.length > 0;
+          const validIds = msgIds.filter((id) => id.trim() && id !== "unknown");
+          delivery.messageIds.push(...validIds);
+          sentIds.push(...validIds);
         } catch (msgError) {
           failed += 1;
           lastFailedError = msgError;
+          delivery.errors.push(msgError);
           params.log.debug?.("individual message send failed, continuing with remaining blocks");
         }
       }
@@ -296,9 +386,17 @@ export function createMSTeamsReplyDispatcher(params: {
           error: lastFailedError,
         });
       }
-    }
-    if (ids.length > 0) {
-      params.onSentMessageIds?.(ids);
+      delivery.blockSettled = true;
+      settlePendingDelivery(delivery);
+      if (sentIds.length > 0) {
+        try {
+          params.onSentMessageIds?.(sentIds);
+        } catch (error) {
+          params.log.warn?.("failed to record sent Teams message ids", {
+            error: formatUnknownError(error),
+          });
+        }
+      }
     }
   };
 
@@ -321,19 +419,30 @@ export function createMSTeamsReplyDispatcher(params: {
     typingCallbacks,
   };
   const delivery: ChannelInboundTurnPlan["delivery"] = {
+    observeMessageSent: true,
     deliver: async (payload) => {
       const preparedPayload = streamController.preparePayload(payload);
-      if (!preparedPayload) {
-        return;
+      const native = streamController.claimNativeDelivery();
+      const messages = preparedPayload ? renderReplyPayload(preparedPayload) : [];
+      if (!native && messages.length === 0) {
+        return {
+          visibleReplySent: false,
+          suppression: { reason: "no_visible_result" },
+        };
       }
 
-      queueReplyPayload(preparedPayload);
+      const pending = queueReplyPayload(payload, messages, native);
 
       // When block streaming is enabled, flush immediately so blocks are
       // delivered progressively instead of batching until markDispatchIdle.
       if (blockStreamingEnabled) {
         await flushPendingMessages();
       }
+      settlePendingDelivery(pending);
+      return {
+        visibleReplySent: false,
+        finalization: pending.finalization.promise,
+      };
     },
     onError: (err, info) => {
       const errMsg = formatUnknownError(err);
@@ -351,29 +460,49 @@ export function createMSTeamsReplyDispatcher(params: {
     },
   };
 
-  const settleDelivery = (): Promise<void> => {
-    return flushPendingMessages()
-      .catch((err: unknown) => {
-        const errMsg = formatUnknownError(err);
-        const classification = classifyMSTeamsSendError(err);
-        const hint = formatMSTeamsSendErrorHint(classification);
-        params.runtime.error?.(`msteams flush reply failed: ${errMsg}${hint ? ` (${hint})` : ""}`);
-        params.log.error("flush reply failed", {
-          error: errMsg,
-          classification,
-          hint,
+  const settleDelivery = async (): Promise<void> => {
+    await flushPendingMessages();
+
+    const nativeDelivery = pendingDeliveries.find(
+      (candidate) => candidate.native && !candidate.nativeSettled,
+    );
+    if (!nativeDelivery) {
+      await streamController.finalize();
+      return;
+    }
+    let nativeResult;
+    try {
+      nativeResult = await streamController.finalize();
+    } catch (error) {
+      nativeDelivery.errors.push(error);
+      nativeDelivery.nativeSettled = true;
+      settlePendingDelivery(nativeDelivery);
+      return;
+    }
+
+    nativeDelivery.visibleReplySent ||= nativeResult.visibleReplySent;
+    nativeDelivery.nativeMessageId = nativeResult.messageId;
+    if (nativeResult.content !== undefined) {
+      nativeDelivery.content = nativeResult.content;
+    }
+    if (nativeResult.fallbackPayload) {
+      nativeDelivery.messages.push(...renderReplyPayload(nativeResult.fallbackPayload));
+      nativeDelivery.blockSettled = nativeDelivery.messages.length === 0;
+    }
+    nativeDelivery.nativeSettled = true;
+    if (!nativeDelivery.blockSettled) {
+      await flushPendingMessages();
+    }
+    settlePendingDelivery(nativeDelivery);
+    if (nativeResult.messageId) {
+      try {
+        params.onSentMessageIds?.([nativeResult.messageId]);
+      } catch (error) {
+        params.log.warn?.("failed to record sent Teams message id", {
+          error: formatUnknownError(error),
         });
-      })
-      .then(async () => {
-        const fallbackPayload = await streamController.finalize().catch((err: unknown) => {
-          params.log.debug?.("stream finalize failed", { error: formatUnknownError(err) });
-          return undefined;
-        });
-        if (fallbackPayload) {
-          queueReplyPayload(fallbackPayload);
-          await flushPendingMessages();
-        }
-      });
+      }
+    }
   };
 
   // Pipe agent tool/plan/approval/command events into the stream controller's

extensions/msteams/src/reply-stream-controller.sdk.test.ts

@@ -176,7 +176,11 @@ describe("Microsoft Teams SDK acknowledged stream fallback", () => {
     expect(acknowledgements).toEqual([{ id: "stream-size-limit", text: acknowledgedPrefix }]);
     expect(controller.preparePayload({ text: completeReply })).toBeUndefined();
 
-    await expect(controller.finalize()).resolves.toEqual({ text: "b".repeat(200) });
+    await expect(controller.finalize()).resolves.toEqual({
+      visibleReplySent: true,
+      content: completeReply,
+      fallbackPayload: { text: "b".repeat(200) },
+    });
     // Finalization queues its own metadata activity; this is a second
     // provider operation, not a retry of the rejected streaming chunk.
     expect(
@@ -214,7 +218,10 @@ describe("Microsoft Teams SDK acknowledged stream fallback", () => {
     });
 
     expect(controller.preparePayload({ text: completeReply })).toBeUndefined();
-    await expect(controller.finalize()).resolves.toBeUndefined();
+    await expect(controller.finalize()).resolves.toEqual({
+      visibleReplySent: true,
+      content: acknowledgedPrefix,
+    });
     expect(requests.filter((request) => request.scenario === "cancel")).toHaveLength(2);
   });
 });

extensions/msteams/src/reply-stream-controller.test.ts

@@ -269,7 +269,11 @@ describe("createTeamsReplyStreamController", () => {
     const ctrl = makeController({ stream });
     ctrl.onPartialReply({ text: "streamed" });
     expect(ctrl.preparePayload({ text: "streamed" })).toBeUndefined();
-    await expect(ctrl.finalize()).resolves.toBeUndefined();
+    await expect(ctrl.finalize()).resolves.toEqual({
+      visibleReplySent: true,
+      messageId: "stream-final",
+      content: "streamed",
+    });
     expect(stream.close).toHaveBeenCalled();
   });
 
@@ -281,7 +285,11 @@ describe("createTeamsReplyStreamController", () => {
     ctrl.onPartialReply({ text: "streamed" });
     expect(ctrl.preparePayload({ text: "streamed final" })).toBeUndefined();
 
-    await expect(ctrl.finalize()).resolves.toEqual({ text: "streamed final" });
+    await expect(ctrl.finalize()).resolves.toEqual({
+      visibleReplySent: true,
+      content: "streamed final",
+      fallbackPayload: { text: "streamed final" },
+    });
   });
 
   it("returns text-only fallback when stream close no-ops after media already queued", async () => {
@@ -296,9 +304,13 @@ describe("createTeamsReplyStreamController", () => {
     });
 
     await expect(ctrl.finalize()).resolves.toEqual({
-      text: "streamed final",
-      mediaUrl: undefined,
-      mediaUrls: undefined,
+      visibleReplySent: true,
+      content: "streamed final",
+      fallbackPayload: {
+        text: "streamed final",
+        mediaUrl: undefined,
+        mediaUrls: undefined,
+      },
     });
   });
 
@@ -310,7 +322,11 @@ describe("createTeamsReplyStreamController", () => {
     ctrl.onPartialReply({ text: "streamed" });
     expect(ctrl.preparePayload({ text: "streamed final" })).toBeUndefined();
 
-    await expect(ctrl.finalize()).resolves.toEqual({ text: "streamed final" });
+    await expect(ctrl.finalize()).resolves.toEqual({
+      visibleReplySent: true,
+      content: "streamed final",
+      fallbackPayload: { text: "streamed final" },
+    });
   });
 
   it("does not close the stream in finalize when no tokens were emitted", async () => {
@@ -512,7 +528,10 @@ describe("createTeamsReplyStreamController", () => {
       });
       // Must not throw — finalize's pre-check on stream.canceled may miss
       // the cancellation that happens between check and emit.
-      await expect(ctrl.finalize()).resolves.toBeUndefined();
+      await expect(ctrl.finalize()).resolves.toEqual({
+        visibleReplySent: true,
+        content: "partial",
+      });
     });
 
     it("latches streamFailed (and does not throw) on non-cancel errors from stream.emit", () => {
@@ -660,7 +679,11 @@ describe("createTeamsReplyStreamController", () => {
       expect(ctrl.preparePayload({ text: "hello world" })).toBeUndefined();
       stream.close.mockRejectedValueOnce(new Error("close failed"));
 
-      await expect(ctrl.finalize()).resolves.toEqual({ text: " world" });
+      await expect(ctrl.finalize()).resolves.toEqual({
+        visibleReplySent: true,
+        content: "hello world",
+        fallbackPayload: { text: " world" },
+      });
       expect(stream.events.off).toHaveBeenCalledWith(0);
     });
 
@@ -673,7 +696,10 @@ describe("createTeamsReplyStreamController", () => {
       expect(ctrl.preparePayload({ text: "hello" })).toBeUndefined();
       stream.close.mockResolvedValueOnce(undefined);
 
-      await expect(ctrl.finalize()).resolves.toBeUndefined();
+      await expect(ctrl.finalize()).resolves.toEqual({
+        visibleReplySent: true,
+        content: "hello",
+      });
       expect(stream.events.off).toHaveBeenCalledWith(0);
     });
 
@@ -686,7 +712,10 @@ describe("createTeamsReplyStreamController", () => {
       stream.canceled = true;
 
       expect(ctrl.preparePayload({ text: "hello world" })).toBeUndefined();
-      await expect(ctrl.finalize()).resolves.toBeUndefined();
+      await expect(ctrl.finalize()).resolves.toEqual({
+        visibleReplySent: true,
+        content: "hello",
+      });
       expect(stream.events.off).toHaveBeenCalledWith(0);
     });
 
@@ -709,7 +738,11 @@ describe("createTeamsReplyStreamController", () => {
       });
       // Finalize must not propagate; it returns the retained payload so the
       // dispatcher can fall back to normal Teams delivery.
-      await expect(ctrl.finalize()).resolves.toEqual({ text: "partial final" });
+      await expect(ctrl.finalize()).resolves.toEqual({
+        visibleReplySent: true,
+        content: "partial final",
+        fallbackPayload: { text: "partial final" },
+      });
     });
 
     it("treats post-cancel stream as inactive without further emit attempts", () => {

extensions/msteams/src/reply-stream-controller.ts

@@ -13,6 +13,7 @@ import {
 } from "openclaw/plugin-sdk/channel-outbound";
 import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime";
 import type { MSTeamsConfig, ReplyPayload } from "../runtime-api.js";
+import { extractMessageId } from "./media-helpers.js";
 import type { MSTeamsMonitorLogger } from "./monitor-types.js";
 import type { MSTeamsTurnContext } from "./sdk-types.js";
 
@@ -30,6 +31,13 @@ type TeamsStreamChunkEvents = {
   off(subscriptionId: number): void;
 };
 
+type MSTeamsNativeDeliveryFinalization = {
+  visibleReplySent: boolean;
+  content?: string;
+  messageId?: string;
+  fallbackPayload?: ReplyPayload;
+};
+
 // The SDK throws StreamCancelledError synchronously from stream.emit/update
 // when the user pressed Stop in Teams (Teams replies 403 to the next chunk
 // update and the SDK flips _canceled). Match by `name` rather than importing
@@ -81,6 +89,8 @@ export function createTeamsReplyStreamController(params: {
   const stream = shouldUseNativeStream ? params.context.stream : undefined;
 
   let tokensEmitted = false;
+  let nativeDispatchStarted = false;
+  let nativeDeliveryClaimed = false;
   let streamFinalizationPending = false;
   let canceledLocally = false;
   // Set when `stream.emit/close` fails for a non-cancel reason after we've
@@ -258,6 +268,7 @@ export function createTeamsReplyStreamController(params: {
         stream.emit(delta);
         emittedText = fullText;
         tokensEmitted = true;
+        nativeDispatchStarted = true;
       } catch (err) {
         if (isStreamCancelledError(err)) {
           canceledLocally = true;
@@ -390,6 +401,7 @@ export function createTeamsReplyStreamController(params: {
       if (streamMode === "progress" && payload.text) {
         try {
           stream.emit(payload.text);
+          nativeDispatchStarted = true;
           pendingFinalPayload = fallbackPayloadForSuppressedFinal(payload);
           streamFinalizationPending = true;
           const hasMedia = Boolean(payload.mediaUrl || payload.mediaUrls?.length);
@@ -409,31 +421,56 @@ export function createTeamsReplyStreamController(params: {
       return payload;
     },
 
-    async finalize(): Promise<Maybe<ReplyPayload>> {
+    claimNativeDelivery(): boolean {
+      if (!nativeDispatchStarted || nativeDeliveryClaimed) {
+        return false;
+      }
+      nativeDeliveryClaimed = true;
+      return true;
+    },
+
+    async finalize(): Promise<MSTeamsNativeDeliveryFinalization> {
       // The delay gate may still hold a pending start timer for fast turns;
       // stop it before closing so it cannot fire against the closed stream.
       progressDraftGate.cancel();
-      if (!stream || !streamFinalizationPending || wasCanceled()) {
+      if (!stream || !nativeDispatchStarted) {
         releaseStreamChunkSubscription();
-        return undefined;
+        return { visibleReplySent: false };
       }
-      // Emit a final MessageActivity carrying the AI-generated marker and (if
-      // enabled) the feedback channelData. The SDK's HttpStream merges this
-      // into the closing activity it sends to Teams, so streamed replies still
-      // get the AI-generated label and thumbs up/down.
-      const finalEntities: Array<Record<string, unknown>> = [
-        {
-          type: "https://schema.org/Message",
-          "@type": "Message",
-          "@context": "https://schema.org",
-          "@id": "",
-          additionalType: ["AIGeneratedContent"],
-        },
-      ];
-      const finalChannelData: Record<string, unknown> = params.feedbackLoopEnabled
-        ? { feedbackLoopEnabled: true }
-        : {};
+      const content = wasCanceled()
+        ? acknowledgedText || undefined
+        : (pendingFinalPayload?.text ?? (emittedText || undefined));
       try {
+        if (wasCanceled()) {
+          pendingFinalPayload = undefined;
+          streamFinalizationPending = false;
+          return {
+            visibleReplySent: content !== undefined,
+            ...(content === undefined ? {} : { content }),
+          };
+        }
+        if (!streamFinalizationPending) {
+          return {
+            visibleReplySent: true,
+            ...(content === undefined ? {} : { content }),
+          };
+        }
+        // Emit a final MessageActivity carrying the AI-generated marker and (if
+        // enabled) the feedback channelData. The SDK's HttpStream merges this
+        // into the closing activity it sends to Teams, so streamed replies still
+        // get the AI-generated label and thumbs up/down.
+        const finalEntities: Array<Record<string, unknown>> = [
+          {
+            type: "https://schema.org/Message",
+            "@type": "Message",
+            "@context": "https://schema.org",
+            "@id": "",
+            additionalType: ["AIGeneratedContent"],
+          },
+        ];
+        const finalChannelData: Record<string, unknown> = params.feedbackLoopEnabled
+          ? { feedbackLoopEnabled: true }
+          : {};
         stream.emit({
           type: "message",
           entities: finalEntities,
@@ -444,18 +481,30 @@ export function createTeamsReplyStreamController(params: {
         if (!result) {
           const fallback = pendingFinalPayload;
           pendingFinalPayload = undefined;
-          return fallback && !wasCanceled()
-            ? fallbackPayloadAfterAcknowledgedText(fallback)
-            : undefined;
+          const fallbackPayload =
+            fallback && !wasCanceled() ? fallbackPayloadAfterAcknowledgedText(fallback) : undefined;
+          return {
+            visibleReplySent: true,
+            ...(content === undefined ? {} : { content }),
+            ...(fallbackPayload ? { fallbackPayload } : {}),
+          };
         }
         pendingFinalPayload = undefined;
-        return undefined;
+        const messageId = extractMessageId(result) ?? undefined;
+        return {
+          visibleReplySent: true,
+          ...(content === undefined ? {} : { content }),
+          ...(messageId ? { messageId } : {}),
+        };
       } catch (err) {
         if (isStreamCancelledError(err)) {
           canceledLocally = true;
           pendingFinalPayload = undefined;
           streamFinalizationPending = false;
-          return undefined;
+          return {
+            visibleReplySent: true,
+            ...(content === undefined ? {} : { content }),
+          };
         }
         // Non-cancel failure during the closing emit/close. The streamed
         // prefix is already visible to the user; the only loss is the
@@ -470,7 +519,14 @@ export function createTeamsReplyStreamController(params: {
         );
         const fallback = pendingFinalPayload;
         pendingFinalPayload = undefined;
-        return fallback ? fallbackPayloadAfterAcknowledgedText(fallback) : undefined;
+        const fallbackPayload = fallback
+          ? fallbackPayloadAfterAcknowledgedText(fallback)
+          : undefined;
+        return {
+          visibleReplySent: true,
+          ...(content === undefined ? {} : { content }),
+          ...(fallbackPayload ? { fallbackPayload } : {}),
+        };
       } finally {
         releaseStreamChunkSubscription();
       }