From 726d348bbd3524bf64b872aa30c79576b59f7a68 Mon Sep 17 00:00:00 2001 From: ndholakia Date: Wed, 29 Jul 2026 03:40:52 -0500 Subject: [PATCH] fix(msteams): finalize sent replies after dispatch (#82354) Co-authored-by: Vincent Koc --- extensions/msteams/src/messenger.test.ts | 71 ++++++ extensions/msteams/src/messenger.ts | 79 ++++--- .../msteams/src/reply-dispatcher.test.ts | 128 +++++++++- extensions/msteams/src/reply-dispatcher.ts | 219 ++++++++++++++---- .../src/reply-stream-controller.sdk.test.ts | 11 +- .../src/reply-stream-controller.test.ts | 55 ++++- .../msteams/src/reply-stream-controller.ts | 106 +++++++-- 7 files changed, 552 insertions(+), 117 deletions(-) diff --git a/extensions/msteams/src/messenger.test.ts b/extensions/msteams/src/messenger.test.ts index 63ab4c2f1371..a40ed93d3d44 100644 --- a/extensions/msteams/src/messenger.test.ts +++ b/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 }> = []; diff --git a/extensions/msteams/src/messenger.ts b/extensions/msteams/src/messenger.ts index 6538ebefc179..120f51f28892 100644 --- a/extensions/msteams/src/messenger.ts +++ b/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, message: MSTeamsRenderedMessage, messageIndex: number, ): Promise => { 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 => { - 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, { diff --git a/extensions/msteams/src/reply-dispatcher.test.ts b/extensions/msteams/src/reply-dispatcher.test.ts index 7e4ae8ad0c17..dc6856eea814 100644 --- a/extensions/msteams/src/reply-dispatcher.test.ts +++ b/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; - deliver: (payload: { text: string }) => Promise | void; + deliver: (payload: { + text: string; + }) => ReturnType["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; + }); }); diff --git a/extensions/msteams/src/reply-dispatcher.ts b/extensions/msteams/src/reply-dispatcher.ts index e2f8cc366452..a639b288611c 100644 --- a/extensions/msteams/src/reply-dispatcher.ts +++ b/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>; + 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 => { 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(); + 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 => { - 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 => { + 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 diff --git a/extensions/msteams/src/reply-stream-controller.sdk.test.ts b/extensions/msteams/src/reply-stream-controller.sdk.test.ts index 5660536a37b2..ac7a0cc2b21d 100644 --- a/extensions/msteams/src/reply-stream-controller.sdk.test.ts +++ b/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); }); }); diff --git a/extensions/msteams/src/reply-stream-controller.test.ts b/extensions/msteams/src/reply-stream-controller.test.ts index 9326cdd0c9d2..d56284ce2ee7 100644 --- a/extensions/msteams/src/reply-stream-controller.test.ts +++ b/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", () => { diff --git a/extensions/msteams/src/reply-stream-controller.ts b/extensions/msteams/src/reply-stream-controller.ts index ebbfbf22a61b..66b93ef13ce5 100644 --- a/extensions/msteams/src/reply-stream-controller.ts +++ b/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> { + claimNativeDelivery(): boolean { + if (!nativeDispatchStarted || nativeDeliveryClaimed) { + return false; + } + nativeDeliveryClaimed = true; + return true; + }, + + async finalize(): Promise { // 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> = [ - { - type: "https://schema.org/Message", - "@type": "Message", - "@context": "https://schema.org", - "@id": "", - additionalType: ["AIGeneratedContent"], - }, - ]; - const finalChannelData: Record = 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> = [ + { + type: "https://schema.org/Message", + "@type": "Message", + "@context": "https://schema.org", + "@id": "", + additionalType: ["AIGeneratedContent"], + }, + ]; + const finalChannelData: Record = 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(); }