From a1b0839a5cdf98f476aa6f2fe2a402408370e27a Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 16:45:55 +0800 Subject: [PATCH 1/8] refactor(msteams): extract inbound activity facts --- .../src/monitor-handler/inbound-facts.test.ts | 75 ++++++ .../src/monitor-handler/inbound-facts.ts | 178 +++++++++++++++ .../src/monitor-handler/message-handler.ts | 213 ++++-------------- 3 files changed, 292 insertions(+), 174 deletions(-) create mode 100644 extensions/msteams/src/monitor-handler/inbound-facts.test.ts create mode 100644 extensions/msteams/src/monitor-handler/inbound-facts.ts diff --git a/extensions/msteams/src/monitor-handler/inbound-facts.test.ts b/extensions/msteams/src/monitor-handler/inbound-facts.test.ts new file mode 100644 index 000000000000..e639cbf9c7fb --- /dev/null +++ b/extensions/msteams/src/monitor-handler/inbound-facts.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; +import type { MSTeamsTurnContext } from "../sdk-types.js"; +import { assembleMSTeamsInboundFacts, prepareMSTeamsDebounceEntry } from "./inbound-facts.js"; + +function context(activity: MSTeamsTurnContext["activity"]): MSTeamsTurnContext { + return { activity } as MSTeamsTurnContext; +} + +describe("msteams inbound facts", () => { + it("prefers activity text, then HTML, then adaptive-card values", async () => { + const textEntry = await prepareMSTeamsDebounceEntry({ + context: context({ + type: "message", + text: " hello ", + attachments: [{ contentType: "text/html", content: "

ignored

" }], + value: { action: "ignored" }, + }), + }); + const htmlEntry = await prepareMSTeamsDebounceEntry({ + context: context({ + type: "message", + attachments: [{ contentType: "text/html", content: "

hello & goodbye

" }], + }), + }); + const valueEntry = await prepareMSTeamsDebounceEntry({ + context: context({ + type: "message", + value: { action: "approve", id: 7 }, + }), + }); + + expect(textEntry.text).toBe("hello"); + expect(htmlEntry.text).toBe("hello & goodbye"); + expect(valueEntry.text).toContain("approve"); + expect(valueEntry.text).toContain("7"); + }); + + it("preserves raw Bot Framework IDs while normalizing routing facts", async () => { + const entry = await prepareMSTeamsDebounceEntry({ + context: context({ + type: "message", + id: "message-1", + text: "hello", + from: { id: "user-1", aadObjectId: "aad-1", name: "Alice" }, + recipient: { id: "bot-1", name: "Bot" }, + conversation: { + id: "19:channel@thread.tacv2;messageid=thread-root", + conversationType: "channel", + }, + channelData: { + team: { id: "team-1" }, + channel: { id: "channel-1" }, + tenant: { id: "tenant-1" }, + }, + serviceUrl: "https://smba.trafficmanager.net/amer/", + }), + }); + + const facts = assembleMSTeamsInboundFacts({ entry, mediaMaxBytes: 1024 }); + + expect(facts.rawConversationId).toBe("19:channel@thread.tacv2;messageid=thread-root"); + expect(facts.conversationId).toBe("19:channel@thread.tacv2"); + expect(facts.conversationMessageId).toBe("thread-root"); + expect(facts.threadId).toBe("thread-root"); + expect(facts.conversationRef).toMatchObject({ + tenantId: "tenant-1", + teamId: "team-1", + threadId: "thread-root", + conversation: { + id: "19:channel@thread.tacv2", + tenantId: "tenant-1", + }, + }); + }); +}); diff --git a/extensions/msteams/src/monitor-handler/inbound-facts.ts b/extensions/msteams/src/monitor-handler/inbound-facts.ts new file mode 100644 index 000000000000..1bc82d52df45 --- /dev/null +++ b/extensions/msteams/src/monitor-handler/inbound-facts.ts @@ -0,0 +1,178 @@ +// Msteams plugin module assembles stable inbound activity facts. +import { serializeMSTeamsAdaptiveCardActionValue } from "../adaptive-card-submit.js"; +import { + resolveMSTeamsAdvertisedMedia, + summarizeMSTeamsHtmlAttachments, + type MSTeamsAttachmentLike, +} from "../attachments.js"; +import { extractHtmlFromAttachment } from "../attachments/shared.js"; +import { tryNormalizeBotFrameworkServiceUrl } from "../bot-framework-service-url.js"; +import type { StoredConversationReference } from "../conversation-store.js"; +import { + extractMSTeamsConversationMessageId, + extractMSTeamsQuoteInfo, + normalizeMSTeamsConversationId, + stripMSTeamsMentionTags, + wasMSTeamsBotMentioned, +} from "../inbound.js"; +import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js"; +import type { MSTeamsTurnContext } from "../sdk-types.js"; +import { wasMSTeamsMessageSentWithPersistence } from "../sent-message-cache.js"; + +function extractTextFromHtmlAttachments(attachments: MSTeamsAttachmentLike[]): string { + for (const attachment of attachments) { + const raw = extractHtmlFromAttachment(attachment); + if (!raw) { + continue; + } + const text = raw + .replace(/]*>.*?<\/at>/gis, " ") + .replace(/]*href=["']([^"']+)["'][^>]*>(.*?)<\/a>/gis, "$2 $1") + .replace(//gi, "\n") + .replace(/<\/p>/gi, "\n") + .replace(/<[^>]+>/g, " ") + .replace(/ /gi, " ") + .replace(/&/gi, "&") + .replace(/\s+/g, " ") + .trim(); + if (text) { + return text; + } + } + return ""; +} + +export type MSTeamsDebounceEntry = { + context: MSTeamsTurnContext; + rawText: string; + text: string; + attachments: MSTeamsAttachmentLike[]; + wasMentioned: boolean; + implicitMentionKinds: Array<"reply_to_bot">; + turnAdoptionLifecycle?: MSTeamsIngressLifecycle; +}; + +export async function prepareMSTeamsDebounceEntry(params: { + context: MSTeamsTurnContext; + turnAdoptionLifecycle?: MSTeamsIngressLifecycle; +}): Promise { + const activity = params.context.activity; + const attachments = Array.isArray(activity.attachments) + ? (activity.attachments as unknown as MSTeamsAttachmentLike[]) + : []; + const rawText = activity.text?.trim() ?? ""; + const htmlText = extractTextFromHtmlAttachments(attachments); + const valueText = + rawText || htmlText ? "" : serializeMSTeamsAdaptiveCardActionValue(activity.value); + const text = stripMSTeamsMentionTags(rawText || htmlText || valueText || ""); + const conversationId = normalizeMSTeamsConversationId(activity.conversation?.id ?? ""); + const replyToId = activity.replyToId ?? undefined; + const implicitMentionKinds: Array<"reply_to_bot"> = + conversationId && + replyToId && + (await wasMSTeamsMessageSentWithPersistence({ conversationId, messageId: replyToId })) + ? ["reply_to_bot"] + : []; + + return { + context: params.context, + rawText, + text, + attachments, + wasMentioned: wasMSTeamsBotMentioned(activity), + implicitMentionKinds, + turnAdoptionLifecycle: params.turnAdoptionLifecycle, + }; +} + +function buildStoredConversationReference(params: { + activity: MSTeamsTurnContext["activity"]; + conversationId: string; + conversationType: string; + teamId?: string; + threadId?: string; +}): StoredConversationReference { + const { activity, conversationId, conversationType, teamId, threadId } = params; + const from = activity.from; + const conversation = activity.conversation; + const clientInfo = activity.entities?.find((entity) => entity.type === "clientInfo") as + | { timezone?: string } + | undefined; + // Proactive sends require the tenant and normalized regional service URL + // captured from the original Bot Framework activity. + const tenantId = activity.channelData?.tenant?.id ?? conversation?.tenantId; + const aadObjectId = from?.aadObjectId; + const serviceUrl = tryNormalizeBotFrameworkServiceUrl(activity.serviceUrl); + return { + activityId: activity.id, + user: from ? { id: from.id, name: from.name, aadObjectId: from.aadObjectId } : undefined, + agent: activity.recipient, + conversation: { + id: conversationId, + conversationType, + tenantId, + }, + ...(tenantId ? { tenantId } : {}), + ...(aadObjectId ? { aadObjectId } : {}), + teamId, + channelId: activity.channelId, + ...(serviceUrl ? { serviceUrl } : {}), + locale: activity.locale, + ...(clientInfo?.timezone ? { timezone: clientInfo.timezone } : {}), + ...(threadId ? { threadId } : {}), + }; +} + +export function assembleMSTeamsInboundFacts(params: { + entry: MSTeamsDebounceEntry; + mediaMaxBytes: number; +}) { + const { entry, mediaMaxBytes } = params; + const activity = entry.context.activity; + const conversation = activity.conversation; + const rawConversationId = conversation?.id ?? ""; + const conversationId = normalizeMSTeamsConversationId(rawConversationId); + const conversationMessageId = extractMSTeamsConversationMessageId(rawConversationId); + const conversationType = conversation?.conversationType ?? "personal"; + const isChannel = conversationType === "channel"; + const teamId = activity.channelData?.team?.id; + const threadId = isChannel + ? (conversationMessageId ?? activity.replyToId ?? undefined) + : undefined; + const advertisedMedia = resolveMSTeamsAdvertisedMedia(entry.attachments, { + maxInlineBytes: mediaMaxBytes, + maxInlineTotalBytes: mediaMaxBytes, + }); + + return { + ...entry, + activity, + from: activity.from, + conversation, + rawBody: entry.text, + advertisedMedia, + quoteInfo: extractMSTeamsQuoteInfo(entry.attachments), + attachmentTypes: entry.attachments + .map((attachment) => + typeof attachment.contentType === "string" ? attachment.contentType : undefined, + ) + .filter(Boolean) + .slice(0, 3), + htmlSummary: summarizeMSTeamsHtmlAttachments(entry.attachments), + rawConversationId, + conversationId, + conversationMessageId, + conversationType, + isChannel, + teamId, + graphChannelId: activity.channelData?.channel?.id?.trim() || conversationId, + threadId, + conversationRef: buildStoredConversationReference({ + activity, + conversationId, + conversationType, + teamId, + threadId, + }), + }; +} diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index b4f756582c69..bd6db2f47859 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -25,15 +25,6 @@ import { type HistoryEntry, } from "openclaw/plugin-sdk/reply-history"; import { sliceUtf16Safe, truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; -import { serializeMSTeamsAdaptiveCardActionValue } from "../adaptive-card-submit.js"; -import { - resolveMSTeamsAdvertisedMedia, - summarizeMSTeamsHtmlAttachments, - type MSTeamsAttachmentLike, -} from "../attachments.js"; -import { extractHtmlFromAttachment } from "../attachments/shared.js"; -import { tryNormalizeBotFrameworkServiceUrl } from "../bot-framework-service-url.js"; -import type { StoredConversationReference } from "../conversation-store.js"; import { formatUnknownError } from "../errors.js"; import { fetchChannelMessage, @@ -42,15 +33,17 @@ import { formatThreadContext, type GraphThreadMessage, } from "../graph-thread.js"; -import { - extractMSTeamsConversationMessageId, - extractMSTeamsQuoteInfo, - normalizeMSTeamsConversationId, - parseMSTeamsActivityTimestamp, - stripMSTeamsMentionTags, - wasMSTeamsBotMentioned, -} from "../inbound.js"; +import { normalizeMSTeamsConversationId, parseMSTeamsActivityTimestamp } from "../inbound.js"; +import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; +import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js"; +import { resolveMSTeamsAllowlistMatch, resolveMSTeamsReplyPolicy } from "../policy.js"; +import { extractMSTeamsPollVote } from "../polls.js"; +import { createMSTeamsReplyDispatcher } from "../reply-dispatcher.js"; import { createMSTeamsInboundDeadline, withMSTeamsRequestDeadline } from "../request-timeout.js"; +import { getMSTeamsRuntime } from "../runtime.js"; +import type { MSTeamsTurnContext } from "../sdk-types.js"; +import { recordMSTeamsSentMessage } from "../sent-message-cache.js"; +import { resolveTeamGroupId } from "../team-identity.js"; import { fetchParentMessageCached, formatParentContextEvent, @@ -58,43 +51,12 @@ import { shouldInjectParentContext, summarizeParentMessage, } from "../thread-parent-context.js"; - -function extractTextFromHtmlAttachments(attachments: MSTeamsAttachmentLike[]): string { - for (const attachment of attachments) { - const raw = extractHtmlFromAttachment(attachment); - if (!raw) { - continue; - } - const text = raw - .replace(/]*>.*?<\/at>/gis, " ") - .replace(/]*href=["']([^"']+)["'][^>]*>(.*?)<\/a>/gis, "$2 $1") - .replace(//gi, "\n") - .replace(/<\/p>/gi, "\n") - .replace(/<[^>]+>/g, " ") - .replace(/ /gi, " ") - .replace(/&/gi, "&") - .replace(/\s+/g, " ") - .trim(); - if (text) { - return text; - } - } - return ""; -} - -import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; -import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js"; -import { resolveMSTeamsAllowlistMatch, resolveMSTeamsReplyPolicy } from "../policy.js"; -import { extractMSTeamsPollVote } from "../polls.js"; -import { createMSTeamsReplyDispatcher } from "../reply-dispatcher.js"; -import { getMSTeamsRuntime } from "../runtime.js"; -import type { MSTeamsTurnContext } from "../sdk-types.js"; -import { - recordMSTeamsSentMessage, - wasMSTeamsMessageSentWithPersistence, -} from "../sent-message-cache.js"; -import { resolveTeamGroupId } from "../team-identity.js"; import { resolveMSTeamsSenderAccess } from "./access.js"; +import { + assembleMSTeamsInboundFacts, + prepareMSTeamsDebounceEntry, + type MSTeamsDebounceEntry, +} from "./inbound-facts.js"; import { resolveMSTeamsInboundMedia, resolveMSTeamsInboundMediaBody, @@ -135,49 +97,6 @@ function formatMSTeamsSenderReason(params: { } } -function buildStoredConversationReference(params: { - activity: MSTeamsTurnContext["activity"]; - conversationId: string; - conversationType: string; - teamId?: string; - /** Thread root message ID for channel thread messages. */ - threadId?: string; -}): StoredConversationReference { - const { activity, conversationId, conversationType, teamId, threadId } = params; - const from = activity.from; - const conversation = activity.conversation; - const agent = activity.recipient; - const clientInfo = activity.entities?.find((e) => e.type === "clientInfo") as - | { timezone?: string } - | undefined; - // Bot Framework requires `tenantId` on outbound proactive activities so the - // connector can route them to the correct Azure AD tenant; missing it causes - // HTTP 403. Channel activities often leave `conversation.tenantId` unset, so - // prefer the canonical `channelData.tenant.id` source when available. - const channelDataTenantId = activity.channelData?.tenant?.id; - const tenantId = channelDataTenantId ?? conversation?.tenantId; - const aadObjectId = from?.aadObjectId; - const serviceUrl = tryNormalizeBotFrameworkServiceUrl(activity.serviceUrl); - return { - activityId: activity.id, - user: from ? { id: from.id, name: from.name, aadObjectId: from.aadObjectId } : undefined, - agent, - conversation: { - id: conversationId, - conversationType, - tenantId, - }, - ...(tenantId ? { tenantId } : {}), - ...(aadObjectId ? { aadObjectId } : {}), - teamId, - channelId: activity.channelId, - ...(serviceUrl ? { serviceUrl } : {}), - locale: activity.locale, - ...(clientInfo?.timezone ? { timezone: clientInfo.timezone } : {}), - ...(threadId ? { threadId } : {}), - }; -} - export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { const { cfg, @@ -214,41 +133,34 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { channel: "msteams", }); - type MSTeamsDebounceEntry = { - context: MSTeamsTurnContext; - rawText: string; - text: string; - attachments: MSTeamsAttachmentLike[]; - wasMentioned: boolean; - implicitMentionKinds: Array<"reply_to_bot">; - turnAdoptionLifecycle?: MSTeamsIngressLifecycle; - }; - const handleTeamsMessageNow = async (params: MSTeamsDebounceEntry) => { - const context = params.context; - const activity = context.activity; - const rawText = params.rawText; - const text = params.text; - const attachments = params.attachments; - const advertisedMedia = resolveMSTeamsAdvertisedMedia(attachments, { - maxInlineBytes: mediaMaxBytes, - maxInlineTotalBytes: mediaMaxBytes, - }); - const rawBody = text; + const facts = assembleMSTeamsInboundFacts({ entry: params, mediaMaxBytes }); + const { + context, + activity, + rawText, + text, + attachments, + advertisedMedia, + rawBody, + quoteInfo, + from, + conversation, + attachmentTypes, + htmlSummary, + conversationId, + conversationMessageId, + conversationType, + isChannel, + teamId, + graphChannelId, + conversationRef, + } = facts; const historyBody = [text, formatMediaPlaceholderText(advertisedMedia)] .filter(Boolean) .join("\n"); - const quoteInfo = extractMSTeamsQuoteInfo(attachments); let quoteSenderId: string | undefined; let quoteSenderName: string | undefined; - const from = activity.from; - const conversation = activity.conversation; - - const attachmentTypes = attachments - .map((att) => (typeof att.contentType === "string" ? att.contentType : undefined)) - .filter(Boolean) - .slice(0, 3); - const htmlSummary = summarizeMSTeamsHtmlAttachments(attachments); log.info("received message", { rawText: truncateUtf16Safe(rawText, 50), @@ -267,28 +179,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { return; } - // Teams conversation.id may include ";messageid=..." suffix - strip it for session key. - const rawConversationId = conversation?.id ?? ""; - const conversationId = normalizeMSTeamsConversationId(rawConversationId); - const conversationMessageId = extractMSTeamsConversationMessageId(rawConversationId); - const conversationType = conversation?.conversationType ?? "personal"; - const teamId = activity.channelData?.team?.id; - const graphChannelId = activity.channelData?.channel?.id?.trim() || conversationId; - // For channel thread messages, resolve the thread root message ID so outbound - // replies land in the correct thread. The root ID comes from the `messageid=` - // portion of conversation.id (preferred) or from activity.replyToId. - const threadId = - conversationType === "channel" - ? (conversationMessageId ?? activity.replyToId ?? undefined) - : undefined; - const conversationRef = buildStoredConversationReference({ - activity, - conversationId, - conversationType, - teamId, - threadId, - }); - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ cfg, surface: "msteams", @@ -314,8 +204,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { const commandAuthorized = commandAccess.requested ? commandAccess.authorized : undefined; const effectiveDmAllowFrom = senderAccess.effectiveAllowFrom; const effectiveGroupAllowFrom = senderAccess.effectiveGroupAllowFrom; - const isChannel = conversationType === "channel"; - if (isDirectMessage && msteamsCfg && senderAccess.decision !== "allow") { if (senderAccess.reasonCode === "dm_policy_disabled") { log.info("dropping dm (dms disabled)", { @@ -1112,34 +1000,11 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { context: MSTeamsTurnContext, turnAdoptionLifecycle?: MSTeamsIngressLifecycle, ) { - const activity = context.activity; - const attachments = Array.isArray(activity.attachments) - ? (activity.attachments as unknown as MSTeamsAttachmentLike[]) - : []; - const rawText = activity.text?.trim() ?? ""; - const htmlText = extractTextFromHtmlAttachments(attachments); - const valueText = - rawText || htmlText ? "" : serializeMSTeamsAdaptiveCardActionValue(activity.value); - const text = stripMSTeamsMentionTags(rawText || htmlText || valueText || ""); - const wasMentioned = wasMSTeamsBotMentioned(activity); - const conversationId = normalizeMSTeamsConversationId(activity.conversation?.id ?? ""); - const replyToId = activity.replyToId ?? undefined; - const implicitMentionKinds: Array<"reply_to_bot"> = - conversationId && - replyToId && - (await wasMSTeamsMessageSentWithPersistence({ conversationId, messageId: replyToId })) - ? ["reply_to_bot"] - : []; - - await inboundDebouncer.enqueue({ + const entry = await prepareMSTeamsDebounceEntry({ context, - rawText, - text, - attachments, - wasMentioned, - implicitMentionKinds, turnAdoptionLifecycle, }); + await inboundDebouncer.enqueue(entry); if (turnAdoptionLifecycle) { // Keep the durable claim held across the debounce window. The merged // flush completes it only when the reply lane adopts (or terminally skips). From f0215b7d621b49810d21572874bfb89956d40855 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 16:48:33 +0800 Subject: [PATCH 2/8] refactor(msteams): isolate message admission --- .../msteams/src/monitor-handler/access.ts | 223 +++++++++++++++++- .../src/monitor-handler/message-handler.ts | 205 ++-------------- 2 files changed, 244 insertions(+), 184 deletions(-) diff --git a/extensions/msteams/src/monitor-handler/access.ts b/extensions/msteams/src/monitor-handler/access.ts index 695a5f6969f1..a6641d65f238 100644 --- a/extensions/msteams/src/monitor-handler/access.ts +++ b/extensions/msteams/src/monitor-handler/access.ts @@ -1,4 +1,6 @@ // Msteams plugin module implements access behavior. +import { formatAllowlistMatchMeta } from "openclaw/plugin-sdk/allow-from"; +import { logInboundDrop } from "openclaw/plugin-sdk/channel-inbound"; import { channelIngressRoutes, resolveStableChannelMessageIngress, @@ -12,8 +14,12 @@ import { resolveDefaultGroupPolicy, type OpenClawConfig, } from "../../runtime-api.js"; +import type { StoredConversationReference } from "../conversation-store.js"; +import type { MSTeamsConversationStore } from "../conversation-store.js"; +import { formatUnknownError } from "../errors.js"; import { normalizeMSTeamsConversationId } from "../inbound.js"; -import { resolveMSTeamsRouteConfig } from "../policy.js"; +import type { MSTeamsMonitorLogger } from "../monitor-types.js"; +import { resolveMSTeamsAllowlistMatch, resolveMSTeamsRouteConfig } from "../policy.js"; import { looksLikeMSTeamsConversationId } from "../resolve-allowlist.js"; import { getMSTeamsRuntime } from "../runtime.js"; import type { MSTeamsTurnContext } from "../sdk-types.js"; @@ -70,6 +76,38 @@ function normalizeAllowlistConversationId(value?: string | null): string | null return trimmed ? normalizeMSTeamsConversationId(trimmed) : null; } +function formatMSTeamsSenderReason(params: { + reasonCode: string; + dmPolicy?: string; + groupPolicy?: string; +}): string { + switch (params.reasonCode) { + case "dm_policy_open": + return "dmPolicy=open"; + case "dm_policy_disabled": + return "dmPolicy=disabled"; + case "dm_policy_pairing_required": + return "dmPolicy=pairing (not allowlisted)"; + case "dm_policy_allowlisted": + return `dmPolicy=${params.dmPolicy ?? "allowlist"} (allowlisted)`; + case "dm_policy_not_allowlisted": + return `dmPolicy=${params.dmPolicy ?? "allowlist"} (not allowlisted)`; + case "group_policy_disabled": + return "groupPolicy=disabled"; + case "group_policy_empty_allowlist": + case "route_sender_empty": + return "groupPolicy=allowlist (empty allowlist)"; + case "group_policy_not_allowlisted": + return "groupPolicy=allowlist (not allowlisted)"; + case "group_policy_open": + return "groupPolicy=open"; + case "group_policy_allowed": + return `groupPolicy=${params.groupPolicy ?? "allowlist"}`; + default: + return params.reasonCode; + } +} + export async function resolveMSTeamsSenderAccess(params: { cfg: OpenClawConfig; activity: MSTeamsTurnContext["activity"]; @@ -169,3 +207,186 @@ export async function resolveMSTeamsSenderAccess(params: { groupPolicy, }; } + +export async function admitMSTeamsMessage(params: { + cfg: OpenClawConfig; + activity: MSTeamsTurnContext["activity"]; + text: string; + conversationId: string; + conversationRef: StoredConversationReference; + isChannel: boolean; + conversationStore: MSTeamsConversationStore; + log: MSTeamsMonitorLogger; + logVerboseMessage: (message: string) => void; +}) { + const core = getMSTeamsRuntime(); + const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ + cfg: params.cfg, + surface: "msteams", + }); + const isControlCommand = + allowTextCommands && core.channel.commands.isControlCommandMessage(params.text, params.cfg); + const access = await resolveMSTeamsSenderAccess({ + cfg: params.cfg, + activity: params.activity, + hasControlCommand: isControlCommand, + }); + const { + dmPolicy, + senderId, + senderName, + pairing, + isDirectMessage, + channelGate, + senderAccess, + commandAccess, + allowNameMatching, + groupPolicy, + msteamsCfg, + } = access; + const effectiveDmAllowFrom = senderAccess.effectiveAllowFrom; + const effectiveGroupAllowFrom = senderAccess.effectiveGroupAllowFrom; + + if (isDirectMessage && msteamsCfg && senderAccess.decision !== "allow") { + if (senderAccess.reasonCode === "dm_policy_disabled") { + params.log.info("dropping dm (dms disabled)", { + sender: senderId, + label: senderName, + }); + params.log.debug?.("dropping dm (dms disabled)"); + return null; + } + const allowMatch = resolveMSTeamsAllowlistMatch({ + allowFrom: effectiveDmAllowFrom, + senderId, + senderName, + allowNameMatching, + }); + if (senderAccess.decision === "pairing") { + params.conversationStore + .upsert(params.conversationId, params.conversationRef) + .catch((err: unknown) => { + params.log.debug?.("failed to save conversation reference", { + error: formatUnknownError(err), + }); + }); + const request = await pairing.upsertPairingRequest({ + id: senderId, + meta: { name: senderName }, + }); + if (request) { + params.log.info("msteams pairing request created", { + sender: senderId, + label: senderName, + }); + } + } + params.log.debug?.("dropping dm (not allowlisted)", { + sender: senderId, + label: senderName, + allowlistMatch: formatAllowlistMatchMeta(allowMatch), + }); + params.log.info("dropping dm (not allowlisted)", { + sender: senderId, + label: senderName, + dmPolicy, + reason: formatMSTeamsSenderReason({ + reasonCode: senderAccess.reasonCode, + dmPolicy, + groupPolicy, + }), + allowlistMatch: formatAllowlistMatchMeta(allowMatch), + }); + return null; + } + + if (!isDirectMessage && msteamsCfg) { + if (channelGate.allowlistConfigured && !channelGate.allowed) { + params.log.info("dropping group message (not in team/channel allowlist)", { + conversationId: params.conversationId, + teamKey: channelGate.teamKey ?? "none", + channelKey: channelGate.channelKey ?? "none", + channelMatchKey: channelGate.channelMatchKey ?? "none", + channelMatchSource: channelGate.channelMatchSource ?? "none", + }); + params.log.debug?.("dropping group message (not in team/channel allowlist)", { + conversationId: params.conversationId, + teamKey: channelGate.teamKey ?? "none", + channelKey: channelGate.channelKey ?? "none", + channelMatchKey: channelGate.channelMatchKey ?? "none", + channelMatchSource: channelGate.channelMatchSource ?? "none", + }); + return null; + } + + if (!senderAccess.allowed && senderAccess.reasonCode === "group_policy_disabled") { + params.log.info("dropping group message (groupPolicy: disabled)", { + conversationId: params.conversationId, + }); + params.log.debug?.("dropping group message (groupPolicy: disabled)", { + conversationId: params.conversationId, + }); + return null; + } + if ( + !senderAccess.allowed && + (senderAccess.reasonCode === "group_policy_empty_allowlist" || + senderAccess.reasonCode === "route_sender_empty") + ) { + params.log.info("dropping group message (groupPolicy: allowlist, no allowlist)", { + conversationId: params.conversationId, + }); + params.log.debug?.("dropping group message (groupPolicy: allowlist, no allowlist)", { + conversationId: params.conversationId, + }); + return null; + } + if (!senderAccess.allowed && senderAccess.reasonCode === "group_policy_not_allowlisted") { + const allowMatch = resolveMSTeamsAllowlistMatch({ + allowFrom: effectiveGroupAllowFrom, + senderId, + senderName, + allowNameMatching, + }); + params.log.debug?.("dropping group message (not in groupAllowFrom)", { + sender: senderId, + label: senderName, + allowlistMatch: formatAllowlistMatchMeta(allowMatch), + }); + params.log.info("dropping group message (not in groupAllowFrom)", { + sender: senderId, + label: senderName, + allowlistMatch: formatAllowlistMatchMeta(allowMatch), + }); + return null; + } + } + + if (commandAccess.shouldBlockControlCommand) { + logInboundDrop({ + log: params.logVerboseMessage, + channel: "msteams", + reason: "control command (unauthorized)", + target: senderId, + }); + return null; + } + + params.conversationStore + .upsert(params.conversationId, params.conversationRef) + .catch((err: unknown) => { + params.log.debug?.("failed to save conversation reference", { + error: formatUnknownError(err), + }); + }); + + return { + ...access, + allowTextCommands, + isControlCommand, + commandAuthorized: commandAccess.requested ? commandAccess.authorized : undefined, + effectiveDmAllowFrom, + effectiveGroupAllowFrom, + isChannel: params.isChannel, + }; +} diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index bd6db2f47859..85f6775e1759 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -1,10 +1,8 @@ // Msteams plugin module implements message handler behavior. -import { formatAllowlistMatchMeta } from "openclaw/plugin-sdk/allow-from"; import { buildChannelInboundEventContext, createChannelInboundEnvelopeBuilder, formatMediaPlaceholderText, - logInboundDrop, resolveInboundMentionDecision, resolveInboundSupplementalSenderAllowed, toInboundMediaFactsWithMetadata, @@ -51,7 +49,7 @@ import { shouldInjectParentContext, summarizeParentMessage, } from "../thread-parent-context.js"; -import { resolveMSTeamsSenderAccess } from "./access.js"; +import { admitMSTeamsMessage } from "./access.js"; import { assembleMSTeamsInboundFacts, prepareMSTeamsDebounceEntry, @@ -65,38 +63,6 @@ import { } from "./inbound-media.js"; import { resolveMSTeamsRouteSessionKey } from "./thread-session.js"; -function formatMSTeamsSenderReason(params: { - reasonCode: string; - dmPolicy?: string; - groupPolicy?: string; -}): string { - switch (params.reasonCode) { - case "dm_policy_open": - return "dmPolicy=open"; - case "dm_policy_disabled": - return "dmPolicy=disabled"; - case "dm_policy_pairing_required": - return "dmPolicy=pairing (not allowlisted)"; - case "dm_policy_allowlisted": - return `dmPolicy=${params.dmPolicy ?? "allowlist"} (allowlisted)`; - case "dm_policy_not_allowlisted": - return `dmPolicy=${params.dmPolicy ?? "allowlist"} (not allowlisted)`; - case "group_policy_disabled": - return "groupPolicy=disabled"; - case "group_policy_empty_allowlist": - case "route_sender_empty": - return "groupPolicy=allowlist (empty allowlist)"; - case "group_policy_not_allowlisted": - return "groupPolicy=allowlist (not allowlisted)"; - case "group_policy_open": - return "groupPolicy=open"; - case "group_policy_allowed": - return `groupPolicy=${params.groupPolicy ?? "allowlist"}`; - default: - return params.reasonCode; - } -} - export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { const { cfg, @@ -179,159 +145,32 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { return; } - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ - cfg, - surface: "msteams", - }); - const isControlCommand = - allowTextCommands && core.channel.commands.isControlCommandMessage(text, cfg); - const { - dmPolicy, - senderId, - senderName, - pairing, - isDirectMessage, - channelGate, - senderAccess, - commandAccess, - allowNameMatching, - groupPolicy, - } = await resolveMSTeamsSenderAccess({ + const admission = await admitMSTeamsMessage({ cfg, activity, - hasControlCommand: isControlCommand, + text, + conversationId, + conversationRef, + isChannel, + conversationStore, + log, + logVerboseMessage, }); - const commandAuthorized = commandAccess.requested ? commandAccess.authorized : undefined; - const effectiveDmAllowFrom = senderAccess.effectiveAllowFrom; - const effectiveGroupAllowFrom = senderAccess.effectiveGroupAllowFrom; - if (isDirectMessage && msteamsCfg && senderAccess.decision !== "allow") { - if (senderAccess.reasonCode === "dm_policy_disabled") { - log.info("dropping dm (dms disabled)", { - sender: senderId, - label: senderName, - }); - log.debug?.("dropping dm (dms disabled)"); - return; - } - const allowMatch = resolveMSTeamsAllowlistMatch({ - allowFrom: effectiveDmAllowFrom, - senderId, - senderName, - allowNameMatching, - }); - if (senderAccess.decision === "pairing") { - conversationStore.upsert(conversationId, conversationRef).catch((err: unknown) => { - log.debug?.("failed to save conversation reference", { - error: formatUnknownError(err), - }); - }); - const request = await pairing.upsertPairingRequest({ - id: senderId, - meta: { name: senderName }, - }); - if (request) { - log.info("msteams pairing request created", { - sender: senderId, - label: senderName, - }); - } - } - log.debug?.("dropping dm (not allowlisted)", { - sender: senderId, - label: senderName, - allowlistMatch: formatAllowlistMatchMeta(allowMatch), - }); - log.info("dropping dm (not allowlisted)", { - sender: senderId, - label: senderName, - dmPolicy, - reason: formatMSTeamsSenderReason({ - reasonCode: senderAccess.reasonCode, - dmPolicy, - groupPolicy, - }), - allowlistMatch: formatAllowlistMatchMeta(allowMatch), - }); + if (!admission) { return; } - - if (!isDirectMessage && msteamsCfg) { - if (channelGate.allowlistConfigured && !channelGate.allowed) { - log.info("dropping group message (not in team/channel allowlist)", { - conversationId, - teamKey: channelGate.teamKey ?? "none", - channelKey: channelGate.channelKey ?? "none", - channelMatchKey: channelGate.channelMatchKey ?? "none", - channelMatchSource: channelGate.channelMatchSource ?? "none", - }); - log.debug?.("dropping group message (not in team/channel allowlist)", { - conversationId, - teamKey: channelGate.teamKey ?? "none", - channelKey: channelGate.channelKey ?? "none", - channelMatchKey: channelGate.channelMatchKey ?? "none", - channelMatchSource: channelGate.channelMatchSource ?? "none", - }); - return; - } - - if (!senderAccess.allowed && senderAccess.reasonCode === "group_policy_disabled") { - log.info("dropping group message (groupPolicy: disabled)", { - conversationId, - }); - log.debug?.("dropping group message (groupPolicy: disabled)", { - conversationId, - }); - return; - } - if ( - !senderAccess.allowed && - (senderAccess.reasonCode === "group_policy_empty_allowlist" || - senderAccess.reasonCode === "route_sender_empty") - ) { - log.info("dropping group message (groupPolicy: allowlist, no allowlist)", { - conversationId, - }); - log.debug?.("dropping group message (groupPolicy: allowlist, no allowlist)", { - conversationId, - }); - return; - } - if (!senderAccess.allowed && senderAccess.reasonCode === "group_policy_not_allowlisted") { - const allowMatch = resolveMSTeamsAllowlistMatch({ - allowFrom: effectiveGroupAllowFrom, - senderId, - senderName, - allowNameMatching, - }); - log.debug?.("dropping group message (not in groupAllowFrom)", { - sender: senderId, - label: senderName, - allowlistMatch: formatAllowlistMatchMeta(allowMatch), - }); - log.info("dropping group message (not in groupAllowFrom)", { - sender: senderId, - label: senderName, - allowlistMatch: formatAllowlistMatchMeta(allowMatch), - }); - return; - } - } - - if (commandAccess.shouldBlockControlCommand) { - logInboundDrop({ - log: logVerboseMessage, - channel: "msteams", - reason: "control command (unauthorized)", - target: senderId, - }); - return; - } - - conversationStore.upsert(conversationId, conversationRef).catch((err: unknown) => { - log.debug?.("failed to save conversation reference", { - error: formatUnknownError(err), - }); - }); + const { + senderId, + senderName, + isDirectMessage, + channelGate, + allowNameMatching, + groupPolicy, + commandAuthorized, + effectiveGroupAllowFrom, + allowTextCommands, + isControlCommand, + } = admission; const pollVote = extractMSTeamsPollVote(activity); if (pollVote) { From 29e98dd96db77e25cf12cc1e0c1578ace898fa4d Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 16:52:35 +0800 Subject: [PATCH 3/8] refactor(msteams): isolate inbound media assembly --- .../src/monitor-handler/inbound-content.ts | 101 ++++++++++++++++++ .../src/monitor-handler/message-handler.ts | 88 ++++----------- 2 files changed, 121 insertions(+), 68 deletions(-) create mode 100644 extensions/msteams/src/monitor-handler/inbound-content.ts diff --git a/extensions/msteams/src/monitor-handler/inbound-content.ts b/extensions/msteams/src/monitor-handler/inbound-content.ts new file mode 100644 index 000000000000..e96fa1521c8d --- /dev/null +++ b/extensions/msteams/src/monitor-handler/inbound-content.ts @@ -0,0 +1,101 @@ +// Msteams plugin module materializes inbound media and agent content. +import type { MSTeamsHtmlAttachmentSummary, MSTeamsInboundMedia } from "../attachments.js"; +import { formatUnknownError } from "../errors.js"; +import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; +import type { MSTeamsRequestDeadline } from "../request-timeout.js"; +import { withMSTeamsRequestDeadline } from "../request-timeout.js"; +import type { MSTeamsDebounceEntry } from "./inbound-facts.js"; +import { + mergeMSTeamsMediaFacts, + resolveMSTeamsInboundMedia, + resolveMSTeamsInboundMediaBody, + shouldAttemptMSTeamsGraphMediaFallback, +} from "./inbound-media.js"; + +export async function prepareMSTeamsInboundContent(params: { + entry: MSTeamsDebounceEntry; + rawBody: string; + advertisedMedia: MSTeamsInboundMedia[]; + htmlSummary?: MSTeamsHtmlAttachmentSummary; + conversationType: string; + conversationId: string; + conversationMessageId?: string; + teamAadGroupId?: string; + resolveTeamAadGroupId: () => Promise; + mediaMaxBytes: number; + tokenProvider: MSTeamsMessageHandlerDeps["tokenProvider"]; + mediaAllowHosts?: Parameters[0]["allowHosts"]; + mediaAuthAllowHosts?: Parameters[0]["authAllowHosts"]; + graphMediaFallback?: boolean; + deadline: MSTeamsRequestDeadline; + log: MSTeamsMessageHandlerDeps["log"]; +}) { + const activity = params.entry.context.activity; + const mayRecoverGraphMedia = + Boolean(params.htmlSummary?.attachmentIds.length) || + shouldAttemptMSTeamsGraphMediaFallback({ + conversationType: params.conversationType, + htmlSummary: params.htmlSummary, + graphMediaFallback: params.graphMediaFallback, + }); + if (!params.rawBody && params.advertisedMedia.length === 0 && !mayRecoverGraphMedia) { + params.log.debug?.("skipping empty message after stripping mentions"); + return null; + } + + let mediaList = [] as Awaited>; + try { + mediaList = await withMSTeamsRequestDeadline({ + deadline: params.deadline, + label: "MS Teams inbound media", + work: () => + resolveMSTeamsInboundMedia({ + attachments: params.entry.attachments, + htmlSummary: params.htmlSummary, + maxBytes: params.mediaMaxBytes, + tokenProvider: params.tokenProvider, + allowHosts: params.mediaAllowHosts, + authAllowHosts: params.mediaAuthAllowHosts, + graphMediaFallback: params.graphMediaFallback, + conversationType: params.conversationType, + conversationId: params.conversationId, + conversationMessageId: params.conversationMessageId, + teamAadGroupId: params.teamAadGroupId, + resolveTeamAadGroupId: params.resolveTeamAadGroupId, + serviceUrl: activity.serviceUrl, + activity: { + id: activity.id, + replyToId: activity.replyToId, + channelData: activity.channelData, + }, + log: params.log, + deadline: params.deadline, + preserveFilenames: false, + }), + }); + } catch (err) { + params.log.debug?.("failed to resolve inbound Teams media", { + error: formatUnknownError(err), + }); + } + + const inboundMedia = mergeMSTeamsMediaFacts(params.advertisedMedia, mediaList); + const nativeMediaForComparison = [ + ...params.advertisedMedia, + ...mediaList.slice(params.advertisedMedia.length).map((media) => ({ + contentType: media.contentType, + kind: media.kind, + })), + ]; + const agentBody = resolveMSTeamsInboundMediaBody({ + body: params.rawBody, + nativeMedia: nativeMediaForComparison, + materializedMedia: inboundMedia, + }); + if (!agentBody && inboundMedia.length === 0) { + params.log.debug?.("skipping empty message after Graph media recovery"); + return null; + } + + return { agentBody, inboundMedia }; +} diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index 85f6775e1759..e4698c11a689 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -50,17 +50,12 @@ import { summarizeParentMessage, } from "../thread-parent-context.js"; import { admitMSTeamsMessage } from "./access.js"; +import { prepareMSTeamsInboundContent } from "./inbound-content.js"; import { assembleMSTeamsInboundFacts, prepareMSTeamsDebounceEntry, type MSTeamsDebounceEntry, } from "./inbound-facts.js"; -import { - resolveMSTeamsInboundMedia, - resolveMSTeamsInboundMediaBody, - mergeMSTeamsMediaFacts, - shouldAttemptMSTeamsGraphMediaFallback, -} from "./inbound-media.js"; import { resolveMSTeamsRouteSessionKey } from "./thread-session.js"; export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { @@ -200,18 +195,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { return; } - const mayRecoverGraphMedia = - Boolean(htmlSummary?.attachmentIds.length) || - shouldAttemptMSTeamsGraphMediaFallback({ - conversationType, - htmlSummary: htmlSummary ?? undefined, - graphMediaFallback: msteamsCfg?.graphMediaFallback, - }); - if (!rawBody && advertisedMedia.length === 0 && !mayRecoverGraphMedia) { - log.debug?.("skipping empty message after stripping mentions"); - return; - } - const teamsFrom = isDirectMessage ? `msteams:${senderId}` : isChannel @@ -324,59 +307,28 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { teamAadGroupId = await teamGroupIdPromise; return teamAadGroupId; }; - let mediaList = [] as Awaited>; - try { - mediaList = await withMSTeamsRequestDeadline({ - deadline: preprocessingDeadline, - label: "MS Teams inbound media", - work: () => - resolveMSTeamsInboundMedia({ - attachments, - htmlSummary: htmlSummary ?? undefined, - maxBytes: mediaMaxBytes, - tokenProvider, - allowHosts: msteamsCfg?.mediaAllowHosts, - authAllowHosts: msteamsCfg?.mediaAuthAllowHosts, - graphMediaFallback: msteamsCfg?.graphMediaFallback, - conversationType, - conversationId, - conversationMessageId: conversationMessageId ?? undefined, - teamAadGroupId, - resolveTeamAadGroupId: resolveChannelTeamGroupId, - serviceUrl: activity.serviceUrl, - activity: { - id: activity.id, - replyToId: activity.replyToId, - channelData: activity.channelData, - }, - log, - deadline: preprocessingDeadline, - preserveFilenames: false, - }), - }); - } catch (err) { - log.debug?.("failed to resolve inbound Teams media", { - error: formatUnknownError(err), - }); - } - - const inboundMedia = mergeMSTeamsMediaFacts(advertisedMedia, mediaList); - const nativeMediaForComparison = [ - ...advertisedMedia, - ...mediaList.slice(advertisedMedia.length).map((media) => ({ - contentType: media.contentType, - kind: media.kind, - })), - ]; - const agentBody = resolveMSTeamsInboundMediaBody({ - body: rawBody, - nativeMedia: nativeMediaForComparison, - materializedMedia: inboundMedia, + const content = await prepareMSTeamsInboundContent({ + entry: params, + rawBody, + advertisedMedia, + htmlSummary: htmlSummary ?? undefined, + conversationType, + conversationId, + conversationMessageId: conversationMessageId ?? undefined, + teamAadGroupId, + resolveTeamAadGroupId: resolveChannelTeamGroupId, + mediaMaxBytes, + tokenProvider, + mediaAllowHosts: msteamsCfg?.mediaAllowHosts, + mediaAuthAllowHosts: msteamsCfg?.mediaAuthAllowHosts, + graphMediaFallback: msteamsCfg?.graphMediaFallback, + deadline: preprocessingDeadline, + log, }); - if (!agentBody && inboundMedia.length === 0) { - log.debug?.("skipping empty message after Graph media recovery"); + if (!content) { return; } + const { agentBody, inboundMedia } = content; enqueuePrimaryMessageSystemEvent(); teamAadGroupId = await resolveChannelTeamGroupId(); From 12d3a2f2f6ff9b535d6f9100533e72fa46de4b20 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 16:55:51 +0800 Subject: [PATCH 4/8] refactor(msteams): isolate thread context routing --- .../src/monitor-handler/message-handler.ts | 237 ++--------------- .../src/monitor-handler/thread-context.ts | 239 ++++++++++++++++++ 2 files changed, 267 insertions(+), 209 deletions(-) create mode 100644 extensions/msteams/src/monitor-handler/thread-context.ts diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index e4698c11a689..7870c9a87e15 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -13,10 +13,7 @@ import { } from "openclaw/plugin-sdk/channel-inbound"; import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime"; import { bindIngressLifecycleToReplyOptions } from "openclaw/plugin-sdk/channel-outbound"; -import { - filterSupplementalContextItems, - resolveChannelContextVisibilityMode, -} from "openclaw/plugin-sdk/context-visibility-runtime"; +import { resolveChannelContextVisibilityMode } from "openclaw/plugin-sdk/context-visibility-runtime"; import { DEFAULT_GROUP_HISTORY_LIMIT, createChannelHistoryWindow, @@ -24,31 +21,15 @@ import { } from "openclaw/plugin-sdk/reply-history"; import { sliceUtf16Safe, truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { formatUnknownError } from "../errors.js"; -import { - fetchChannelMessage, - fetchChatMessageText, - fetchThreadReplies, - formatThreadContext, - type GraphThreadMessage, -} from "../graph-thread.js"; import { normalizeMSTeamsConversationId, parseMSTeamsActivityTimestamp } from "../inbound.js"; import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js"; import { resolveMSTeamsAllowlistMatch, resolveMSTeamsReplyPolicy } from "../policy.js"; import { extractMSTeamsPollVote } from "../polls.js"; import { createMSTeamsReplyDispatcher } from "../reply-dispatcher.js"; -import { createMSTeamsInboundDeadline, withMSTeamsRequestDeadline } from "../request-timeout.js"; import { getMSTeamsRuntime } from "../runtime.js"; import type { MSTeamsTurnContext } from "../sdk-types.js"; import { recordMSTeamsSentMessage } from "../sent-message-cache.js"; -import { resolveTeamGroupId } from "../team-identity.js"; -import { - fetchParentMessageCached, - formatParentContextEvent, - markParentContextInjected, - shouldInjectParentContext, - summarizeParentMessage, -} from "../thread-parent-context.js"; import { admitMSTeamsMessage } from "./access.js"; import { prepareMSTeamsInboundContent } from "./inbound-content.js"; import { @@ -56,7 +37,7 @@ import { prepareMSTeamsDebounceEntry, type MSTeamsDebounceEntry, } from "./inbound-facts.js"; -import { resolveMSTeamsRouteSessionKey } from "./thread-session.js"; +import { prepareMSTeamsThreadRouting, resolveMSTeamsThreadContext } from "./thread-context.js"; export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { const { @@ -120,9 +101,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { const historyBody = [text, formatMediaPlaceholderText(advertisedMedia)] .filter(Boolean) .join("\n"); - let quoteSenderId: string | undefined; - let quoteSenderName: string | undefined; - log.info("received message", { rawText: truncateUtf16Safe(rawText, 50), text: truncateUtf16Safe(text, 50), @@ -202,27 +180,18 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { : `msteams:group:${conversationId}`; const teamsTo = isDirectMessage ? `user:${senderId}` : `conversation:${conversationId}`; - const route = core.channel.routing.resolveAgentRoute({ + const threadRouting = prepareMSTeamsThreadRouting({ cfg, - channel: "msteams", - teamId, - peer: { - kind: isDirectMessage ? "direct" : isChannel ? "channel" : "group", - id: isDirectMessage ? senderId : conversationId, - }, - }); - - // Isolate channel thread sessions: each thread gets its own session key so - // context does not bleed across threads. Prefer conversationMessageId (the - // ;messageid= portion of conversation.id, i.e. the thread root) over - // activity.replyToId (which may point to a non-root parent in deep threads). - // DMs and group chats are unaffected — only channel thread replies fork. - route.sessionKey = resolveMSTeamsRouteSessionKey({ - baseSessionKey: route.sessionKey, + context, + isDirectMessage, isChannel, - conversationMessageId, - replyToId: activity.replyToId, + senderId, + conversationId, + conversationMessageId: conversationMessageId ?? undefined, + teamId, + log, }); + const { route, deadline: preprocessingDeadline } = threadRouting; const preview = sliceUtf16Safe(rawBody.replace(/\s+/g, " "), 0, 160); const inboundLabel = isDirectMessage @@ -284,29 +253,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { return; } } - const preprocessingDeadline = createMSTeamsInboundDeadline(); - let teamAadGroupId = activity.channelData?.team?.aadGroupId?.trim() || undefined; - const conversationTeamId = isChannel ? teamId : undefined; - let teamGroupIdPromise: Promise | undefined; - const resolveChannelTeamGroupId = async (): Promise => { - if (!conversationTeamId) { - return undefined; - } - teamGroupIdPromise ??= resolveTeamGroupId({ - conversationTeamId, - aadGroupId: teamAadGroupId, - getTeamDetails: context.getTeamDetails, - deadline: preprocessingDeadline, - }).catch((err: unknown) => { - log.debug?.("failed to resolve Teams AAD group ID", { - teamId: conversationTeamId, - error: formatUnknownError(err), - }); - return undefined; - }); - teamAadGroupId = await teamGroupIdPromise; - return teamAadGroupId; - }; const content = await prepareMSTeamsInboundContent({ entry: params, rawBody, @@ -315,8 +261,8 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { conversationType, conversationId, conversationMessageId: conversationMessageId ?? undefined, - teamAadGroupId, - resolveTeamAadGroupId: resolveChannelTeamGroupId, + teamAadGroupId: threadRouting.getTeamAadGroupId(), + resolveTeamAadGroupId: threadRouting.resolveTeamAadGroupId, mediaMaxBytes, tokenProvider, mediaAllowHosts: msteamsCfg?.mediaAllowHosts, @@ -330,149 +276,22 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { } const { agentBody, inboundMedia } = content; enqueuePrimaryMessageSystemEvent(); - teamAadGroupId = await resolveChannelTeamGroupId(); - // Media is the primary payload, so optional quote enrichment only gets the - // remaining preprocessing budget. DMs alone may fetch the full quote: group - // and channel quotes retain their visibility-filtered preview. - let quoteBodyFull: string | undefined; - const quoteMessageId = quoteInfo?.id; - if (quoteMessageId && isDirectMessage && conversationId.startsWith("19:")) { - try { - const graphToken = await withMSTeamsRequestDeadline({ - deadline: preprocessingDeadline, - label: "MS Teams quote token", - work: () => tokenProvider.getAccessToken("https://graph.microsoft.com"), - }); - quoteBodyFull = await withMSTeamsRequestDeadline({ - deadline: preprocessingDeadline, - label: "MS Teams quote lookup", - work: () => - fetchChatMessageText(graphToken, conversationId, quoteMessageId, preprocessingDeadline), - }); - } catch (err) { - log.debug?.("failed to fetch full quoted message text", { - error: formatUnknownError(err), - }); - } - } - - // Fetch thread history when the message is a reply inside a Teams channel thread. - // This is a best-effort enhancement; errors are logged and do not block the reply. - // - // We also enqueue a compact `Replying to @sender: …` system event when the parent - // is resolvable. On brand-new thread sessions (see PR #62713), this gives the agent - // immediate parent context even before the fuller `[Thread history]` block is assembled. - // Parent fetches are cached (5 min LRU, 100 entries) and per-session deduped so - // consecutive replies in the same thread do not re-inject identical context. - let threadContext: string | undefined; - const threadParentId = activity.replyToId; - const channelGroupId = teamAadGroupId; - if (threadParentId && isChannel && channelGroupId) { - try { - const graphToken = await withMSTeamsRequestDeadline({ - deadline: preprocessingDeadline, - label: "MS Teams thread token", - work: () => tokenProvider.getAccessToken("https://graph.microsoft.com"), - }); - // Use allSettled so a failure in one fetch does not discard the other. - // For example, reply-fetch 403 should not throw away a successful parent fetch. - const [parentResult, repliesResult] = await withMSTeamsRequestDeadline({ - deadline: preprocessingDeadline, - label: "MS Teams thread history", - work: () => - Promise.allSettled([ - fetchParentMessageCached( - graphToken, - channelGroupId, - conversationId, - threadParentId, - (token, groupId, requestedChannelId, messageId) => - fetchChannelMessage( - token, - groupId, - requestedChannelId, - messageId, - preprocessingDeadline, - ), - ), - fetchThreadReplies( - graphToken, - channelGroupId, - conversationId, - threadParentId, - 50, - preprocessingDeadline, - ), - ]), - }); - const parentMsg = parentResult.status === "fulfilled" ? parentResult.value : undefined; - const replies = repliesResult.status === "fulfilled" ? repliesResult.value : []; - if (parentResult.status === "rejected") { - log.debug?.("failed to fetch parent message", { - error: formatUnknownError(parentResult.reason), - }); - } - if (repliesResult.status === "rejected") { - log.debug?.("failed to fetch thread replies", { - error: formatUnknownError(repliesResult.reason), - }); - } - const isThreadSenderAllowed = (msg: GraphThreadMessage) => - resolveInboundSupplementalSenderAllowed({ - isGroup: isChannel, - groupPolicy, - allowFrom: effectiveGroupAllowFrom, - isSenderAllowed: (allowFrom) => - resolveMSTeamsAllowlistMatch({ - allowFrom, - senderId: msg.from?.user?.id ?? "", - senderName: msg.from?.user?.displayName, - allowNameMatching, - }).allowed, - }); - const parentSummary = summarizeParentMessage(parentMsg); - const visibleParentMessages = parentMsg - ? filterSupplementalContextItems({ - items: [parentMsg], - mode: contextVisibilityMode, - kind: "thread", - isSenderAllowed: isThreadSenderAllowed, - }).items - : []; - if ( - parentSummary && - visibleParentMessages.length > 0 && - shouldInjectParentContext(route.sessionKey, threadParentId) - ) { - core.system.enqueueSystemEvent(formatParentContextEvent(parentSummary), { - sessionKey: route.sessionKey, - contextKey: `msteams:thread-parent:${conversationId}:${threadParentId}`, - }); - markParentContextInjected(route.sessionKey, threadParentId); - } - const allMessages = parentMsg ? [parentMsg, ...replies] : replies; - quoteSenderId = parentMsg?.from?.user?.id ?? parentMsg?.from?.application?.id ?? undefined; - quoteSenderName = - parentMsg?.from?.user?.displayName ?? - parentMsg?.from?.application?.displayName ?? - quoteInfo?.sender; - const { items: threadMessages } = filterSupplementalContextItems({ - items: allMessages, - mode: contextVisibilityMode, - kind: "thread", - isSenderAllowed: isThreadSenderAllowed, - }); - const formatted = formatThreadContext(threadMessages, activity.id); - if (formatted) { - threadContext = formatted; - } - } catch (err) { - log.debug?.("failed to fetch thread history", { error: formatUnknownError(err) }); - // Graceful degradation: thread history is an optional enhancement. - } - } - quoteSenderName ??= quoteInfo?.sender; + const { teamAadGroupId, quoteBodyFull, quoteSenderId, quoteSenderName, threadContext } = + await resolveMSTeamsThreadContext({ + routing: threadRouting, + context, + tokenProvider, + quoteInfo, + isDirectMessage, + isChannel, + conversationId, + contextVisibilityMode, + groupPolicy, + effectiveGroupAllowFrom, + allowNameMatching, + log, + }); const envelopeFrom = isDirectMessage ? senderName : conversationType; const buildEnvelope = createChannelInboundEnvelopeBuilder({ cfg, route }); diff --git a/extensions/msteams/src/monitor-handler/thread-context.ts b/extensions/msteams/src/monitor-handler/thread-context.ts new file mode 100644 index 000000000000..f58e14245cee --- /dev/null +++ b/extensions/msteams/src/monitor-handler/thread-context.ts @@ -0,0 +1,239 @@ +// Msteams plugin module owns thread routing and Graph parent context. +import { resolveInboundSupplementalSenderAllowed } from "openclaw/plugin-sdk/channel-inbound"; +import { filterSupplementalContextItems } from "openclaw/plugin-sdk/context-visibility-runtime"; +import type { OpenClawConfig } from "../../runtime-api.js"; +import { formatUnknownError } from "../errors.js"; +import { + fetchChannelMessage, + fetchChatMessageText, + fetchThreadReplies, + formatThreadContext, + type GraphThreadMessage, +} from "../graph-thread.js"; +import type { extractMSTeamsQuoteInfo } from "../inbound.js"; +import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; +import { resolveMSTeamsAllowlistMatch } from "../policy.js"; +import { createMSTeamsInboundDeadline, withMSTeamsRequestDeadline } from "../request-timeout.js"; +import { getMSTeamsRuntime } from "../runtime.js"; +import type { MSTeamsTurnContext } from "../sdk-types.js"; +import { resolveTeamGroupId } from "../team-identity.js"; +import { + fetchParentMessageCached, + formatParentContextEvent, + markParentContextInjected, + shouldInjectParentContext, + summarizeParentMessage, +} from "../thread-parent-context.js"; +import { resolveMSTeamsRouteSessionKey } from "./thread-session.js"; + +export function prepareMSTeamsThreadRouting(params: { + cfg: OpenClawConfig; + context: MSTeamsTurnContext; + isDirectMessage: boolean; + isChannel: boolean; + senderId: string; + conversationId: string; + conversationMessageId?: string; + teamId?: string; + log: MSTeamsMessageHandlerDeps["log"]; +}) { + const core = getMSTeamsRuntime(); + const route = core.channel.routing.resolveAgentRoute({ + cfg: params.cfg, + channel: "msteams", + teamId: params.teamId, + peer: { + kind: params.isDirectMessage ? "direct" : params.isChannel ? "channel" : "group", + id: params.isDirectMessage ? params.senderId : params.conversationId, + }, + }); + route.sessionKey = resolveMSTeamsRouteSessionKey({ + baseSessionKey: route.sessionKey, + isChannel: params.isChannel, + conversationMessageId: params.conversationMessageId, + replyToId: params.context.activity.replyToId, + }); + + const deadline = createMSTeamsInboundDeadline(); + let teamAadGroupId = params.context.activity.channelData?.team?.aadGroupId?.trim() || undefined; + const conversationTeamId = params.isChannel ? params.teamId : undefined; + let teamGroupIdPromise: Promise | undefined; + const resolveTeamAadGroupId = async (): Promise => { + if (!conversationTeamId) { + return undefined; + } + teamGroupIdPromise ??= resolveTeamGroupId({ + conversationTeamId, + aadGroupId: teamAadGroupId, + getTeamDetails: params.context.getTeamDetails, + deadline, + }).catch((err: unknown) => { + params.log.debug?.("failed to resolve Teams AAD group ID", { + teamId: conversationTeamId, + error: formatUnknownError(err), + }); + return undefined; + }); + teamAadGroupId = await teamGroupIdPromise; + return teamAadGroupId; + }; + + return { + route, + deadline, + resolveTeamAadGroupId, + getTeamAadGroupId: () => teamAadGroupId, + }; +} + +export async function resolveMSTeamsThreadContext(params: { + routing: ReturnType; + context: MSTeamsTurnContext; + tokenProvider: MSTeamsMessageHandlerDeps["tokenProvider"]; + quoteInfo: ReturnType; + isDirectMessage: boolean; + isChannel: boolean; + conversationId: string; + contextVisibilityMode: "all" | "allowlist" | "allowlist_quote"; + groupPolicy: Parameters[0]["groupPolicy"]; + effectiveGroupAllowFrom: Parameters< + typeof resolveInboundSupplementalSenderAllowed + >[0]["allowFrom"]; + allowNameMatching: boolean; + log: MSTeamsMessageHandlerDeps["log"]; +}) { + const core = getMSTeamsRuntime(); + const activity = params.context.activity; + const { route, deadline } = params.routing; + const teamAadGroupId = await params.routing.resolveTeamAadGroupId(); + let quoteBodyFull: string | undefined; + let quoteSenderId: string | undefined; + let quoteSenderName: string | undefined; + const quoteMessageId = params.quoteInfo?.id; + if (quoteMessageId && params.isDirectMessage && params.conversationId.startsWith("19:")) { + try { + const graphToken = await withMSTeamsRequestDeadline({ + deadline, + label: "MS Teams quote token", + work: () => params.tokenProvider.getAccessToken("https://graph.microsoft.com"), + }); + quoteBodyFull = await withMSTeamsRequestDeadline({ + deadline, + label: "MS Teams quote lookup", + work: () => + fetchChatMessageText(graphToken, params.conversationId, quoteMessageId, deadline), + }); + } catch (err) { + params.log.debug?.("failed to fetch full quoted message text", { + error: formatUnknownError(err), + }); + } + } + + let threadContext: string | undefined; + const threadParentId = activity.replyToId; + if (threadParentId && params.isChannel && teamAadGroupId) { + try { + const graphToken = await withMSTeamsRequestDeadline({ + deadline, + label: "MS Teams thread token", + work: () => params.tokenProvider.getAccessToken("https://graph.microsoft.com"), + }); + // Keep successful parent context when the replies endpoint fails, and vice versa. + const [parentResult, repliesResult] = await withMSTeamsRequestDeadline({ + deadline, + label: "MS Teams thread history", + work: () => + Promise.allSettled([ + fetchParentMessageCached( + graphToken, + teamAadGroupId, + params.conversationId, + threadParentId, + (token, groupId, requestedChannelId, messageId) => + fetchChannelMessage(token, groupId, requestedChannelId, messageId, deadline), + ), + fetchThreadReplies( + graphToken, + teamAadGroupId, + params.conversationId, + threadParentId, + 50, + deadline, + ), + ]), + }); + const parentMsg = parentResult.status === "fulfilled" ? parentResult.value : undefined; + const replies = repliesResult.status === "fulfilled" ? repliesResult.value : []; + if (parentResult.status === "rejected") { + params.log.debug?.("failed to fetch parent message", { + error: formatUnknownError(parentResult.reason), + }); + } + if (repliesResult.status === "rejected") { + params.log.debug?.("failed to fetch thread replies", { + error: formatUnknownError(repliesResult.reason), + }); + } + const isThreadSenderAllowed = (message: GraphThreadMessage) => + resolveInboundSupplementalSenderAllowed({ + isGroup: params.isChannel, + groupPolicy: params.groupPolicy, + allowFrom: params.effectiveGroupAllowFrom, + isSenderAllowed: (allowFrom) => + resolveMSTeamsAllowlistMatch({ + allowFrom, + senderId: message.from?.user?.id ?? "", + senderName: message.from?.user?.displayName, + allowNameMatching: params.allowNameMatching, + }).allowed, + }); + const parentSummary = summarizeParentMessage(parentMsg); + const visibleParentMessages = parentMsg + ? filterSupplementalContextItems({ + items: [parentMsg], + mode: params.contextVisibilityMode, + kind: "thread", + isSenderAllowed: isThreadSenderAllowed, + }).items + : []; + if ( + parentSummary && + visibleParentMessages.length > 0 && + shouldInjectParentContext(route.sessionKey, threadParentId) + ) { + core.system.enqueueSystemEvent(formatParentContextEvent(parentSummary), { + sessionKey: route.sessionKey, + contextKey: `msteams:thread-parent:${params.conversationId}:${threadParentId}`, + }); + markParentContextInjected(route.sessionKey, threadParentId); + } + const allMessages = parentMsg ? [parentMsg, ...replies] : replies; + quoteSenderId = parentMsg?.from?.user?.id ?? parentMsg?.from?.application?.id ?? undefined; + quoteSenderName = + parentMsg?.from?.user?.displayName ?? + parentMsg?.from?.application?.displayName ?? + params.quoteInfo?.sender; + const { items: threadMessages } = filterSupplementalContextItems({ + items: allMessages, + mode: params.contextVisibilityMode, + kind: "thread", + isSenderAllowed: isThreadSenderAllowed, + }); + threadContext = formatThreadContext(threadMessages, activity.id) || undefined; + } catch (err) { + params.log.debug?.("failed to fetch thread history", { + error: formatUnknownError(err), + }); + } + } + quoteSenderName ??= params.quoteInfo?.sender; + + return { + teamAadGroupId, + quoteBodyFull, + quoteSenderId, + quoteSenderName, + threadContext, + }; +} From 40bceb9f2b43805614bf15c538171565cabcf9be Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 16:59:21 +0800 Subject: [PATCH 5/8] refactor(msteams): isolate inbound dispatch lifecycle --- .../src/monitor-handler/inbound-dispatch.ts | 320 ++++++++++++++++++ .../src/monitor-handler/message-handler.ts | 310 ++--------------- 2 files changed, 355 insertions(+), 275 deletions(-) create mode 100644 extensions/msteams/src/monitor-handler/inbound-dispatch.ts diff --git a/extensions/msteams/src/monitor-handler/inbound-dispatch.ts b/extensions/msteams/src/monitor-handler/inbound-dispatch.ts new file mode 100644 index 000000000000..2ce335b36c49 --- /dev/null +++ b/extensions/msteams/src/monitor-handler/inbound-dispatch.ts @@ -0,0 +1,320 @@ +// Msteams plugin module dispatches prepared inbound turns and owns reply lifecycle handling. +import { + buildChannelInboundEventContext, + createChannelInboundEnvelopeBuilder, + hasFinalInboundReplyDispatch, + resolveInboundReplyDispatchCounts, + resolveInboundSupplementalSenderAllowed, + toInboundMediaFactsWithMetadata, +} from "openclaw/plugin-sdk/channel-inbound"; +import { bindIngressLifecycleToReplyOptions } from "openclaw/plugin-sdk/channel-outbound"; +import { createChannelHistoryWindow, type HistoryEntry } from "openclaw/plugin-sdk/reply-history"; +import { sliceUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import type { RuntimeEnv } from "../../runtime-api.js"; +import { formatUnknownError } from "../errors.js"; +import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; +import { resolveMSTeamsAllowlistMatch, resolveMSTeamsReplyPolicy } from "../policy.js"; +import { createMSTeamsReplyDispatcher } from "../reply-dispatcher.js"; +import { getMSTeamsRuntime } from "../runtime.js"; +import { recordMSTeamsSentMessage } from "../sent-message-cache.js"; +import type { admitMSTeamsMessage } from "./access.js"; +import type { prepareMSTeamsInboundContent } from "./inbound-content.js"; +import type { assembleMSTeamsInboundFacts } from "./inbound-facts.js"; +import type { prepareMSTeamsThreadRouting, resolveMSTeamsThreadContext } from "./thread-context.js"; + +export type MSTeamsInboundDispatchResult = + | { kind: "completed"; finalResponses: number } + | { kind: "failed" }; + +export async function dispatchMSTeamsInboundTurn(params: { + cfg: MSTeamsMessageHandlerDeps["cfg"]; + runtime: RuntimeEnv; + appId: string; + app: MSTeamsMessageHandlerDeps["app"]; + tokenProvider: MSTeamsMessageHandlerDeps["tokenProvider"]; + textLimit: number; + log: MSTeamsMessageHandlerDeps["log"]; + logVerboseMessage: (message: string) => void; + facts: ReturnType; + admission: NonNullable>>; + content: NonNullable>>; + routing: ReturnType; + thread: Awaited>; + replyStyle: ReturnType["replyStyle"]; + timestamp?: Date; + contextVisibilityMode: "all" | "allowlist" | "allowlist_quote"; + mentionWasEffective: boolean; + conversationHistories: Map; + historyLimit: number; +}): Promise { + const core = getMSTeamsRuntime(); + const { + cfg, + runtime, + appId, + app, + tokenProvider, + textLimit, + log, + logVerboseMessage, + facts, + admission, + content, + routing, + thread, + replyStyle, + timestamp, + contextVisibilityMode, + conversationHistories, + historyLimit, + } = params; + const { context, activity, rawBody, text, quoteInfo, conversationRef } = facts; + const { + senderId, + senderName, + isDirectMessage, + allowNameMatching, + groupPolicy, + commandAuthorized, + effectiveGroupAllowFrom, + } = admission; + const { route } = routing; + const { agentBody, inboundMedia } = content; + const { teamAadGroupId, quoteBodyFull, quoteSenderId, quoteSenderName, threadContext } = thread; + const { conversationId, conversationType, isChannel, teamId, graphChannelId } = facts; + const teamsFrom = isDirectMessage + ? `msteams:${senderId}` + : isChannel + ? `msteams:channel:${conversationId}` + : `msteams:group:${conversationId}`; + const teamsTo = isDirectMessage ? `user:${senderId}` : `conversation:${conversationId}`; + const envelopeFrom = isDirectMessage ? senderName : conversationType; + const buildEnvelope = createChannelInboundEnvelopeBuilder({ cfg, route }); + const body = buildEnvelope({ + channel: "Teams", + from: envelopeFrom, + timestamp, + body: agentBody, + }); + let combinedBody = body; + const isRoomish = !isDirectMessage; + const historyKey = isRoomish ? conversationId : undefined; + if (isRoomish && historyKey) { + const channelHistory = createChannelHistoryWindow({ historyMap: conversationHistories }); + combinedBody = channelHistory.buildPendingContext({ + historyKey, + limit: historyLimit, + currentMessage: combinedBody, + formatEntry: (entry) => + buildEnvelope({ + channel: "Teams", + from: conversationType, + timestamp: entry.timestamp, + previousTimestamp: null, + body: `${entry.sender}: ${entry.body}${entry.messageId ? ` [id:${entry.messageId}]` : ""}`, + }), + }); + } + + const inboundHistory = + isRoomish && historyKey && historyLimit > 0 + ? createChannelHistoryWindow({ historyMap: conversationHistories }).buildInboundHistory({ + historyKey, + limit: historyLimit, + }) + : undefined; + const commandBody = text.trim(); + const quoteSenderAllowed = + quoteInfo && quoteInfo.sender + ? resolveInboundSupplementalSenderAllowed({ + isGroup: !isDirectMessage, + groupPolicy, + allowFrom: effectiveGroupAllowFrom, + isSenderAllowed: (allowFrom) => + resolveMSTeamsAllowlistMatch({ + allowFrom, + senderId: quoteSenderId ?? "", + senderName: quoteSenderName, + allowNameMatching, + }).allowed, + }) + : true; + const bodyForAgent = threadContext + ? `[Thread history]\n${threadContext}\n[/Thread history]\n\n${agentBody}` + : agentBody; + // Teams channel actions need both the AAD group and Graph channel ids. + const nativeChannelId = + isChannel && teamAadGroupId ? `${teamAadGroupId}/${graphChannelId}` : undefined; + const ctxPayload = buildChannelInboundEventContext({ + channel: "msteams", + contextVisibility: contextVisibilityMode, + supplemental: { + quote: quoteInfo + ? { + id: quoteInfo.id ?? activity.replyToId ?? undefined, + body: quoteBodyFull ?? quoteInfo.body, + sender: quoteInfo.sender, + senderAllowed: quoteSenderAllowed, + isQuote: true, + } + : undefined, + }, + media: await toInboundMediaFactsWithMetadata(inboundMedia), + messageId: activity.id, + timestamp: timestamp?.getTime() ?? Date.now(), + from: teamsFrom, + sender: { + id: senderId, + name: senderName, + }, + conversation: { + kind: isDirectMessage ? "direct" : isChannel ? "channel" : "group", + id: conversationId, + label: envelopeFrom, + spaceId: teamId, + nativeChannelId, + }, + route: { + agentId: route.agentId, + dmScope: route.dmScope, + accountId: route.accountId, + routeSessionKey: route.sessionKey, + }, + reply: { + to: teamsTo, + replyToId: activity.replyToId ?? undefined, + nativeChannelId, + }, + message: { + body: combinedBody, + bodyForAgent, + inboundHistory, + rawBody, + commandBody, + }, + sessionTranscript: { historyLimit: isRoomish ? historyLimit : 0 }, + access: { + mentions: { + canDetectMention: !isDirectMessage, + wasMentioned: isDirectMessage || params.mentionWasEffective, + }, + commands: { + authorized: commandAuthorized === true, + }, + }, + extra: { + GroupSubject: !isDirectMessage ? conversationType : undefined, + ReplyToIsQuote: quoteInfo ? true : undefined, + }, + }); + + const preview = sliceUtf16Safe(rawBody.replace(/\s+/g, " "), 0, 160); + logVerboseMessage(`msteams inbound: from=${ctxPayload.From} preview="${preview}"`); + + const { dispatcherOptions, delivery, replyOptions } = createMSTeamsReplyDispatcher({ + cfg, + agentId: route.agentId, + sessionKey: route.sessionKey, + accountId: route.accountId, + runtime, + log, + app, + appId, + conversationRef, + context, + replyStyle, + textLimit, + onSentMessageIds: (ids) => { + for (const id of ids) { + recordMSTeamsSentMessage(conversationId, id); + } + }, + tokenProvider, + sharePointSiteId: cfg.channels?.msteams?.sharePointSiteId, + }); + + const activityClientInfo = activity.entities?.find((entity) => entity.type === "clientInfo") as + | { timezone?: string } + | undefined; + const senderTimezone = activityClientInfo?.timezone || conversationRef.timezone; + const turnConfig = + senderTimezone && !cfg.agents?.defaults?.userTimezone + ? { + ...cfg, + agents: { + ...cfg.agents, + defaults: { ...cfg.agents?.defaults, userTimezone: senderTimezone }, + }, + } + : cfg; + log.info("dispatching to agent", { sessionKey: route.sessionKey }); + try { + const turnResult = await core.channel.inbound.run({ + channel: "msteams", + accountId: route.accountId, + raw: context, + adapter: { + ingest: () => ({ + id: activity.id ?? `${teamsFrom}:${Date.now()}`, + timestamp: timestamp?.getTime(), + rawText: rawBody, + textForAgent: bodyForAgent, + textForCommands: commandBody, + raw: activity, + }), + resolveTurn: () => ({ + cfg: turnConfig, + channel: "msteams", + accountId: route.accountId, + route: { agentId: route.agentId, sessionKey: route.sessionKey }, + ctxPayload, + record: { + onRecordError: (err) => { + logVerboseMessage( + `msteams: failed updating session meta: ${formatUnknownError(err)}`, + ); + }, + }, + history: { + isGroup: isRoomish, + historyKey, + historyMap: conversationHistories, + limit: historyLimit, + }, + dispatcherOptions, + delivery, + replyOptions: { + ...replyOptions, + ...(facts.turnAdoptionLifecycle + ? bindIngressLifecycleToReplyOptions(facts.turnAdoptionLifecycle) + : {}), + }, + }), + }, + }); + const dispatchResult = turnResult.dispatched ? turnResult.dispatchResult : undefined; + const counts = resolveInboundReplyDispatchCounts(dispatchResult); + log.info("dispatch complete", { + queuedFinal: dispatchResult?.queuedFinal ?? false, + counts, + }); + if (hasFinalInboundReplyDispatch(dispatchResult)) { + logVerboseMessage( + `msteams: delivered ${counts.final} repl${counts.final === 1 ? "y" : "ies"} to ${teamsTo}`, + ); + } + return { kind: "completed", finalResponses: counts.final }; + } catch (err) { + log.error("dispatch failed", { error: formatUnknownError(err) }); + runtime.error(`msteams dispatch failed: ${formatUnknownError(err)}`); + if (facts.turnAdoptionLifecycle) { + throw err; + } + try { + await context.sendActivity("⚠️ Something went wrong. Please try again."); + } catch { + // Best effort. + } + return { kind: "failed" }; + } +} diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index 7870c9a87e15..9be9de18cec0 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -1,37 +1,27 @@ // Msteams plugin module implements message handler behavior. import { - buildChannelInboundEventContext, - createChannelInboundEnvelopeBuilder, formatMediaPlaceholderText, resolveInboundMentionDecision, - resolveInboundSupplementalSenderAllowed, - toInboundMediaFactsWithMetadata, -} from "openclaw/plugin-sdk/channel-inbound"; -import { - hasFinalInboundReplyDispatch, - resolveInboundReplyDispatchCounts, } from "openclaw/plugin-sdk/channel-inbound"; import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime"; -import { bindIngressLifecycleToReplyOptions } from "openclaw/plugin-sdk/channel-outbound"; import { resolveChannelContextVisibilityMode } from "openclaw/plugin-sdk/context-visibility-runtime"; import { DEFAULT_GROUP_HISTORY_LIMIT, createChannelHistoryWindow, type HistoryEntry, } from "openclaw/plugin-sdk/reply-history"; -import { sliceUtf16Safe, truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { formatUnknownError } from "../errors.js"; import { normalizeMSTeamsConversationId, parseMSTeamsActivityTimestamp } from "../inbound.js"; import type { MSTeamsMessageHandlerDeps } from "../monitor-handler.types.js"; import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js"; -import { resolveMSTeamsAllowlistMatch, resolveMSTeamsReplyPolicy } from "../policy.js"; +import { resolveMSTeamsReplyPolicy } from "../policy.js"; import { extractMSTeamsPollVote } from "../polls.js"; -import { createMSTeamsReplyDispatcher } from "../reply-dispatcher.js"; import { getMSTeamsRuntime } from "../runtime.js"; import type { MSTeamsTurnContext } from "../sdk-types.js"; -import { recordMSTeamsSentMessage } from "../sent-message-cache.js"; import { admitMSTeamsMessage } from "./access.js"; import { prepareMSTeamsInboundContent } from "./inbound-content.js"; +import { dispatchMSTeamsInboundTurn } from "./inbound-dispatch.js"; import { assembleMSTeamsInboundFacts, prepareMSTeamsDebounceEntry, @@ -95,7 +85,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { conversationType, isChannel, teamId, - graphChannelId, conversationRef, } = facts; const historyBody = [text, formatMediaPlaceholderText(advertisedMedia)] @@ -173,13 +162,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { return; } - const teamsFrom = isDirectMessage - ? `msteams:${senderId}` - : isChannel - ? `msteams:channel:${conversationId}` - : `msteams:group:${conversationId}`; - const teamsTo = isDirectMessage ? `user:${senderId}` : `conversation:${conversationId}`; - const threadRouting = prepareMSTeamsThreadRouting({ cfg, context, @@ -193,7 +175,6 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { }); const { route, deadline: preprocessingDeadline } = threadRouting; - const preview = sliceUtf16Safe(rawBody.replace(/\s+/g, " "), 0, 160); const inboundLabel = isDirectMessage ? `Teams DM from ${senderName}` : `Teams message in ${conversationType} from ${senderName}`; @@ -274,264 +255,44 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { if (!content) { return; } - const { agentBody, inboundMedia } = content; enqueuePrimaryMessageSystemEvent(); - const { teamAadGroupId, quoteBodyFull, quoteSenderId, quoteSenderName, threadContext } = - await resolveMSTeamsThreadContext({ - routing: threadRouting, - context, - tokenProvider, - quoteInfo, - isDirectMessage, - isChannel, - conversationId, - contextVisibilityMode, - groupPolicy, - effectiveGroupAllowFrom, - allowNameMatching, - log, - }); - - const envelopeFrom = isDirectMessage ? senderName : conversationType; - const buildEnvelope = createChannelInboundEnvelopeBuilder({ cfg, route }); - const body = buildEnvelope({ - channel: "Teams", - from: envelopeFrom, - timestamp, - body: agentBody, - }); - let combinedBody = body; - const isRoomish = !isDirectMessage; - const historyKey = isRoomish ? conversationId : undefined; - if (isRoomish && historyKey) { - const channelHistory = createChannelHistoryWindow({ historyMap: conversationHistories }); - combinedBody = channelHistory.buildPendingContext({ - historyKey, - limit: historyLimit, - currentMessage: combinedBody, - formatEntry: (entry) => - buildEnvelope({ - channel: "Teams", - from: conversationType, - timestamp: entry.timestamp, - previousTimestamp: null, - body: `${entry.sender}: ${entry.body}${entry.messageId ? ` [id:${entry.messageId}]` : ""}`, - }), - }); - } - - const inboundHistory = - isRoomish && historyKey && historyLimit > 0 - ? createChannelHistoryWindow({ historyMap: conversationHistories }).buildInboundHistory({ - historyKey, - limit: historyLimit, - }) - : undefined; - const commandBody = text.trim(); - const quoteSenderAllowed = - quoteInfo && quoteInfo.sender - ? resolveInboundSupplementalSenderAllowed({ - isGroup: !isDirectMessage, - groupPolicy, - allowFrom: effectiveGroupAllowFrom, - isSenderAllowed: (allowFrom) => - resolveMSTeamsAllowlistMatch({ - allowFrom, - senderId: quoteSenderId ?? "", - senderName: quoteSenderName, - allowNameMatching, - }).allowed, - }) - : true; - // Prepend thread history to the agent body so the agent has full thread context. - const bodyForAgent = threadContext - ? `[Thread history]\n${threadContext}\n[/Thread history]\n\n${agentBody}` - : agentBody; - - // For Teams *channel* messages (not group chats / DMs), preserve the - // `aadGroupId/channelId` pair on NativeChannelId so downstream action handlers - // can route through `/teams/{aadGroupId}/channels/{channelId}` via Graph API. - // The bare conversation id (`19:...@thread.tacv2`) is insufficient on its - // own because channel Graph endpoints require the owning team id too. - const nativeChannelId = - isChannel && teamAadGroupId ? `${teamAadGroupId}/${graphChannelId}` : undefined; - const ctxPayload = buildChannelInboundEventContext({ - channel: "msteams", - contextVisibility: contextVisibilityMode, - supplemental: { - quote: quoteInfo - ? { - id: quoteInfo.id ?? activity.replyToId ?? undefined, - body: quoteBodyFull ?? quoteInfo.body, - sender: quoteInfo.sender, - senderAllowed: quoteSenderAllowed, - isQuote: true, - } - : undefined, - }, - media: await toInboundMediaFactsWithMetadata(inboundMedia), - messageId: activity.id, - timestamp: timestamp?.getTime() ?? Date.now(), - from: teamsFrom, - sender: { - id: senderId, - name: senderName, - }, - conversation: { - kind: isDirectMessage ? "direct" : isChannel ? "channel" : "group", - id: conversationId, - label: envelopeFrom, - spaceId: teamId, - nativeChannelId, - }, - route: { - agentId: route.agentId, - dmScope: route.dmScope, - accountId: route.accountId, - routeSessionKey: route.sessionKey, - }, - reply: { - to: teamsTo, - replyToId: activity.replyToId ?? undefined, - nativeChannelId, - }, - message: { - body: combinedBody, - bodyForAgent, - inboundHistory, - rawBody, - commandBody, - }, - sessionTranscript: { historyLimit: isRoomish ? historyLimit : 0 }, - access: { - mentions: { - canDetectMention: !isDirectMessage, - wasMentioned: isDirectMessage || mentionDecision.effectiveWasMentioned, - }, - commands: { - authorized: commandAuthorized === true, - }, - }, - extra: { - GroupSubject: !isDirectMessage ? conversationType : undefined, - ReplyToIsQuote: quoteInfo ? true : undefined, - }, - }); - - logVerboseMessage(`msteams inbound: from=${ctxPayload.From} preview="${preview}"`); - - const sharePointSiteId = msteamsCfg?.sharePointSiteId; - const { dispatcherOptions, delivery, replyOptions } = createMSTeamsReplyDispatcher({ - cfg, - agentId: route.agentId, - sessionKey: route.sessionKey, - accountId: route.accountId, - runtime, - log, - app, - appId, - conversationRef, + const thread = await resolveMSTeamsThreadContext({ + routing: threadRouting, context, - replyStyle, - textLimit, - onSentMessageIds: (ids) => { - for (const id of ids) { - recordMSTeamsSentMessage(conversationId, id); - } - }, tokenProvider, - sharePointSiteId, + quoteInfo, + isDirectMessage, + isChannel, + conversationId, + contextVisibilityMode, + groupPolicy, + effectiveGroupAllowFrom, + allowNameMatching, + log, }); - // Use Teams clientInfo timezone if no explicit userTimezone is configured. - // This ensures the agent knows the sender's timezone for time-aware responses - // and proactive sends within the same session. - const activityClientInfo = activity.entities?.find((e) => e.type === "clientInfo") as - | { timezone?: string } - | undefined; - const senderTimezone = activityClientInfo?.timezone || conversationRef.timezone; - const turnConfig = - senderTimezone && !cfg.agents?.defaults?.userTimezone - ? { - ...cfg, - agents: { - ...cfg.agents, - defaults: { ...cfg.agents?.defaults, userTimezone: senderTimezone }, - }, - } - : cfg; - log.info("dispatching to agent", { sessionKey: route.sessionKey }); - try { - const turnResult = await core.channel.inbound.run({ - channel: "msteams", - accountId: route.accountId, - raw: context, - adapter: { - ingest: () => ({ - id: activity.id ?? `${teamsFrom}:${Date.now()}`, - timestamp: timestamp?.getTime(), - rawText: rawBody, - textForAgent: bodyForAgent, - textForCommands: commandBody, - raw: activity, - }), - resolveTurn: () => ({ - cfg: turnConfig, - channel: "msteams", - accountId: route.accountId, - route: { agentId: route.agentId, sessionKey: route.sessionKey }, - ctxPayload, - record: { - onRecordError: (err) => { - logVerboseMessage( - `msteams: failed updating session meta: ${formatUnknownError(err)}`, - ); - }, - }, - history: { - isGroup: isRoomish, - historyKey, - historyMap: conversationHistories, - limit: historyLimit, - }, - dispatcherOptions, - delivery, - replyOptions: { - ...replyOptions, - ...(params.turnAdoptionLifecycle - ? bindIngressLifecycleToReplyOptions(params.turnAdoptionLifecycle) - : {}), - }, - }), - }, - }); - const dispatchResult = turnResult.dispatched ? turnResult.dispatchResult : undefined; - const queuedFinal = dispatchResult?.queuedFinal ?? false; - const counts = resolveInboundReplyDispatchCounts(dispatchResult); - const hasFinalResponse = hasFinalInboundReplyDispatch(dispatchResult); - - log.info("dispatch complete", { queuedFinal, counts }); - - if (!hasFinalResponse) { - return; - } - const finalCount = counts.final; - logVerboseMessage( - `msteams: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${teamsTo}`, - ); - } catch (err) { - log.error("dispatch failed", { error: formatUnknownError(err) }); - runtime.error(`msteams dispatch failed: ${formatUnknownError(err)}`); - if (params.turnAdoptionLifecycle) { - throw err; - } - try { - await context.sendActivity("⚠️ Something went wrong. Please try again."); - } catch { - // Best effort. - } - } + await dispatchMSTeamsInboundTurn({ + cfg, + runtime, + appId, + app, + tokenProvider, + textLimit, + log, + logVerboseMessage, + facts, + admission, + content, + routing: threadRouting, + thread, + replyStyle, + timestamp, + contextVisibilityMode, + mentionWasEffective: mentionDecision.effectiveWasMentioned, + conversationHistories, + historyLimit, + }); }; const inboundDebouncer = core.channel.debounce.createInboundDebouncer({ @@ -623,4 +384,3 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { return undefined; }; } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ From dedb27d19c57adadc63a065fb616afc9ef55ab42 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 17:03:24 +0800 Subject: [PATCH 6/8] fix(msteams): keep dispatch result internal --- extensions/msteams/src/monitor-handler/inbound-dispatch.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/extensions/msteams/src/monitor-handler/inbound-dispatch.ts b/extensions/msteams/src/monitor-handler/inbound-dispatch.ts index 2ce335b36c49..26e5c5d9df97 100644 --- a/extensions/msteams/src/monitor-handler/inbound-dispatch.ts +++ b/extensions/msteams/src/monitor-handler/inbound-dispatch.ts @@ -22,7 +22,7 @@ import type { prepareMSTeamsInboundContent } from "./inbound-content.js"; import type { assembleMSTeamsInboundFacts } from "./inbound-facts.js"; import type { prepareMSTeamsThreadRouting, resolveMSTeamsThreadContext } from "./thread-context.js"; -export type MSTeamsInboundDispatchResult = +type MSTeamsInboundDispatchResult = | { kind: "completed"; finalResponses: number } | { kind: "failed" }; From d1c1aaaffe85b8c75ca7409f4467e807f2cde679 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 17:07:17 +0800 Subject: [PATCH 7/8] fix(msteams): type thread allowlist inputs --- extensions/msteams/src/monitor-handler/thread-context.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/extensions/msteams/src/monitor-handler/thread-context.ts b/extensions/msteams/src/monitor-handler/thread-context.ts index f58e14245cee..2b615c617c27 100644 --- a/extensions/msteams/src/monitor-handler/thread-context.ts +++ b/extensions/msteams/src/monitor-handler/thread-context.ts @@ -96,9 +96,7 @@ export async function resolveMSTeamsThreadContext(params: { conversationId: string; contextVisibilityMode: "all" | "allowlist" | "allowlist_quote"; groupPolicy: Parameters[0]["groupPolicy"]; - effectiveGroupAllowFrom: Parameters< - typeof resolveInboundSupplementalSenderAllowed - >[0]["allowFrom"]; + effectiveGroupAllowFrom: readonly (string | number)[]; allowNameMatching: boolean; log: MSTeamsMessageHandlerDeps["log"]; }) { From b6627f2000b733043ab5bd85b39d0245cebce15c Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 29 Jul 2026 17:49:58 +0800 Subject: [PATCH 8/8] chore(msteams): prune max-lines baseline --- config/max-lines-baseline.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 93854540b8ac..96e46021cd49 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -182,7 +182,6 @@ extensions/migrate-hermes/config.test.ts extensions/migrate-hermes/secrets.test.ts extensions/msteams/src/channel.actions.test.ts extensions/msteams/src/channel.ts -extensions/msteams/src/monitor-handler/message-handler.ts extensions/mxc/test/mxc-backend.test.ts extensions/oc-path/src/oc-path/find.ts extensions/oc-path/src/oc-path/universal.ts