From 0ff2c82033302fd3a969be5ea8612263bd4199cc Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 25 Jul 2026 10:56:58 -0700 Subject: [PATCH] refactor(mattermost): split inbound monitor (#113775) * refactor(mattermost): split inbound monitor * test(channels): follow mattermost history split --- config/max-lines-baseline.txt | 1 - .../src/mattermost/monitor-interactions.ts | 252 +++ .../src/mattermost/monitor-model-picker.ts | 359 ++++ .../src/mattermost/monitor-posts.ts | 468 ++++ .../src/mattermost/monitor-reactions.ts | 99 + .../mattermost/src/mattermost/monitor-turn.ts | 538 +++++ .../src/mattermost/monitor-types.ts | 27 + .../mattermost/src/mattermost/monitor.ts | 1889 +---------------- .../turn/message-turn-guardrails.test.ts | 4 +- 9 files changed, 1816 insertions(+), 1821 deletions(-) create mode 100644 extensions/mattermost/src/mattermost/monitor-interactions.ts create mode 100644 extensions/mattermost/src/mattermost/monitor-model-picker.ts create mode 100644 extensions/mattermost/src/mattermost/monitor-posts.ts create mode 100644 extensions/mattermost/src/mattermost/monitor-reactions.ts create mode 100644 extensions/mattermost/src/mattermost/monitor-turn.ts create mode 100644 extensions/mattermost/src/mattermost/monitor-types.ts diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 57c73223d8b7..fa34fd3a1282 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -151,7 +151,6 @@ extensions/matrix/src/onboarding.ts extensions/mattermost/src/channel.test.ts extensions/mattermost/src/channel.ts extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts -extensions/mattermost/src/mattermost/monitor.ts extensions/mattermost/src/mattermost/slash-http.ts extensions/memory-core/doctor-contract-api.test.ts extensions/memory-core/src/cli.runtime.ts diff --git a/extensions/mattermost/src/mattermost/monitor-interactions.ts b/extensions/mattermost/src/mattermost/monitor-interactions.ts new file mode 100644 index 000000000000..03a0ba13e470 --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-interactions.ts @@ -0,0 +1,252 @@ +// Mattermost plugin module registers interactive callback transport handling. +import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; +import { finalizeInboundContext } from "openclaw/plugin-sdk/reply-runtime"; +import { resolveMattermostReplyToMode } from "./accounts.js"; +import { createMattermostInteractionHandler } from "./interactions.js"; +import { + authorizeMattermostCommandInvocation, + mapMattermostChannelTypeToChatType, +} from "./monitor-auth.js"; +import { + resolveMattermostReplyRootId, + resolveMattermostThreadSessionContext, +} from "./monitor-context.js"; +import type { MattermostModelPickerInteractionHandler } from "./monitor-model-picker.js"; +import type { MattermostMonitorContext } from "./monitor-types.js"; +import { + createMattermostReplyDeliveryBarrier, + deliverMattermostReplyPayload, +} from "./reply-delivery.js"; +import type { ReplyPayload } from "./runtime-api.js"; +import { + createChannelMessageReplyPipeline, + logTypingFailure, + registerPluginHttpRoute, +} from "./runtime-api.js"; +import { sendMessageMattermost } from "./send.js"; + +export function registerMattermostInteractions(params: { + monitor: MattermostMonitorContext; + interactionPath: string; + allowedSourceIps: string[]; + handleModelPickerInteraction: MattermostModelPickerInteractionHandler; +}): (() => void) | undefined { + const { monitor } = params; + const { account, botUserId, cfg, client, core, pairing, resources, runtime } = monitor; + const { resolveChannelInfo, sendTypingIndicator } = resources; + return registerPluginHttpRoute({ + path: params.interactionPath, + fallbackPath: "/mattermost/interactions/default", + auth: "plugin", + handler: createMattermostInteractionHandler({ + client, + botUserId, + accountId: account.accountId, + allowedSourceIps: params.allowedSourceIps, + trustedProxies: cfg.gateway?.trustedProxies, + allowRealIpFallback: cfg.gateway?.allowRealIpFallback === true, + handleInteraction: params.handleModelPickerInteraction, + authorizeButtonClick: async ({ payload, post }) => { + const channelInfo = await resolveChannelInfo(payload.channel_id); + const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ + cfg, + surface: "mattermost", + }); + const decision = await authorizeMattermostCommandInvocation({ + account, + cfg, + senderId: payload.user_id, + senderName: payload.user_name ?? "", + channelId: payload.channel_id, + channelInfo, + readStoreAllowFrom: pairing.readAllowFromStore, + allowTextCommands, + hasControlCommand: false, + }); + if (decision.ok) { + return { ok: true }; + } + return { + ok: false, + response: { + update: { + message: post.message ?? "", + props: post.props ?? undefined, + }, + ephemeral_text: `OpenClaw ignored this action for ${decision.roomLabel}.`, + }, + }; + }, + resolveSessionKey: async ({ channelId, userId, post }) => { + const channelInfo = await resolveChannelInfo(channelId); + if (!channelInfo?.type) { + monitor.logVerboseMessage( + `mattermost: drop interaction session event (cannot resolve channel type for ${channelId})`, + ); + throw new Error("Mattermost channel type could not be resolved"); + } + const kind = mapMattermostChannelTypeToChatType(channelInfo.type); + const route = core.channel.routing.resolveAgentRoute({ + cfg, + channel: "mattermost", + accountId: account.accountId, + teamId: channelInfo.team_id ?? undefined, + peer: { + kind, + id: kind === "direct" ? userId : channelId, + }, + }); + return resolveMattermostThreadSessionContext({ + baseSessionKey: route.sessionKey, + kind, + postId: post.id || undefined, + replyToMode: resolveMattermostReplyToMode(account, kind), + threadRootId: post.root_id, + }).sessionKey; + }, + dispatchButtonClick: async (button) => { + const channelInfo = await resolveChannelInfo(button.channelId); + if (!channelInfo?.type) { + monitor.logVerboseMessage( + `mattermost: drop interaction dispatch (cannot resolve channel type for ${button.channelId})`, + ); + return; + } + const kind = mapMattermostChannelTypeToChatType(channelInfo.type); + const teamId = channelInfo.team_id ?? undefined; + const channelName = channelInfo.name ?? undefined; + const channelDisplay = channelInfo.display_name ?? channelName ?? button.channelId; + const route = core.channel.routing.resolveAgentRoute({ + cfg, + channel: "mattermost", + accountId: account.accountId, + teamId, + peer: { + kind, + id: kind === "direct" ? button.userId : button.channelId, + }, + }); + const threadContext = resolveMattermostThreadSessionContext({ + baseSessionKey: route.sessionKey, + kind, + postId: button.post.id || button.postId, + replyToMode: resolveMattermostReplyToMode(account, kind), + threadRootId: button.post.root_id, + }); + const to = kind === "direct" ? `user:${button.userId}` : `channel:${button.channelId}`; + const bodyText = `[Button click: user @${button.userName} selected "${button.actionName}"]`; + const ctxPayload = finalizeInboundContext({ + Body: bodyText, + BodyForAgent: bodyText, + RawBody: bodyText, + CommandBody: bodyText, + From: + kind === "direct" + ? `mattermost:${button.userId}` + : kind === "group" + ? `mattermost:group:${button.channelId}` + : `mattermost:channel:${button.channelId}`, + To: to, + SessionKey: threadContext.sessionKey, + DmScope: route.dmScope, + ParentSessionKey: threadContext.parentSessionKey, + AccountId: route.accountId, + ChatType: kind, + ConversationLabel: `mattermost:${button.userName}`, + GroupSubject: kind !== "direct" ? channelDisplay : undefined, + GroupChannel: channelName ? `#${channelName}` : undefined, + GroupSpace: teamId, + SenderName: button.userName, + SenderId: button.userId, + Provider: "mattermost" as const, + Surface: "mattermost" as const, + MessageSid: `interaction:${button.postId}:${button.actionId}`, + ReplyToId: threadContext.effectiveReplyToId, + MessageThreadId: threadContext.effectiveReplyToId, + WasMentioned: true, + CommandAuthorized: false, + OriginatingChannel: "mattermost" as const, + OriginatingTo: to, + }); + + const textLimit = core.channel.text.resolveTextChunkLimit( + cfg, + "mattermost", + account.accountId, + { fallbackLimit: account.textChunkLimit ?? 4000 }, + ); + const tableMode = core.channel.text.resolveMarkdownTableMode({ + cfg, + channel: "mattermost", + accountId: account.accountId, + }); + const { onModelSelected, typingCallbacks, ...replyPipeline } = + createChannelMessageReplyPipeline({ + cfg, + agentId: route.agentId, + channel: "mattermost", + accountId: account.accountId, + typing: { + start: () => sendTypingIndicator(button.channelId, threadContext.effectiveReplyToId), + onStartError: (err) => { + logTypingFailure({ + log: monitor.logDebugMessage, + channel: "mattermost", + target: button.channelId, + error: err, + }); + }, + }, + }); + const deliveryBarrier = createMattermostReplyDeliveryBarrier({ + isDirect: kind === "direct", + dmRetryOptions: account.config.dmChannelRetry, + }); + await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ + ctx: ctxPayload, + cfg, + dispatcherOptions: { + ...replyPipeline, + resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, + onDeliverySettled: deliveryBarrier.markDeliverySettled, + humanDelay: resolveHumanDelayConfig(cfg, route.agentId), + deliver: async (payload: ReplyPayload) => { + await deliverMattermostReplyPayload({ + core, + cfg, + payload, + to, + accountId: account.accountId, + agentId: route.agentId, + replyToId: resolveMattermostReplyRootId({ + kind, + threadRootId: threadContext.effectiveReplyToId, + replyToId: payload.replyToId, + }), + textLimit, + tableMode, + sendMessage: sendMessageMattermost, + onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, + }); + runtime.log?.(`delivered button-click reply to ${to}`); + }, + onError: (err, info) => { + runtime.error?.(`mattermost button-click ${info.kind} reply failed: ${String(err)}`); + }, + typingCallbacks, + }, + replyOptions: { + disableBlockStreaming: + typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined, + onModelSelected, + }, + }); + }, + log: (message) => runtime.log?.(message), + }), + pluginId: "mattermost", + source: "mattermost-interactions", + accountId: account.accountId, + log: (message: string) => runtime.log?.(message), + }); +} diff --git a/extensions/mattermost/src/mattermost/monitor-model-picker.ts b/extensions/mattermost/src/mattermost/monitor-model-picker.ts new file mode 100644 index 000000000000..75db41aa9d44 --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-model-picker.ts @@ -0,0 +1,359 @@ +// Mattermost plugin module owns native model-picker interactions. +import { finalizeInboundContext } from "openclaw/plugin-sdk/reply-runtime"; +import type { ResolvedAgentRoute } from "openclaw/plugin-sdk/routing"; +import { resolveMattermostReplyToMode } from "./accounts.js"; +import type { MattermostPost } from "./client.js"; +import type { MattermostInteractionResponse } from "./interactions.js"; +import { + buildMattermostAllowedModelRefs, + parseMattermostModelPickerContext, + renderMattermostModelsPickerView, + renderMattermostProviderPickerView, + resolveMattermostModelPickerCurrentModel, +} from "./model-picker.js"; +import { authorizeMattermostCommandInvocation } from "./monitor-auth.js"; +import { + buildMattermostModelPickerSelectMessageSid, + resolveMattermostReplyRootId, + resolveMattermostThreadSessionContext, +} from "./monitor-context.js"; +import type { MattermostMonitorContext } from "./monitor-types.js"; +import { + createMattermostReplyDeliveryBarrier, + deliverMattermostReplyPayload, +} from "./reply-delivery.js"; +import type { ChatType, ReplyPayload } from "./runtime-api.js"; +import { + buildModelsProviderData, + createChannelMessageReplyPipeline, + logTypingFailure, +} from "./runtime-api.js"; +import { sendMessageMattermost } from "./send.js"; + +type RunModelPickerCommandParams = { + commandText: string; + commandAuthorized: boolean; + route: ResolvedAgentRoute; + sessionKey: string; + parentSessionKey?: string; + channelId: string; + senderId: string; + senderName: string; + kind: ChatType; + channelName?: string; + channelDisplay?: string; + roomLabel: string; + teamId?: string; + messageSid: string; + effectiveReplyToId?: string; +}; + +export type MattermostModelPickerInteractionHandler = (params: { + payload: { + channel_id: string; + post_id: string; + team_id?: string; + user_id: string; + }; + userName: string; + context: Record; + post: MattermostPost; +}) => Promise; + +export function createMattermostModelPickerInteractionHandler( + monitor: MattermostMonitorContext, +): MattermostModelPickerInteractionHandler { + const { account, cfg, core, pairing, resources, runtime } = monitor; + const { resolveChannelInfo, sendTypingIndicator, updateModelPickerPost } = resources; + + const runModelPickerCommand = async (params: RunModelPickerCommandParams): Promise => { + const to = params.kind === "direct" ? `user:${params.senderId}` : `channel:${params.channelId}`; + const fromLabel = + params.kind === "direct" + ? `Mattermost DM from ${params.senderName}` + : `Mattermost message in ${params.roomLabel} from ${params.senderName}`; + const ctxPayload = finalizeInboundContext({ + Body: params.commandText, + BodyForAgent: params.commandText, + RawBody: params.commandText, + CommandBody: params.commandText, + From: + params.kind === "direct" + ? `mattermost:${params.senderId}` + : params.kind === "group" + ? `mattermost:group:${params.channelId}` + : `mattermost:channel:${params.channelId}`, + To: to, + SessionKey: params.sessionKey, + DmScope: params.route.dmScope, + ParentSessionKey: params.parentSessionKey, + AccountId: params.route.accountId, + ChatType: params.kind, + ConversationLabel: fromLabel, + GroupSubject: + params.kind !== "direct" ? params.channelDisplay || params.roomLabel : undefined, + GroupChannel: params.channelName ? `#${params.channelName}` : undefined, + GroupSpace: params.teamId, + SenderName: params.senderName, + SenderId: params.senderId, + Provider: "mattermost" as const, + Surface: "mattermost" as const, + MessageSid: params.messageSid, + ReplyToId: params.effectiveReplyToId, + MessageThreadId: params.effectiveReplyToId, + Timestamp: Date.now(), + WasMentioned: true, + CommandAuthorized: params.commandAuthorized, + CommandSource: "native" as const, + OriginatingChannel: "mattermost" as const, + OriginatingTo: to, + }); + + const tableMode = core.channel.text.resolveMarkdownTableMode({ + cfg, + channel: "mattermost", + accountId: account.accountId, + }); + const textLimit = core.channel.text.resolveTextChunkLimit( + cfg, + "mattermost", + account.accountId, + { fallbackLimit: account.textChunkLimit ?? 4000 }, + ); + const { onModelSelected, typingCallbacks, ...replyPipeline } = + createChannelMessageReplyPipeline({ + cfg, + agentId: params.route.agentId, + channel: "mattermost", + accountId: account.accountId, + typing: { + start: () => sendTypingIndicator(params.channelId, params.effectiveReplyToId), + onStartError: (err) => { + logTypingFailure({ + log: monitor.logDebugMessage, + channel: "mattermost", + target: params.channelId, + error: err, + }); + }, + }, + }); + const deliveryBarrier = createMattermostReplyDeliveryBarrier({ + isDirect: params.kind === "direct", + dmRetryOptions: account.config.dmChannelRetry, + }); + await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ + ctx: ctxPayload, + cfg, + dispatcherOptions: { + ...replyPipeline, + resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, + onDeliverySettled: deliveryBarrier.markDeliverySettled, + // Picker-triggered confirmations should stay immediate. + deliver: async (payload: ReplyPayload) => { + const trimmedPayload = { + ...payload, + text: core.channel.text.convertMarkdownTables(payload.text ?? "", tableMode).trim(), + }; + await deliverMattermostReplyPayload({ + core, + cfg, + payload: trimmedPayload, + to, + accountId: account.accountId, + agentId: params.route.agentId, + replyToId: resolveMattermostReplyRootId({ + kind: params.kind, + threadRootId: params.effectiveReplyToId, + replyToId: trimmedPayload.replyToId, + }), + textLimit, + // The picker path already converts and trims text before delivery. + tableMode: "off", + sendMessage: sendMessageMattermost, + onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, + }); + }, + onError: (err, info) => { + runtime.error?.(`mattermost model picker ${info.kind} reply failed: ${String(err)}`); + }, + typingCallbacks, + }, + replyOptions: { + disableBlockStreaming: + typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined, + onModelSelected, + }, + }); + }; + + return async (params) => { + const pickerState = parseMattermostModelPickerContext(params.context); + if (!pickerState) { + return null; + } + if (pickerState.ownerUserId !== params.payload.user_id) { + return { ephemeral_text: "Only the person who opened this picker can use it." }; + } + const updatePickerPost = (message: string, buttons?: Array) => + updateModelPickerPost({ + channelId: params.payload.channel_id, + postId: params.payload.post_id, + message, + buttons, + }); + + const channelInfo = await resolveChannelInfo(params.payload.channel_id); + const pickerCommandText = + pickerState.action === "select" + ? `/model ${pickerState.provider}/${pickerState.model}` + : pickerState.action === "list" + ? `/models ${pickerState.provider}` + : "/models"; + const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ + cfg, + surface: "mattermost", + }); + const auth = await authorizeMattermostCommandInvocation({ + account, + cfg, + senderId: params.payload.user_id, + senderName: params.userName, + channelId: params.payload.channel_id, + channelInfo, + readStoreAllowFrom: pairing.readAllowFromStore, + allowTextCommands, + hasControlCommand: core.channel.text.hasControlCommand(pickerCommandText, cfg), + }); + if (!auth.ok) { + if (auth.denyReason === "dm-pairing") { + const { code } = await pairing.upsertPairingRequest({ + id: params.payload.user_id, + meta: { name: params.userName }, + }); + return { + ephemeral_text: core.channel.pairing.buildPairingReply({ + channel: "mattermost", + idLine: `Your Mattermost user id: ${params.payload.user_id}`, + code, + }), + }; + } + const denyText = + auth.denyReason === "unknown-channel" + ? "Temporary error: unable to determine channel type. Please try again." + : auth.denyReason === "dm-disabled" + ? "This bot is not accepting direct messages." + : auth.denyReason === "channels-disabled" + ? "Model picker actions are disabled in channels." + : auth.denyReason === "channel-no-allowlist" + ? "Model picker actions are not configured for this channel." + : "Unauthorized."; + return { ephemeral_text: denyText }; + } + + const { channelDisplay, channelName, kind, roomLabel } = auth; + const teamId = auth.channelInfo.team_id ?? params.payload.team_id ?? undefined; + const route = core.channel.routing.resolveAgentRoute({ + cfg, + channel: "mattermost", + accountId: account.accountId, + teamId, + peer: { + kind, + id: kind === "direct" ? params.payload.user_id : params.payload.channel_id, + }, + }); + const threadContext = resolveMattermostThreadSessionContext({ + baseSessionKey: route.sessionKey, + kind, + postId: params.post.id || params.payload.post_id, + replyToMode: resolveMattermostReplyToMode(account, kind), + threadRootId: params.post.root_id, + }); + const modelSessionRoute = { agentId: route.agentId, sessionKey: threadContext.sessionKey }; + const data = await buildModelsProviderData(cfg, route.agentId); + if (data.providers.length === 0) { + return await updatePickerPost("No models available."); + } + + if (pickerState.action === "providers" || pickerState.action === "back") { + const currentModel = resolveMattermostModelPickerCurrentModel({ + cfg, + route: modelSessionRoute, + data, + }); + const view = renderMattermostProviderPickerView({ + ownerUserId: pickerState.ownerUserId, + data, + currentModel, + }); + return await updatePickerPost(view.text, view.buttons); + } + + if (pickerState.action === "list") { + const currentModel = resolveMattermostModelPickerCurrentModel({ + cfg, + route: modelSessionRoute, + data, + }); + const view = renderMattermostModelsPickerView({ + ownerUserId: pickerState.ownerUserId, + data, + provider: pickerState.provider, + page: pickerState.page, + currentModel, + }); + return await updatePickerPost(view.text, view.buttons); + } + + const targetModelRef = `${pickerState.provider}/${pickerState.model}`; + if (!buildMattermostAllowedModelRefs(data).has(targetModelRef)) { + return { ephemeral_text: `That model is no longer available: ${targetModelRef}` }; + } + + void (async () => { + try { + await runModelPickerCommand({ + commandText: `/model ${targetModelRef}`, + commandAuthorized: auth.commandAuthorized, + route, + sessionKey: threadContext.sessionKey, + parentSessionKey: threadContext.parentSessionKey, + channelId: params.payload.channel_id, + senderId: params.payload.user_id, + senderName: params.userName, + kind, + channelName: channelName || undefined, + channelDisplay: channelDisplay || channelName || params.payload.channel_id, + roomLabel, + teamId, + messageSid: buildMattermostModelPickerSelectMessageSid({ + postId: params.payload.post_id, + provider: pickerState.provider, + model: pickerState.model, + }), + effectiveReplyToId: threadContext.effectiveReplyToId, + }); + const currentModel = resolveMattermostModelPickerCurrentModel({ + cfg, + route: modelSessionRoute, + data, + readConsistency: "latest", + }); + const view = renderMattermostModelsPickerView({ + ownerUserId: pickerState.ownerUserId, + data, + provider: pickerState.provider, + page: pickerState.page, + currentModel, + }); + await updatePickerPost(view.text, view.buttons); + } catch (err) { + runtime.error?.(`mattermost model picker select failed: ${String(err)}`); + } + })(); + + return {}; + }; +} diff --git a/extensions/mattermost/src/mattermost/monitor-posts.ts b/extensions/mattermost/src/mattermost/monitor-posts.ts new file mode 100644 index 000000000000..f3f6c48aadfe --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-posts.ts @@ -0,0 +1,468 @@ +// Mattermost plugin module normalizes accepted posts into inbound turns. +import { + formatInboundEnvelope, + implicitMentionKindWhen, +} from "openclaw/plugin-sdk/channel-inbound"; +import { finalizeInboundContext } from "openclaw/plugin-sdk/reply-runtime"; +import { resolvePinnedMainDmOwnerFromAllowlist } from "openclaw/plugin-sdk/security-runtime"; +import { + normalizeLowercaseStringOrEmpty, + normalizeOptionalString, + normalizeTrimmedStringList, + uniqueStrings, +} from "openclaw/plugin-sdk/string-coerce-runtime"; +import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import { resolveMattermostReplyToMode } from "./accounts.js"; +import type { MattermostPost } from "./client.js"; +import { resolveMattermostInboundMentionDecision } from "./monitor-activation.js"; +import { + formatMattermostDirectMessageDropLog, + normalizeMattermostAllowEntry, + resolveMattermostMonitorInboundAccess, + resolveMattermostTrustedChatKind, +} from "./monitor-auth.js"; +import { + resolveMattermostPendingHistoryKey, + resolveMattermostThreadSessionContext, +} from "./monitor-context.js"; +import { + formatInboundFromLabel, + normalizeMention, + shouldDropEmptyMattermostBody, +} from "./monitor-helpers.js"; +import type { MattermostIngressLifecycle } from "./monitor-ingress.js"; +import { resolveOncharPrefixes, stripOncharPrefix } from "./monitor-onchar.js"; +import { + buildMattermostInboundMediaPayload, + formatMattermostInboundMediaText, + formatMattermostPendingMediaText, +} from "./monitor-resources.js"; +import { dispatchMattermostInboundTurn } from "./monitor-turn.js"; +import type { MattermostMonitorContext } from "./monitor-types.js"; +import type { MattermostEventPayload } from "./monitor-websocket.js"; +import { + createChannelHistoryWindow, + DEFAULT_GROUP_HISTORY_LIMIT, + logInboundDrop, + type HistoryEntry, +} from "./runtime-api.js"; +import { sendMessageMattermost } from "./send.js"; +import { hasMattermostThreadParticipationWithPersistence } from "./thread-participation.js"; + +export function createMattermostPostHandler(monitor: MattermostMonitorContext) { + const { account, botUserId, botUsername, cfg, core, groupPolicy, pairing, resources } = monitor; + const { resolveChannelInfo, resolveMattermostMedia, resolveUserInfo } = resources; + const channelHistories = new Map(); + const historyLimit = Math.max( + 0, + cfg.messages?.groupChat?.historyLimit ?? DEFAULT_GROUP_HISTORY_LIMIT, + ); + + return async ( + post: MattermostPost, + payload: MattermostEventPayload, + turnAdoptionLifecycle?: MattermostIngressLifecycle, + messageIds?: string[], + ) => { + const channelId = post.channel_id ?? payload.data?.channel_id ?? payload.broadcast?.channel_id; + if (!channelId) { + monitor.logVerboseMessage("mattermost: drop post (missing channel id)"); + return; + } + if (!post.id) { + monitor.logVerboseMessage("mattermost: drop post (missing message id)"); + return; + } + const allMessageIds = messageIds?.length ? messageIds : [post.id]; + const senderId = post.user_id ?? payload.broadcast?.user_id; + if (!senderId) { + monitor.logVerboseMessage("mattermost: drop post (missing sender id)"); + return; + } + if (senderId === botUserId) { + monitor.logVerboseMessage(`mattermost: drop post (self sender=${senderId})`); + return; + } + if (normalizeOptionalString(post.type) !== undefined) { + monitor.logVerboseMessage( + `mattermost: drop post (system post type=${post.type ?? "unknown"})`, + ); + return; + } + + const channelInfo = await resolveChannelInfo(channelId); + const channelType = + normalizeOptionalString(channelInfo?.type) ?? + normalizeOptionalString(payload.data?.channel_type); + if (!channelType) { + monitor.logVerboseMessage( + `mattermost: drop post (cannot resolve channel type for ${channelId})`, + ); + return; + } + const kind = resolveMattermostTrustedChatKind({ channelType }); + const senderName = + normalizeOptionalString(payload.data?.sender_name) ?? + normalizeOptionalString((await resolveUserInfo(senderId))?.username) ?? + senderId; + const rawPostText = typeof post.message === "string" ? post.message : ""; + const rawText = normalizeOptionalString(rawPostText) ?? ""; + const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ + cfg, + surface: "mattermost", + }); + const isControlCommand = + allowTextCommands && core.channel.commands.isControlCommandMessage(rawText, cfg); + const accessDecision = await resolveMattermostMonitorInboundAccess({ + account, + cfg, + senderId, + senderName, + channelId, + kind, + groupPolicy, + readStoreAllowFrom: pairing.readAllowFromStore, + allowTextCommands, + hasControlCommand: isControlCommand, + eventKind: "message", + mayPair: true, + }); + const commandAuthorized = accessDecision.commandAccess.authorized; + + if (accessDecision.ingress.decision !== "allow") { + if (kind === "direct") { + if (accessDecision.ingress.reasonCode === "dm_policy_disabled") { + monitor.logVerboseMessage(`mattermost: drop dm (dmPolicy=disabled sender=${senderId})`); + return; + } + if (accessDecision.ingress.decision === "pairing") { + const { code, created } = await pairing.upsertPairingRequest({ + id: senderId, + meta: { name: senderName }, + }); + monitor.logVerboseMessage( + `mattermost: pairing request sender=${senderId} created=${created}`, + ); + if (created) { + try { + await sendMessageMattermost( + `user:${senderId}`, + core.channel.pairing.buildPairingReply({ + channel: "mattermost", + idLine: `Your Mattermost user id: ${senderId}`, + code, + }), + { cfg, accountId: account.accountId }, + ); + monitor.statusSink?.({ lastOutboundAt: Date.now() }); + } catch (err) { + monitor.logVerboseMessage( + `mattermost: pairing reply failed for ${senderId}: ${String(err)}`, + ); + } + } + return; + } + monitor.logVerboseMessage( + formatMattermostDirectMessageDropLog({ + senderId, + dmPolicy: account.config.dmPolicy ?? "pairing", + reasonCode: accessDecision.senderAccess.reasonCode, + }), + ); + return; + } + if (accessDecision.ingress.reasonCode === "group_policy_disabled") { + monitor.logVerboseMessage("mattermost: drop group message (groupPolicy=disabled)"); + return; + } + if (accessDecision.ingress.reasonCode === "group_policy_empty_allowlist") { + monitor.logVerboseMessage("mattermost: drop group message (no group allowlist)"); + return; + } + if (accessDecision.ingress.reasonCode === "group_policy_not_allowlisted") { + monitor.logVerboseMessage( + `mattermost: drop group sender=${senderId} (not in groupAllowFrom)`, + ); + return; + } + monitor.logVerboseMessage( + `mattermost: drop group message (groupPolicy=${groupPolicy} reason=${accessDecision.senderAccess.reasonCode})`, + ); + return; + } + + if (kind !== "direct" && accessDecision.commandAccess.shouldBlockControlCommand) { + logInboundDrop({ + log: monitor.logVerboseMessage, + channel: "mattermost", + reason: "control command (unauthorized)", + target: senderId, + }); + return; + } + + const teamId = payload.data?.team_id ?? channelInfo?.team_id ?? undefined; + const channelName = payload.data?.channel_name ?? channelInfo?.name ?? ""; + const channelDisplay = + payload.data?.channel_display_name ?? channelInfo?.display_name ?? channelName; + const roomLabel = channelName ? `#${channelName}` : channelDisplay || `#${channelId}`; + const route = core.channel.routing.resolveAgentRoute({ + cfg, + channel: "mattermost", + accountId: account.accountId, + teamId, + peer: { + kind, + id: kind === "direct" ? senderId : channelId, + }, + }); + const threadContext = resolveMattermostThreadSessionContext({ + baseSessionKey: route.sessionKey, + kind, + postId: post.id, + replyToMode: resolveMattermostReplyToMode(account, kind), + threadRootId: normalizeOptionalString(post.root_id), + }); + const { effectiveReplyToId, sessionKey, parentSessionKey } = threadContext; + const historyKey = resolveMattermostPendingHistoryKey({ kind, sessionKey }); + const fileIds = uniqueStrings(normalizeTrimmedStringList(post.file_ids ?? [])); + const nativeMedia = fileIds.map(() => ({})); + const mentionRegexes = core.channel.mentions.buildMentionRegexes(cfg, route.agentId); + const wasMentioned = + kind !== "direct" && + ((botUsername + ? normalizeLowercaseStringOrEmpty(rawText).includes( + `@${normalizeLowercaseStringOrEmpty(botUsername)}`, + ) + : false) || + core.channel.mentions.matchesMentionPatterns(rawText, mentionRegexes)); + const pendingBody = formatMattermostPendingMediaText({ body: rawText, media: nativeMedia }); + const recordPendingHistory = () => { + const trimmed = pendingBody.trim(); + createChannelHistoryWindow({ historyMap: channelHistories }).record({ + limit: historyLimit, + historyKey: historyKey ?? "", + entry: + historyKey && trimmed + ? { + sender: senderName, + body: trimmed, + timestamp: typeof post.create_at === "number" ? post.create_at : undefined, + messageId: post.id, + } + : null, + }); + }; + + const oncharEnabled = account.chatmode === "onchar" && kind !== "direct"; + const oncharPrefixes = oncharEnabled ? resolveOncharPrefixes(account.oncharPrefixes) : []; + const oncharResult = oncharEnabled + ? stripOncharPrefix(rawText, oncharPrefixes) + : { triggered: false, stripped: rawText }; + const oncharTriggered = oncharResult.triggered; + const canDetectMention = Boolean(botUsername) || mentionRegexes.length > 0; + // Threads the bot already replied in auto-engage: follow-ups resume without + // a re-mention even under requireMention. Keyed by the thread root id. + const threadAlreadyEngaged = + kind !== "direct" && effectiveReplyToId + ? await hasMattermostThreadParticipationWithPersistence({ + accountId: account.accountId, + channelId, + threadRootId: effectiveReplyToId, + }) + : false; + const shouldRequireMention = + kind !== "direct" && + core.channel.groups.resolveRequireMention({ + cfg, + channel: "mattermost", + accountId: account.accountId, + groupId: channelId, + requireMentionOverride: account.requireMention, + }); + const mentionDecision = resolveMattermostInboundMentionDecision({ + cfg, + accountId: account.accountId, + kind, + requireMention: shouldRequireMention || oncharEnabled, + canDetectMention: canDetectMention || oncharEnabled, + wasMentioned: wasMentioned || oncharTriggered, + implicitMentionKinds: implicitMentionKindWhen("bot_thread_participant", threadAlreadyEngaged), + allowTextCommands, + hasControlCommand: isControlCommand, + commandAuthorized, + }); + const { shouldBypassMention } = mentionDecision; + + if ( + mentionDecision.shouldSkip && + oncharEnabled && + !oncharTriggered && + !wasMentioned && + !shouldBypassMention + ) { + monitor.logVerboseMessage( + `mattermost: drop group message (onchar not triggered channel=${channelId} sender=${senderId})`, + ); + recordPendingHistory(); + return; + } + if (mentionDecision.shouldSkip) { + monitor.logVerboseMessage( + `mattermost: drop group message (missing mention channel=${channelId} sender=${senderId} requireMention=${shouldRequireMention} bypass=${shouldBypassMention} canDetectMention=${canDetectMention})`, + ); + recordPendingHistory(); + return; + } + + const mediaList = await resolveMattermostMedia(fileIds); + const bodySource = oncharTriggered ? oncharResult.stripped : rawText; + const baseText = formatMattermostInboundMediaText({ + body: bodySource, + nativeMedia, + materializedMedia: mediaList, + }); + const bodyText = normalizeMention(baseText, botUsername); + if ( + mediaList.length === 0 && + shouldDropEmptyMattermostBody({ bodyText, rawText: rawPostText, botUsername }) + ) { + monitor.logVerboseMessage( + `mattermost: drop message (empty body after normalization channel=${channelId} sender=${senderId} wasMentioned=${wasMentioned})`, + ); + return; + } + // Mention-only turns need non-empty agent text; the shared reply runner rejects empty + // bodies before model invocation. The guard above ensures this fallback is a bot mention. + const bodyForAgent = bodyText || rawText.trim(); + core.channel.activity.record({ + channel: "mattermost", + accountId: account.accountId, + direction: "inbound", + }); + + const fromLabel = formatInboundFromLabel({ + isGroup: kind !== "direct", + groupLabel: channelDisplay || roomLabel, + groupId: channelId, + groupFallback: roomLabel || "Channel", + directLabel: senderName, + directId: senderId, + }); + const textWithId = `${bodyText}\n[mattermost message id: ${post.id} channel: ${channelId}]`; + const body = formatInboundEnvelope({ + channel: "Mattermost", + from: fromLabel, + timestamp: typeof post.create_at === "number" ? post.create_at : undefined, + body: textWithId, + chatType: kind, + sender: { name: senderName, id: senderId }, + }); + let combinedBody = body; + if (historyKey) { + const channelHistory = createChannelHistoryWindow({ historyMap: channelHistories }); + combinedBody = channelHistory.buildPendingContext({ + historyKey, + limit: historyLimit, + currentMessage: combinedBody, + formatEntry: (entry) => + formatInboundEnvelope({ + channel: "Mattermost", + from: fromLabel, + timestamp: entry.timestamp, + body: `${entry.body}${ + entry.messageId ? ` [id:${entry.messageId} channel:${channelId}]` : "" + }`, + chatType: kind, + senderLabel: entry.sender, + }), + }); + } + + const to = kind === "direct" ? `user:${senderId}` : `channel:${channelId}`; + const commandBody = rawText.trim(); + const inboundHistory = + historyKey && historyLimit > 0 + ? createChannelHistoryWindow({ historyMap: channelHistories }).buildInboundHistory({ + historyKey, + limit: historyLimit, + }) + : undefined; + const ctxPayload = finalizeInboundContext({ + Body: combinedBody, + BodyForAgent: bodyForAgent, + InboundHistory: inboundHistory, + RawBody: commandBody, + CommandBody: commandBody, + BodyForCommands: commandBody, + From: + kind === "direct" + ? `mattermost:${senderId}` + : kind === "group" + ? `mattermost:group:${channelId}` + : `mattermost:channel:${channelId}`, + To: to, + SessionKey: sessionKey, + DmScope: route.dmScope, + ParentSessionKey: parentSessionKey, + AccountId: route.accountId, + ChatType: kind, + ConversationLabel: fromLabel, + GroupSubject: kind !== "direct" ? channelDisplay || roomLabel : undefined, + GroupChannel: channelName ? `#${channelName}` : undefined, + GroupSpace: teamId, + SenderName: senderName, + SenderId: senderId, + Provider: "mattermost" as const, + Surface: "mattermost" as const, + MessageSid: post.id, + MessageSids: allMessageIds.length > 1 ? allMessageIds : undefined, + MessageSidFirst: allMessageIds.length > 1 ? allMessageIds[0] : undefined, + MessageSidLast: + allMessageIds.length > 1 ? allMessageIds[allMessageIds.length - 1] : undefined, + ReplyToId: effectiveReplyToId, + MessageThreadId: effectiveReplyToId, + Timestamp: typeof post.create_at === "number" ? post.create_at : undefined, + WasMentioned: kind !== "direct" ? mentionDecision.effectiveWasMentioned : undefined, + CommandAuthorized: commandAuthorized, + // Tag typed text-slash control commands (e.g. ` /new`, ` /reset` sent via the regular + // post path rather than Mattermost's native slash UI) so the explicit-command turn + // exception in source-reply-delivery-mode.ts surfaces their acknowledgements under + // message_tool_only delivery modes (e.g. Codex harness DMs). Mirrors iMessage #82642. + CommandSource: commandAuthorized && isControlCommand ? ("text" as const) : undefined, + OriginatingChannel: "mattermost" as const, + OriginatingTo: to, + ...buildMattermostInboundMediaPayload(mediaList), + }); + const pinnedMainDmOwner = + kind === "direct" + ? resolvePinnedMainDmOwnerFromAllowlist({ + dmScope: cfg.session?.dmScope, + allowFrom: account.config.allowFrom, + normalizeEntry: normalizeMattermostAllowEntry, + }) + : null; + const previewLine = truncateUtf16Safe(bodyText, 200).replace(/\n/g, "\\n"); + monitor.logVerboseMessage( + `mattermost inbound: from=${ctxPayload.From} len=${bodyText.length} preview="${previewLine}"`, + ); + + await dispatchMattermostInboundTurn(monitor, { + post, + rawText, + ctxPayload, + kind, + route, + channelId, + senderId, + to, + effectiveReplyToId, + historyKey, + historyLimit, + channelHistories, + pinnedMainDmOwner, + turnAdoptionLifecycle, + }); + }; +} diff --git a/extensions/mattermost/src/mattermost/monitor-reactions.ts b/extensions/mattermost/src/mattermost/monitor-reactions.ts new file mode 100644 index 000000000000..780bfb9e716a --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-reactions.ts @@ -0,0 +1,99 @@ +// Mattermost plugin module maps reaction transport events into system events. +import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + mapMattermostChannelTypeToChatType, + resolveMattermostMonitorInboundAccess, +} from "./monitor-auth.js"; +import { resolveMattermostReactionChannelId } from "./monitor-context.js"; +import type { MattermostMonitorContext } from "./monitor-types.js"; +import type { MattermostEventPayload } from "./monitor-websocket.js"; + +type MattermostReaction = { user_id?: string; post_id?: string; emoji_name?: string }; + +export function createMattermostReactionHandler(monitor: MattermostMonitorContext) { + const { account, botUserId, cfg, core, groupPolicy, pairing, resources } = monitor; + const { resolveChannelInfo, resolveUserInfo } = resources; + return async (payload: MattermostEventPayload) => { + const reactionData = payload.data?.reaction; + if (!reactionData) { + return; + } + let reaction: MattermostReaction | null = null; + if (typeof reactionData === "string") { + try { + reaction = JSON.parse(reactionData) as MattermostReaction; + } catch { + return; + } + } else if (typeof reactionData === "object") { + reaction = reactionData as MattermostReaction; + } + const userId = reaction?.user_id?.trim(); + const postId = reaction?.post_id?.trim(); + const emojiName = reaction?.emoji_name?.trim(); + if (!userId || !postId || !emojiName || userId === botUserId) { + return; + } + + const action = payload.event === "reaction_removed" ? "removed" : "added"; + const senderName = normalizeOptionalString((await resolveUserInfo(userId))?.username) ?? userId; + const channelId = resolveMattermostReactionChannelId(payload); + if (!channelId) { + // Without a channel id we cannot verify DM/group policies — drop to be safe. + monitor.logVerboseMessage( + "mattermost: drop reaction (no channel_id in broadcast, cannot enforce policy)", + ); + return; + } + const channelInfo = await resolveChannelInfo(channelId); + if (!channelInfo?.type) { + // Cannot determine channel type — drop to avoid policy bypass. + monitor.logVerboseMessage( + `mattermost: drop reaction (cannot resolve channel type for ${channelId})`, + ); + return; + } + const kind = mapMattermostChannelTypeToChatType(channelInfo.type); + const reactionAccess = await resolveMattermostMonitorInboundAccess({ + account, + cfg, + senderId: userId, + senderName, + channelId, + kind, + groupPolicy, + readStoreAllowFrom: pairing.readAllowFromStore, + allowTextCommands: false, + hasControlCommand: false, + eventKind: "reaction", + mayPair: false, + }); + if (reactionAccess.ingress.decision !== "allow") { + monitor.logVerboseMessage( + kind === "direct" + ? `mattermost: drop reaction (dmPolicy=${account.config.dmPolicy ?? "pairing"} sender=${userId} reason=${reactionAccess.senderAccess.reasonCode})` + : `mattermost: drop reaction (groupPolicy=${groupPolicy} sender=${userId} reason=${reactionAccess.senderAccess.reasonCode} channel=${channelId})`, + ); + return; + } + + const route = core.channel.routing.resolveAgentRoute({ + cfg, + channel: "mattermost", + accountId: account.accountId, + teamId: channelInfo.team_id ?? undefined, + peer: { + kind, + id: kind === "direct" ? userId : channelId, + }, + }); + const eventText = `Mattermost reaction ${action}: :${emojiName}: by @${senderName} on post ${postId} in channel ${channelId}`; + core.system.enqueueSystemEvent(eventText, { + sessionKey: route.sessionKey, + contextKey: `mattermost:reaction:${postId}:${emojiName}:${userId}:${action}`, + }); + monitor.logVerboseMessage( + `mattermost reaction: ${action} :${emojiName}: by ${senderName} on ${postId}`, + ); + }; +} diff --git a/extensions/mattermost/src/mattermost/monitor-turn.ts b/extensions/mattermost/src/mattermost/monitor-turn.ts new file mode 100644 index 000000000000..56e10b560b21 --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-turn.ts @@ -0,0 +1,538 @@ +// Mattermost plugin module owns one accepted message's reply turn and delivery. +import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; +import type { ChannelInboundTurnPlan } from "openclaw/plugin-sdk/channel-inbound"; +import { + bindIngressLifecycleToReplyOptions, + buildChannelProgressDraftLineForEntry, + createChannelProgressDraftCompositor, +} from "openclaw/plugin-sdk/channel-outbound"; +import type { finalizeInboundContext } from "openclaw/plugin-sdk/reply-runtime"; +import { + resolveInboundLastRouteSessionKey, + type ResolvedAgentRoute, +} from "openclaw/plugin-sdk/routing"; +import type { MattermostPost } from "./client.js"; +import { + createMattermostDraftPreviewBoundaryController, + createMattermostDraftStream, +} from "./draft-stream.js"; +import { normalizeMattermostAllowEntry } from "./monitor-auth.js"; +import { + formatMattermostFinalDeliveryOutcomeLog, + resolveMattermostReplyRootId, + shouldSuppressMattermostDefaultToolProgressMessages, + shouldUpdateMattermostDraftToolProgress, +} from "./monitor-context.js"; +import { + deliverMattermostReplyWithDraftPreview, + type MattermostDraftPreviewState, +} from "./monitor-draft-delivery.js"; +import type { MattermostIngressLifecycle } from "./monitor-ingress.js"; +import type { MattermostMonitorContext } from "./monitor-types.js"; +import { + createMattermostReplyDeliveryBarrier, + deliverMattermostReplyPayload, +} from "./reply-delivery.js"; +import type { ChatType, HistoryEntry, ReplyPayload } from "./runtime-api.js"; +import { createChannelMessageReplyPipeline, logTypingFailure } from "./runtime-api.js"; +import { sendMessageMattermost } from "./send.js"; +import { recordMattermostThreadParticipation } from "./thread-participation.js"; + +type MattermostInboundTurnParams = { + post: MattermostPost; + rawText: string; + ctxPayload: ReturnType; + kind: ChatType; + route: ResolvedAgentRoute; + channelId: string; + senderId: string; + to: string; + effectiveReplyToId?: string; + historyKey: string | null; + historyLimit: number; + channelHistories: Map; + pinnedMainDmOwner: string | null; + turnAdoptionLifecycle?: MattermostIngressLifecycle; +}; + +function createDisabledMattermostDraftStream(): ReturnType { + const noopAsync = async () => {}; + return { + update: () => {}, + updateAssistantText: () => {}, + flush: noopAsync, + postId: () => undefined, + clear: noopAsync, + discardPending: noopAsync, + seal: noopAsync, + stop: noopAsync, + forceNewMessage: noopAsync, + settleBoundaries: noopAsync, + resolveFinalText: (text) => ({ kind: "full", text }), + }; +} + +export async function dispatchMattermostInboundTurn( + monitor: MattermostMonitorContext, + params: MattermostInboundTurnParams, +): Promise { + const { account, cfg, client, core, runtime } = monitor; + const { sendTypingIndicator } = monitor.resources; + const { + channelHistories, + channelId, + ctxPayload, + effectiveReplyToId, + historyKey, + historyLimit, + kind, + pinnedMainDmOwner, + post, + rawText, + route, + senderId, + to, + turnAdoptionLifecycle, + } = params; + const textLimit = core.channel.text.resolveTextChunkLimit(cfg, "mattermost", account.accountId, { + fallbackLimit: account.textChunkLimit ?? 4000, + }); + const tableMode = core.channel.text.resolveMarkdownTableMode({ + cfg, + channel: "mattermost", + accountId: account.accountId, + }); + const chunkMode = core.channel.text.resolveChunkMode(cfg, "mattermost", account.accountId); + const { onModelSelected, typingCallbacks, resolveResponsePrefix, ...replyPipeline } = + createChannelMessageReplyPipeline({ + cfg, + agentId: route.agentId, + channel: "mattermost", + accountId: account.accountId, + typing: { + start: () => sendTypingIndicator(channelId, effectiveReplyToId), + onStartError: (err) => { + logTypingFailure({ + log: monitor.logDebugMessage, + channel: "mattermost", + target: channelId, + error: err, + }); + }, + }, + }); + const draftPreviewEnabled = account.streamingMode !== "off"; + const draftToolProgressEnabled = shouldUpdateMattermostDraftToolProgress(account); + const suppressDefaultToolProgressMessages = + shouldSuppressMattermostDefaultToolProgressMessages(account); + const draftStream = draftPreviewEnabled + ? createMattermostDraftStream({ + client, + channelId, + rootId: effectiveReplyToId, + throttleMs: 1200, + chunkText: (value) => + core.channel.text.chunkMarkdownTextWithMode( + core.channel.text.convertMarkdownTables(value, tableMode), + textLimit, + chunkMode, + ), + log: monitor.logVerboseMessage, + warn: monitor.logVerboseMessage, + }) + : createDisabledMattermostDraftStream(); + const previewBoundaryController = createMattermostDraftPreviewBoundaryController({ + enabled: draftPreviewEnabled && account.streamingMode === "block", + forceNewMessage: async () => { + await draftStream.forceNewMessage(); + }, + }); + let lastPartialText = ""; + let firstAssistantPreviewPrefix: string | undefined; + let firstAssistantPreviewPrefixPending = true; + let currentAssistantPreviewUsesPrefix = false; + let blockPreviewActivity: "none" | "reasoning" | "text" | "tool" = "none"; + let blockPreviewAssistantMessagePending = false; + const progressDraft = createChannelProgressDraftCompositor({ + entry: account.config, + mode: account.streamingMode, + active: draftPreviewEnabled, + seed: `${account.accountId}:${channelId}`, + update: async (previewText, options) => { + draftStream.update(previewText); + if (options?.flush) { + await draftStream.flush(); + } + }, + }); + const enterBlockPreviewActivity = (activity: "reasoning" | "text" | "tool") => { + if (account.streamingMode !== "block") { + return undefined; + } + const continuingToolActivity = activity === "tool" && blockPreviewActivity === "tool"; + const continuingTextActivity = + activity === "text" && + blockPreviewActivity === "text" && + !blockPreviewAssistantMessagePending; + const continuingReasoningActivity = + activity === "reasoning" && blockPreviewActivity === "reasoning"; + const continuesCurrentActivity = + continuingToolActivity || continuingTextActivity || continuingReasoningActivity; + // Reasoning placeholders are transient: a visible successor reuses them, while entering from durable text/tool rotates generations. + const startsNewGeneration = !continuesCurrentActivity && blockPreviewActivity !== "reasoning"; + if (startsNewGeneration) { + currentAssistantPreviewUsesPrefix = false; + } + const boundarySettled = startsNewGeneration + ? previewBoundaryController.noteBoundary() + : undefined; + // Message-start is only a candidate boundary: consecutive tools stay together, while the first visible text or reasoning starts a new block. + if (!continuesCurrentActivity) { + progressDraft.reset(); + } + blockPreviewActivity = activity; + blockPreviewAssistantMessagePending = false; + if (activity === "tool") { + lastPartialText = ""; + } + return boundarySettled; + }; + const previewState: MattermostDraftPreviewState = { finalizedViaPreviewPost: false }; + + const resolvePreviewFinalText = (text?: string) => { + if (typeof text !== "string") { + return undefined; + } + const resolution = draftStream.resolveFinalText(text); + const deliveryText = resolution.kind === "already-delivered" ? "" : resolution.text; + const formatted = core.channel.text.convertMarkdownTables(deliveryText, tableMode); + const chunks = core.channel.text.chunkMarkdownTextWithMode(formatted, textLimit, chunkMode); + if (!chunks.length && formatted) { + chunks.push(formatted); + } + if (chunks.length !== 1) { + return undefined; + } + const trimmed = chunks[0]?.trim(); + if (!trimmed) { + return undefined; + } + if ( + lastPartialText && + lastPartialText.startsWith(trimmed) && + trimmed.length < lastPartialText.length + ) { + return undefined; + } + return trimmed; + }; + + const updateDraftFromPartial = (text?: string) => { + const cleaned = text?.trim(); + if (!cleaned || cleaned === lastPartialText) { + return undefined; + } + if ( + lastPartialText && + lastPartialText.startsWith(cleaned) && + cleaned.length < lastPartialText.length + ) { + return undefined; + } + const boundarySettled = enterBlockPreviewActivity("text"); + lastPartialText = cleaned; + if (firstAssistantPreviewPrefixPending) { + firstAssistantPreviewPrefix = resolveResponsePrefix?.(); + firstAssistantPreviewPrefixPending = false; + currentAssistantPreviewUsesPrefix = Boolean(firstAssistantPreviewPrefix); + } + const previewText = + currentAssistantPreviewUsesPrefix && firstAssistantPreviewPrefix + ? cleaned.startsWith(firstAssistantPreviewPrefix) + ? cleaned + : `${firstAssistantPreviewPrefix} ${cleaned}` + : cleaned; + draftStream.updateAssistantText(previewText); + previewBoundaryController.noteUpdate(); + return boundarySettled; + }; + + const deliveryBarrier = createMattermostReplyDeliveryBarrier({ + isDirect: kind === "direct", + dmRetryOptions: account.config.dmChannelRetry, + }); + const dispatcherOptions: NonNullable = { + ...replyPipeline, + resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, + onDeliverySettled: deliveryBarrier.markDeliverySettled, + humanDelay: resolveHumanDelayConfig(cfg, route.agentId), + typingCallbacks, + }; + const delivery: ChannelInboundTurnPlan["delivery"] = { + deliver: async (payloadEntry: ReplyPayload, info) => { + if (info.kind === "final") { + await enterBlockPreviewActivity("text"); + // Final text uses only confirmed-visible generations, so join prior boundary work before deciding whether to edit in place. + await draftStream.settleBoundaries(); + progressDraft.markFinalReplyStarted(); + } + // A visible same-thread final can be a send or an in-place draft edit; either path records participation. + const markThreadParticipation = () => { + if (kind !== "direct" && effectiveReplyToId) { + recordMattermostThreadParticipation(account.accountId, channelId, effectiveReplyToId, { + agentId: route.agentId, + }); + } + }; + await deliverMattermostReplyWithDraftPreview({ + payload: payloadEntry, + info, + kind, + client, + draftStream, + effectiveReplyToId, + resolvePreviewFinalText, + previewState, + logVerboseMessage: monitor.logVerboseMessage, + recordThreadParticipation: markThreadParticipation, + deliverPayload: async (payloadToDeliver) => { + const finalTextResolution = + info.kind === "final" && + !payloadToDeliver.isError && + typeof payloadToDeliver.text === "string" + ? draftStream.resolveFinalText(payloadToDeliver.text) + : undefined; + const resolvedPayload = finalTextResolution + ? { + ...payloadToDeliver, + text: + finalTextResolution.kind === "already-delivered" ? "" : finalTextResolution.text, + } + : payloadToDeliver; + const outcome = await deliverMattermostReplyPayload({ + core, + cfg, + payload: resolvedPayload, + to, + accountId: account.accountId, + agentId: route.agentId, + replyToId: resolveMattermostReplyRootId({ + kind, + threadRootId: effectiveReplyToId, + replyToId: payloadToDeliver.replyToId, + }), + textLimit, + tableMode, + sendMessage: sendMessageMattermost, + onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, + }); + // Record only visible sends so reasoning-only, empty, or suppressed threads do not auto-engage later. + if (outcome === "text" || outcome === "media") { + markThreadParticipation(); + } else if (outcome === "empty" && finalTextResolution?.kind === "already-delivered") { + // The terminal payload confirms the already-published assistant block as + // the visible final reply even though this delivery has no remaining text. + markThreadParticipation(); + } + const deliveryLog = formatMattermostFinalDeliveryOutcomeLog({ + outcome, + payload: resolvedPayload, + to, + accountId: account.accountId, + agentId: route.agentId, + }); + if (deliveryLog) { + runtime.log?.(deliveryLog); + } + }, + }); + if (info.kind === "final") { + progressDraft.markFinalReplyDelivered(); + } + }, + onError: (err, info) => { + runtime.error?.(`mattermost ${info.kind} reply failed: ${String(err)}`); + }, + }; + const inboundLastRouteSessionKey = resolveInboundLastRouteSessionKey({ + route, + sessionKey: route.sessionKey, + }); + + try { + await core.channel.inbound.run({ + channel: "mattermost", + accountId: route.accountId, + raw: post, + adapter: { + ingest: () => ({ + id: post.id ?? `${to}:${Date.now()}`, + timestamp: post.create_at ?? undefined, + rawText, + textForAgent: ctxPayload.BodyForAgent, + textForCommands: ctxPayload.CommandBody, + raw: post, + }), + resolveTurn: () => ({ + cfg, + channel: "mattermost", + accountId: route.accountId, + route: { + agentId: route.agentId, + dmScope: route.dmScope, + sessionKey: route.sessionKey, + }, + ctxPayload, + record: { + updateLastRoute: + kind === "direct" + ? { + sessionKey: inboundLastRouteSessionKey, + channel: "mattermost", + to, + accountId: route.accountId, + mainDmOwnerPin: + inboundLastRouteSessionKey === route.mainSessionKey && pinnedMainDmOwner + ? { + ownerRecipient: pinnedMainDmOwner, + senderRecipient: normalizeMattermostAllowEntry(senderId), + onSkip: ({ ownerRecipient, senderRecipient }) => { + monitor.logVerboseMessage( + `mattermost: skip main-session last route for ${senderRecipient} (pinned owner ${ownerRecipient})`, + ); + }, + } + : undefined, + } + : undefined, + onRecordError: (err) => { + monitor.logVerboseMessage( + `mattermost: failed updating session meta id=${post.id ?? "unknown"}: ${String(err)}`, + ); + }, + }, + history: { + isGroup: Boolean(historyKey), + historyKey: historyKey ?? undefined, + historyMap: channelHistories, + limit: historyLimit, + }, + dispatcherOptions, + delivery, + replyOptions: { + ...(turnAdoptionLifecycle + ? bindIngressLifecycleToReplyOptions(turnAdoptionLifecycle) + : {}), + allowProgressCallbacksWhenSourceDeliverySuppressed: draftToolProgressEnabled + ? true + : undefined, + preserveProgressCallbackStartOrder: draftPreviewEnabled ? true : undefined, + onObservedReplyDelivery: draftToolProgressEnabled + ? () => draftStream.clear() + : undefined, + disableBlockStreaming: draftPreviewEnabled + ? true + : typeof account.blockStreaming === "boolean" + ? !account.blockStreaming + : undefined, + ...(suppressDefaultToolProgressMessages + ? { suppressDefaultToolProgressMessages: true } + : {}), + onModelSelected, + onPartialReply: (payloadResult) => + account.streamingMode === "progress" + ? undefined + : updateDraftFromPartial(payloadResult.text), + onAssistantMessageStart: () => { + lastPartialText = ""; + progressDraft.resetReasoningProgress(); + if (account.streamingMode === "block") { + blockPreviewAssistantMessagePending = true; + return; + } + if (account.streamingMode !== "progress") { + progressDraft.reset(); + } + }, + onReasoningEnd: () => { + // Hidden reasoning has no boundary; only rendered text, reasoning, or tools rotate preview posts. + lastPartialText = ""; + progressDraft.resetReasoningProgress(); + if (account.streamingMode !== "block" && account.streamingMode !== "progress") { + progressDraft.reset(); + } + }, + onReasoningStream: async (payloadResult) => { + if (account.streamingMode === "progress") { + await progressDraft.pushReasoningProgress(payloadResult.text || "Thinking…", { + snapshot: payloadResult.isReasoningSnapshot === true, + }); + return; + } + if (!lastPartialText) { + const boundarySettled = enterBlockPreviewActivity("reasoning"); + draftStream.update("Thinking…"); + previewBoundaryController.noteUpdate(); + await boundarySettled; + } + }, + onToolStart: async (payloadValue) => { + if (!draftToolProgressEnabled) { + return; + } + const boundarySettled = enterBlockPreviewActivity("tool"); + // Boundary detach and progress staging both happen synchronously before + // their first await; agent callbacks may be dispatched fire-and-forget. + const progressSettled = progressDraft.pushToolProgress( + buildChannelProgressDraftLineForEntry( + account.config, + { + event: "tool", + itemId: payloadValue.itemId, + toolCallId: payloadValue.toolCallId, + name: payloadValue.name, + phase: payloadValue.phase, + args: payloadValue.args, + }, + payloadValue.detailMode ? { detailMode: payloadValue.detailMode } : undefined, + ), + { startImmediately: true }, + ); + previewBoundaryController.noteUpdate(); + await Promise.all([boundarySettled, progressSettled]); + }, + onItemEvent: async (payloadLocal) => { + if (!draftToolProgressEnabled) { + return; + } + const boundarySettled = enterBlockPreviewActivity("tool"); + const progressSettled = progressDraft.pushToolProgress( + buildChannelProgressDraftLineForEntry(account.config, { + event: "item", + itemId: payloadLocal.itemId, + itemKind: payloadLocal.kind, + title: payloadLocal.title, + name: payloadLocal.name, + phase: payloadLocal.phase, + status: payloadLocal.status, + summary: payloadLocal.summary, + progressText: payloadLocal.progressText, + meta: payloadLocal.meta, + }), + { startImmediately: true }, + ); + previewBoundaryController.noteUpdate(); + await Promise.all([boundarySettled, progressSettled]); + }, + }, + }), + }, + }); + } finally { + try { + await draftStream.stop(); + } catch (err) { + monitor.logVerboseMessage(`mattermost draft preview cleanup failed: ${String(err)}`); + } + } +} diff --git a/extensions/mattermost/src/mattermost/monitor-types.ts b/extensions/mattermost/src/mattermost/monitor-types.ts new file mode 100644 index 000000000000..8d7bb578240e --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-types.ts @@ -0,0 +1,27 @@ +// Mattermost plugin module shares monitor-scoped runtime dependencies. +import type { getMattermostRuntime } from "../runtime.js"; +import type { ResolvedMattermostAccount } from "./accounts.js"; +import type { MattermostClient } from "./client.js"; +import type { createMattermostMonitorResources } from "./monitor-resources.js"; +import type { + ChannelAccountSnapshot, + createChannelPairingController, + OpenClawConfig, + RuntimeEnv, +} from "./runtime-api.js"; + +export type MattermostMonitorContext = { + core: ReturnType; + runtime: RuntimeEnv; + cfg: OpenClawConfig; + account: ResolvedMattermostAccount; + client: MattermostClient; + pairing: ReturnType; + botUserId: string; + botUsername?: string; + groupPolicy: "allowlist" | "open" | "disabled"; + resources: ReturnType; + logDebugMessage: (message: string) => void; + logVerboseMessage: (message: string) => void; + statusSink?: (patch: Partial) => void; +}; diff --git a/extensions/mattermost/src/mattermost/monitor.ts b/extensions/mattermost/src/mattermost/monitor.ts index f2c44c43bf88..7683c1001907 100644 --- a/extensions/mattermost/src/mattermost/monitor.ts +++ b/extensions/mattermost/src/mattermost/monitor.ts @@ -1,30 +1,13 @@ -// Mattermost plugin module implements monitor behavior. -import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; -import { - formatInboundEnvelope, - implicitMentionKindWhen, - type ChannelInboundTurnPlan, -} from "openclaw/plugin-sdk/channel-inbound"; +// Mattermost plugin module orchestrates monitor setup, ingress, and teardown. import { fanInChannelIngressLifecycles } from "openclaw/plugin-sdk/channel-ingress-runtime"; -import { - bindIngressLifecycleToReplyOptions, - buildChannelProgressDraftLineForEntry, - createChannelProgressDraftCompositor, -} from "openclaw/plugin-sdk/channel-outbound"; import { isLoopbackHost } from "openclaw/plugin-sdk/gateway-runtime"; -import { finalizeInboundContext } from "openclaw/plugin-sdk/reply-runtime"; -import { resolveInboundLastRouteSessionKey } from "openclaw/plugin-sdk/routing"; -import { resolvePinnedMainDmOwnerFromAllowlist } from "openclaw/plugin-sdk/security-runtime"; import { isPrivateNetworkOptInEnabled } from "openclaw/plugin-sdk/ssrf-runtime"; import { - normalizeLowercaseStringOrEmpty, normalizeOptionalString, normalizeTrimmedStringList, - uniqueStrings, } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { getMattermostRuntime } from "../runtime.js"; -import { resolveMattermostAccount, resolveMattermostReplyToMode } from "./accounts.js"; +import { resolveMattermostAccount } from "./accounts.js"; import { createMattermostClient, fetchMattermostMe, @@ -32,104 +15,39 @@ import { type MattermostPost, type MattermostUser, } from "./client.js"; -import { - createMattermostDraftPreviewBoundaryController, - createMattermostDraftStream, -} from "./draft-stream.js"; import { computeInteractionCallbackUrl, - createMattermostInteractionHandler, resolveInteractionCallbackPath, setInteractionCallbackUrl, setInteractionSecret, - type MattermostInteractionResponse, } from "./interactions.js"; -import { - buildMattermostAllowedModelRefs, - parseMattermostModelPickerContext, - renderMattermostModelsPickerView, - renderMattermostProviderPickerView, - resolveMattermostModelPickerCurrentModel, -} from "./model-picker.js"; -import { resolveMattermostInboundMentionDecision } from "./monitor-activation.js"; -import { - authorizeMattermostCommandInvocation, - formatMattermostDirectMessageDropLog, - mapMattermostChannelTypeToChatType, - normalizeMattermostAllowEntry, - resolveMattermostTrustedChatKind, - resolveMattermostMonitorInboundAccess, -} from "./monitor-auth.js"; -import { - buildMattermostModelPickerSelectMessageSid, - formatMattermostFinalDeliveryOutcomeLog, - resolveMattermostPendingHistoryKey, - resolveMattermostReactionChannelId, - resolveMattermostReplyRootId, - resolveMattermostThreadSessionContext, - shouldSuppressMattermostDefaultToolProgressMessages, - shouldUpdateMattermostDraftToolProgress, -} from "./monitor-context.js"; -import { - deliverMattermostReplyWithDraftPreview, - type MattermostDraftPreviewState, -} from "./monitor-draft-delivery.js"; -import { - formatInboundFromLabel, - normalizeMention, - shouldDropEmptyMattermostBody, -} from "./monitor-helpers.js"; import { createMattermostIngressMonitor, type MattermostIngressLifecycle, } from "./monitor-ingress.js"; -import { resolveOncharPrefixes, stripOncharPrefix } from "./monitor-onchar.js"; -import { - buildMattermostInboundMediaPayload, - createMattermostMonitorResources, - formatMattermostInboundMediaText, - formatMattermostPendingMediaText, -} from "./monitor-resources.js"; +import { registerMattermostInteractions } from "./monitor-interactions.js"; +import { createMattermostModelPickerInteractionHandler } from "./monitor-model-picker.js"; +import { createMattermostPostHandler } from "./monitor-posts.js"; +import { createMattermostReactionHandler } from "./monitor-reactions.js"; +import { createMattermostMonitorResources } from "./monitor-resources.js"; import { registerMattermostMonitorSlashCommands } from "./monitor-slash.js"; +import type { MattermostMonitorContext } from "./monitor-types.js"; import { createMattermostConnectOnce, type MattermostEventPayload, type MattermostWebSocketFactory, } from "./monitor-websocket.js"; import { runWithReconnect } from "./reconnect.js"; +import type { ChannelAccountSnapshot, OpenClawConfig, RuntimeEnv } from "./runtime-api.js"; import { - createMattermostReplyDeliveryBarrier, - deliverMattermostReplyPayload, -} from "./reply-delivery.js"; -import type { - ChannelAccountSnapshot, - ChatType, - OpenClawConfig, - ReplyPayload, - RuntimeEnv, -} from "./runtime-api.js"; -import { - buildModelsProviderData, - createChannelHistoryWindow, createChannelPairingController, - createChannelMessageReplyPipeline, - DEFAULT_GROUP_HISTORY_LIMIT, - logInboundDrop, - logTypingFailure, - registerPluginHttpRoute, resolveAllowlistProviderRuntimeGroupPolicy, resolveChannelMediaMaxBytes, resolveDefaultGroupPolicy, warnMissingProviderGroupPolicyFallbackOnce, - type HistoryEntry, } from "./runtime-api.js"; -import { sendMessageMattermost } from "./send.js"; import { cleanupSlashCommands } from "./slash-commands.js"; import { deactivateSlashCommands, getSlashCommandState } from "./slash-state.js"; -import { - hasMattermostThreadParticipationWithPersistence, - recordMattermostThreadParticipation, -} from "./thread-participation.js"; type MonitorMattermostOpts = { botToken?: string; @@ -142,76 +60,19 @@ type MonitorMattermostOpts = { webSocketFactory?: MattermostWebSocketFactory; }; -type MattermostReaction = { - user_id?: string; - post_id?: string; - emoji_name?: string; - create_at?: number; -}; -function normalizeInteractionSourceIps(values?: string[]): string[] { - return normalizeTrimmedStringList(values); -} - -function resolveRuntime(opts: MonitorMattermostOpts): RuntimeEnv { - return ( - opts.runtime ?? { +export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {}): Promise { + const core = getMattermostRuntime(); + const runtime = + opts.runtime ?? + ({ log: console.log, error: console.error, exit: (code: number): never => { throw new Error(`exit ${code}`); }, - } - ); -} - -function isSystemPost(post: MattermostPost): boolean { - return normalizeOptionalString(post.type) !== undefined; -} - -function channelChatType(kind: ChatType): "direct" | "group" | "channel" { - if (kind === "direct") { - return "direct"; - } - if (kind === "group") { - return "group"; - } - return "channel"; -} - -function createDisabledMattermostDraftStream(): ReturnType { - const noopAsync = async () => {}; - return { - update: () => {}, - updateAssistantText: () => {}, - flush: noopAsync, - postId: () => undefined, - clear: noopAsync, - discardPending: noopAsync, - seal: noopAsync, - stop: noopAsync, - forceNewMessage: noopAsync, - settleBoundaries: noopAsync, - resolveFinalText: (text) => ({ kind: "full", text }), - }; -} - -function buildMattermostWsUrl(baseUrl: string): string { - const normalized = normalizeMattermostBaseUrl(baseUrl); - if (!normalized) { - throw new Error("Mattermost baseUrl is required"); - } - const wsBase = normalized.replace(/^http/i, "ws"); - return `${wsBase}/api/v4/websocket`; -} - -export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {}): Promise { - const core = getMattermostRuntime(); - const runtime = resolveRuntime(opts); + } satisfies RuntimeEnv); const cfg = (opts.config ?? core.config.current()) as OpenClawConfig; - const account = resolveMattermostAccount({ - cfg, - accountId: opts.accountId, - }); + const account = resolveMattermostAccount({ cfg, accountId: opts.accountId }); const pairing = createChannelPairingController({ core, channel: "mattermost", @@ -230,7 +91,6 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} `Mattermost baseUrl missing for account "${account.accountId}" (set channels.mattermost.accounts.${account.accountId}.baseUrl or MATTERMOST_URL for default).`, ); } - const client = createMattermostClient({ baseUrl, botToken, @@ -239,9 +99,8 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} // Wait for the Mattermost API to accept our bot token before proceeding. // When a bot account is disabled and re-enabled, the session is invalidated - // and API calls return 401 until the account is fully active again. Retrying - // here (with exponential backoff) keeps the monitor alive and prevents the - // framework's auto-restart budget from being exhausted. + // and API calls return 401 until the account is fully active again. Retrying + // here keeps the monitor alive instead of exhausting the auto-restart budget. let botUser!: MattermostUser; await runWithReconnect( async () => { @@ -276,24 +135,18 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} }); const slashEnabled = getSlashCommandState(account.accountId) != null; - // ─── Interactive buttons registration ────────────────────────────────────── - // Derive a stable HMAC secret from the bot token so CLI and gateway share it. + // Derive a stable HMAC secret so CLI and gateway validate the same callbacks. setInteractionSecret(account.accountId, botToken); - - // Register HTTP callback endpoint for interactive button clicks. - // Mattermost POSTs to this URL when a user clicks a button action. const interactionPath = resolveInteractionCallbackPath(account.accountId); - // Recompute from config on each monitor start so reconnects or config reloads can refresh the - // cached callback URL for downstream callers such as `message action=send`. + // Refresh the cached URL on every monitor start for reconnects and config reloads. const callbackUrl = computeInteractionCallbackUrl(account.accountId, { gateway: cfg.gateway, interactions: account.config.interactions, }); setInteractionCallbackUrl(account.accountId, callbackUrl); - const allowedInteractionSourceIps = normalizeInteractionSourceIps( + const allowedInteractionSourceIps = normalizeTrimmedStringList( account.config.interactions?.allowedSourceIps, ); - try { const mmHost = new URL(baseUrl).hostname; const callbackHost = new URL(callbackUrl).hostname; @@ -308,240 +161,14 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} ); } } catch { - // URL parse failed; ignore and continue (we will fail naturally if callbacks cannot be delivered). + // Invalid callback URLs fail naturally if Mattermost tries to deliver to them. } - const effectiveInteractionSourceIps = - allowedInteractionSourceIps.length > 0 ? allowedInteractionSourceIps : ["127.0.0.1", "::1"]; - - const unregisterInteractions = registerPluginHttpRoute({ - path: interactionPath, - fallbackPath: "/mattermost/interactions/default", - auth: "plugin", - handler: createMattermostInteractionHandler({ - client, - botUserId, - accountId: account.accountId, - allowedSourceIps: effectiveInteractionSourceIps, - trustedProxies: cfg.gateway?.trustedProxies, - allowRealIpFallback: cfg.gateway?.allowRealIpFallback === true, - handleInteraction: handleModelPickerInteraction, - authorizeButtonClick: async ({ payload, post }) => { - const channelInfo = await resolveChannelInfo(payload.channel_id); - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ - cfg, - surface: "mattermost", - }); - const decision = await authorizeMattermostCommandInvocation({ - account, - cfg, - senderId: payload.user_id, - senderName: payload.user_name ?? "", - channelId: payload.channel_id, - channelInfo, - readStoreAllowFrom: pairing.readAllowFromStore, - allowTextCommands, - hasControlCommand: false, - }); - if (decision.ok) { - return { ok: true }; - } - return { - ok: false, - response: { - update: { - message: post.message ?? "", - props: post.props ?? undefined, - }, - ephemeral_text: `OpenClaw ignored this action for ${decision.roomLabel}.`, - }, - }; - }, - resolveSessionKey: async ({ channelId, userId, post }) => { - const channelInfo = await resolveChannelInfo(channelId); - if (!channelInfo?.type) { - logVerboseMessage( - `mattermost: drop interaction session event (cannot resolve channel type for ${channelId})`, - ); - throw new Error("Mattermost channel type could not be resolved"); - } - const kind = mapMattermostChannelTypeToChatType(channelInfo.type); - const teamId = channelInfo?.team_id ?? undefined; - const route = core.channel.routing.resolveAgentRoute({ - cfg, - channel: "mattermost", - accountId: account.accountId, - teamId, - peer: { - kind, - id: kind === "direct" ? userId : channelId, - }, - }); - const replyToMode = resolveMattermostReplyToMode(account, kind); - return resolveMattermostThreadSessionContext({ - baseSessionKey: route.sessionKey, - kind, - postId: post.id || undefined, - replyToMode, - threadRootId: post.root_id, - }).sessionKey; - }, - dispatchButtonClick: async (optsLocal) => { - const channelInfo = await resolveChannelInfo(optsLocal.channelId); - if (!channelInfo?.type) { - logVerboseMessage( - `mattermost: drop interaction dispatch (cannot resolve channel type for ${optsLocal.channelId})`, - ); - return; - } - const kind = mapMattermostChannelTypeToChatType(channelInfo.type); - const chatType = channelChatType(kind); - const teamId = channelInfo?.team_id ?? undefined; - const channelName = channelInfo?.name ?? undefined; - const channelDisplay = channelInfo?.display_name ?? channelName ?? optsLocal.channelId; - const route = core.channel.routing.resolveAgentRoute({ - cfg, - channel: "mattermost", - accountId: account.accountId, - teamId, - peer: { - kind, - id: kind === "direct" ? optsLocal.userId : optsLocal.channelId, - }, - }); - const replyToMode = resolveMattermostReplyToMode(account, kind); - const threadContext = resolveMattermostThreadSessionContext({ - baseSessionKey: route.sessionKey, - kind, - postId: optsLocal.post.id || optsLocal.postId, - replyToMode, - threadRootId: optsLocal.post.root_id, - }); - const to = - kind === "direct" ? `user:${optsLocal.userId}` : `channel:${optsLocal.channelId}`; - const bodyText = `[Button click: user @${optsLocal.userName} selected "${optsLocal.actionName}"]`; - const ctxPayload = finalizeInboundContext({ - Body: bodyText, - BodyForAgent: bodyText, - RawBody: bodyText, - CommandBody: bodyText, - From: - kind === "direct" - ? `mattermost:${optsLocal.userId}` - : kind === "group" - ? `mattermost:group:${optsLocal.channelId}` - : `mattermost:channel:${optsLocal.channelId}`, - To: to, - SessionKey: threadContext.sessionKey, - DmScope: route.dmScope, - ParentSessionKey: threadContext.parentSessionKey, - AccountId: route.accountId, - ChatType: chatType, - ConversationLabel: `mattermost:${optsLocal.userName}`, - GroupSubject: kind !== "direct" ? channelDisplay : undefined, - GroupChannel: channelName ? `#${channelName}` : undefined, - GroupSpace: teamId, - SenderName: optsLocal.userName, - SenderId: optsLocal.userId, - Provider: "mattermost" as const, - Surface: "mattermost" as const, - MessageSid: `interaction:${optsLocal.postId}:${optsLocal.actionId}`, - ReplyToId: threadContext.effectiveReplyToId, - MessageThreadId: threadContext.effectiveReplyToId, - WasMentioned: true, - CommandAuthorized: false, - OriginatingChannel: "mattermost" as const, - OriginatingTo: to, - }); - - const textLimit = core.channel.text.resolveTextChunkLimit( - cfg, - "mattermost", - account.accountId, - { fallbackLimit: account.textChunkLimit ?? 4000 }, - ); - const tableMode = core.channel.text.resolveMarkdownTableMode({ - cfg, - channel: "mattermost", - accountId: account.accountId, - }); - const { onModelSelected, typingCallbacks, ...replyPipeline } = - createChannelMessageReplyPipeline({ - cfg, - agentId: route.agentId, - channel: "mattermost", - accountId: account.accountId, - typing: { - start: () => - sendTypingIndicator(optsLocal.channelId, threadContext.effectiveReplyToId), - onStartError: (err) => { - logTypingFailure({ - log: (message) => logger.debug?.(message), - channel: "mattermost", - target: optsLocal.channelId, - error: err, - }); - }, - }, - }); - const deliveryBarrier = createMattermostReplyDeliveryBarrier({ - isDirect: kind === "direct", - dmRetryOptions: account.config.dmChannelRetry, - }); - await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ - ctx: ctxPayload, - cfg, - dispatcherOptions: { - ...replyPipeline, - resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, - onDeliverySettled: deliveryBarrier.markDeliverySettled, - humanDelay: resolveHumanDelayConfig(cfg, route.agentId), - deliver: async (payload: ReplyPayload) => { - await deliverMattermostReplyPayload({ - core, - cfg, - payload, - to, - accountId: account.accountId, - agentId: route.agentId, - replyToId: resolveMattermostReplyRootId({ - kind, - threadRootId: threadContext.effectiveReplyToId, - replyToId: payload.replyToId, - }), - textLimit, - tableMode, - sendMessage: sendMessageMattermost, - onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, - }); - runtime.log?.(`delivered button-click reply to ${to}`); - }, - onError: (err, info) => { - runtime.error?.(`mattermost button-click ${info.kind} reply failed: ${String(err)}`); - }, - typingCallbacks, - }, - replyOptions: { - disableBlockStreaming: - typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined, - onModelSelected, - }, - }); - }, - log: (msg) => runtime.log?.(msg), - }), - pluginId: "mattermost", - source: "mattermost-interactions", - accountId: account.accountId, - log: (msg: string) => runtime.log?.(msg), - }); - const logger = core.logging.getChildLogger({ module: "mattermost" }); const logVerboseMessage = (message: string) => { - if (!core.logging.shouldLogVerbose()) { - return; + if (core.logging.shouldLogVerbose()) { + logger.debug?.(message); } - logger.debug?.(message); }; const mediaMaxBytes = resolveChannelMediaMaxBytes({ @@ -549,1411 +176,61 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} resolveChannelLimitMb: () => undefined, accountId: account.accountId, }) ?? 8 * 1024 * 1024; - const historyLimit = Math.max( - 0, - cfg.messages?.groupChat?.historyLimit ?? DEFAULT_GROUP_HISTORY_LIMIT, - ); - const channelHistories = new Map(); - const defaultGroupPolicy = resolveDefaultGroupPolicy(cfg); - const dmPolicy = account.config.dmPolicy ?? "pairing"; const { groupPolicy, providerMissingFallbackApplied } = resolveAllowlistProviderRuntimeGroupPolicy({ providerConfigPresent: cfg.channels?.mattermost !== undefined, groupPolicy: account.config.groupPolicy, - defaultGroupPolicy, + defaultGroupPolicy: resolveDefaultGroupPolicy(cfg), }); warnMissingProviderGroupPolicyFallbackOnce({ providerMissingFallbackApplied, providerKey: "mattermost", accountId: account.accountId, - log: (message) => logVerboseMessage(message), + log: logVerboseMessage, }); - - const { - resolveMattermostMedia, - sendTypingIndicator, - resolveChannelInfo, - resolveUserInfo, - updateModelPickerPost, - } = createMattermostMonitorResources({ + const resources = createMattermostMonitorResources({ accountId: account.accountId, callbackUrl, client, - logger: { - debug: (message) => logger.debug?.(String(message)), - }, + logger: { debug: (message) => logger.debug?.(String(message)) }, mediaMaxBytes, saveRemoteMedia: (params) => core.channel.media.saveRemoteMedia(params), mediaKindFromMime: (contentType) => core.media.mediaKindFromMime(contentType), }); - - const runModelPickerCommand = async (params: { - commandText: string; - commandAuthorized: boolean; - route: ReturnType; - sessionKey: string; - parentSessionKey?: string; - channelId: string; - senderId: string; - senderName: string; - kind: ChatType; - chatType: "direct" | "group" | "channel"; - channelName?: string; - channelDisplay?: string; - roomLabel: string; - teamId?: string; - postId: string; - messageSid?: string; - effectiveReplyToId?: string; - deliverReplies?: boolean; - }): Promise => { - const to = params.kind === "direct" ? `user:${params.senderId}` : `channel:${params.channelId}`; - const fromLabel = - params.kind === "direct" - ? `Mattermost DM from ${params.senderName}` - : `Mattermost message in ${params.roomLabel} from ${params.senderName}`; - const ctxPayload = finalizeInboundContext({ - Body: params.commandText, - BodyForAgent: params.commandText, - RawBody: params.commandText, - CommandBody: params.commandText, - From: - params.kind === "direct" - ? `mattermost:${params.senderId}` - : params.kind === "group" - ? `mattermost:group:${params.channelId}` - : `mattermost:channel:${params.channelId}`, - To: to, - SessionKey: params.sessionKey, - DmScope: params.route.dmScope, - ParentSessionKey: params.parentSessionKey, - AccountId: params.route.accountId, - ChatType: params.chatType, - ConversationLabel: fromLabel, - GroupSubject: - params.kind !== "direct" ? params.channelDisplay || params.roomLabel : undefined, - GroupChannel: params.channelName ? `#${params.channelName}` : undefined, - GroupSpace: params.teamId, - SenderName: params.senderName, - SenderId: params.senderId, - Provider: "mattermost" as const, - Surface: "mattermost" as const, - MessageSid: params.messageSid ?? `interaction:${params.postId}:${Date.now()}`, - ReplyToId: params.effectiveReplyToId, - MessageThreadId: params.effectiveReplyToId, - Timestamp: Date.now(), - WasMentioned: true, - CommandAuthorized: params.commandAuthorized, - CommandSource: "native" as const, - OriginatingChannel: "mattermost" as const, - OriginatingTo: to, - }); - - const tableMode = core.channel.text.resolveMarkdownTableMode({ - cfg, - channel: "mattermost", - accountId: account.accountId, - }); - const textLimit = core.channel.text.resolveTextChunkLimit( - cfg, - "mattermost", - account.accountId, - { - fallbackLimit: account.textChunkLimit ?? 4000, - }, - ); - const shouldDeliverReplies = params.deliverReplies === true; - const { onModelSelected, typingCallbacks, ...replyPipeline } = - createChannelMessageReplyPipeline({ - cfg, - agentId: params.route.agentId, - channel: "mattermost", - accountId: account.accountId, - typing: shouldDeliverReplies - ? { - start: () => sendTypingIndicator(params.channelId, params.effectiveReplyToId), - onStartError: (err) => { - logTypingFailure({ - log: (message) => logger.debug?.(message), - channel: "mattermost", - target: params.channelId, - error: err, - }); - }, - } - : undefined, - }); - const capturedTexts: string[] = []; - const deliveryBarrier = createMattermostReplyDeliveryBarrier({ - isDirect: params.kind === "direct", - dmRetryOptions: account.config.dmChannelRetry, - }); - await core.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ - ctx: ctxPayload, - cfg, - dispatcherOptions: { - ...replyPipeline, - resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, - onDeliverySettled: deliveryBarrier.markDeliverySettled, - // Picker-triggered confirmations should stay immediate. - deliver: async (payload: ReplyPayload) => { - const trimmedPayload = { - ...payload, - text: core.channel.text.convertMarkdownTables(payload.text ?? "", tableMode).trim(), - }; - - if (!shouldDeliverReplies) { - if (trimmedPayload.text) { - capturedTexts.push(trimmedPayload.text); - } - return; - } - - await deliverMattermostReplyPayload({ - core, - cfg, - payload: trimmedPayload, - to, - accountId: account.accountId, - agentId: params.route.agentId, - replyToId: resolveMattermostReplyRootId({ - kind: params.kind, - threadRootId: params.effectiveReplyToId, - replyToId: trimmedPayload.replyToId, - }), - textLimit, - // The picker path already converts and trims text before capture/delivery. - tableMode: "off", - sendMessage: sendMessageMattermost, - onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, - }); - }, - onError: (err, info) => { - runtime.error?.(`mattermost model picker ${info.kind} reply failed: ${String(err)}`); - }, - typingCallbacks, - }, - replyOptions: { - disableBlockStreaming: - typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined, - onModelSelected, - }, - }); - - return capturedTexts.join("\n\n").trim(); - }; - - async function handleModelPickerInteraction(params: { - payload: { - channel_id: string; - post_id: string; - team_id?: string; - user_id: string; - }; - userName: string; - context: Record; - post: MattermostPost; - }): Promise { - const pickerState = parseMattermostModelPickerContext(params.context); - if (!pickerState) { - return null; - } - - if (pickerState.ownerUserId !== params.payload.user_id) { - return { - ephemeral_text: "Only the person who opened this picker can use it.", - }; - } - - const channelInfo = await resolveChannelInfo(params.payload.channel_id); - const pickerCommandText = - pickerState.action === "select" - ? `/model ${pickerState.provider}/${pickerState.model}` - : pickerState.action === "list" - ? `/models ${pickerState.provider}` - : "/models"; - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ - cfg, - surface: "mattermost", - }); - const hasControlCommand = core.channel.text.hasControlCommand(pickerCommandText, cfg); - const auth = await authorizeMattermostCommandInvocation({ - account, - cfg, - senderId: params.payload.user_id, - senderName: params.userName, - channelId: params.payload.channel_id, - channelInfo, - readStoreAllowFrom: pairing.readAllowFromStore, - allowTextCommands, - hasControlCommand, - }); - if (!auth.ok) { - if (auth.denyReason === "dm-pairing") { - const { code } = await pairing.upsertPairingRequest({ - id: params.payload.user_id, - meta: { name: params.userName }, - }); - return { - ephemeral_text: core.channel.pairing.buildPairingReply({ - channel: "mattermost", - idLine: `Your Mattermost user id: ${params.payload.user_id}`, - code, - }), - }; - } - const denyText = - auth.denyReason === "unknown-channel" - ? "Temporary error: unable to determine channel type. Please try again." - : auth.denyReason === "dm-disabled" - ? "This bot is not accepting direct messages." - : auth.denyReason === "channels-disabled" - ? "Model picker actions are disabled in channels." - : auth.denyReason === "channel-no-allowlist" - ? "Model picker actions are not configured for this channel." - : "Unauthorized."; - return { - ephemeral_text: denyText, - }; - } - const kind = auth.kind; - const chatType = auth.chatType; - const teamId = auth.channelInfo.team_id ?? params.payload.team_id ?? undefined; - const channelName = auth.channelName || undefined; - const channelDisplay = auth.channelDisplay || auth.channelName || params.payload.channel_id; - const roomLabel = auth.roomLabel; - const route = core.channel.routing.resolveAgentRoute({ - cfg, - channel: "mattermost", - accountId: account.accountId, - teamId, - peer: { - kind, - id: kind === "direct" ? params.payload.user_id : params.payload.channel_id, - }, - }); - const replyToMode = resolveMattermostReplyToMode(account, kind); - const threadContext = resolveMattermostThreadSessionContext({ - baseSessionKey: route.sessionKey, - kind, - postId: params.post.id || params.payload.post_id, - replyToMode, - threadRootId: params.post.root_id, - }); - const modelSessionRoute = { - agentId: route.agentId, - sessionKey: threadContext.sessionKey, - }; - - const data = await buildModelsProviderData(cfg, route.agentId); - if (data.providers.length === 0) { - return await updateModelPickerPost({ - channelId: params.payload.channel_id, - postId: params.payload.post_id, - message: "No models available.", - }); - } - - if (pickerState.action === "providers" || pickerState.action === "back") { - const currentModel = resolveMattermostModelPickerCurrentModel({ - cfg, - route: modelSessionRoute, - data, - }); - const view = renderMattermostProviderPickerView({ - ownerUserId: pickerState.ownerUserId, - data, - currentModel, - }); - return await updateModelPickerPost({ - channelId: params.payload.channel_id, - postId: params.payload.post_id, - message: view.text, - buttons: view.buttons, - }); - } - - if (pickerState.action === "list") { - const currentModel = resolveMattermostModelPickerCurrentModel({ - cfg, - route: modelSessionRoute, - data, - }); - const view = renderMattermostModelsPickerView({ - ownerUserId: pickerState.ownerUserId, - data, - provider: pickerState.provider, - page: pickerState.page, - currentModel, - }); - return await updateModelPickerPost({ - channelId: params.payload.channel_id, - postId: params.payload.post_id, - message: view.text, - buttons: view.buttons, - }); - } - - const targetModelRef = `${pickerState.provider}/${pickerState.model}`; - if (!buildMattermostAllowedModelRefs(data).has(targetModelRef)) { - return { - ephemeral_text: `That model is no longer available: ${targetModelRef}`, - }; - } - - void (async () => { - try { - await runModelPickerCommand({ - commandText: `/model ${targetModelRef}`, - commandAuthorized: auth.commandAuthorized, - route, - sessionKey: threadContext.sessionKey, - parentSessionKey: threadContext.parentSessionKey, - channelId: params.payload.channel_id, - senderId: params.payload.user_id, - senderName: params.userName, - kind, - chatType, - channelName, - channelDisplay, - roomLabel, - teamId, - postId: params.payload.post_id, - messageSid: buildMattermostModelPickerSelectMessageSid({ - postId: params.payload.post_id, - provider: pickerState.provider, - model: pickerState.model, - }), - effectiveReplyToId: threadContext.effectiveReplyToId, - deliverReplies: true, - }); - const updatedModel = resolveMattermostModelPickerCurrentModel({ - cfg, - route: modelSessionRoute, - data, - readConsistency: "latest", - }); - const view = renderMattermostModelsPickerView({ - ownerUserId: pickerState.ownerUserId, - data, - provider: pickerState.provider, - page: pickerState.page, - currentModel: updatedModel, - }); - - await updateModelPickerPost({ - channelId: params.payload.channel_id, - postId: params.payload.post_id, - message: view.text, - buttons: view.buttons, - }); - } catch (err) { - runtime.error?.(`mattermost model picker select failed: ${String(err)}`); - } - })(); - - return {}; - } - - const handlePost = async ( - post: MattermostPost, - payload: MattermostEventPayload, - turnAdoptionLifecycle?: MattermostIngressLifecycle, - messageIds?: string[], - ) => { - const channelId = post.channel_id ?? payload.data?.channel_id ?? payload.broadcast?.channel_id; - if (!channelId) { - logVerboseMessage("mattermost: drop post (missing channel id)"); - return; - } - - if (!post.id) { - logVerboseMessage("mattermost: drop post (missing message id)"); - return; - } - const allMessageIds = messageIds?.length ? messageIds : [post.id]; - const senderId = post.user_id ?? payload.broadcast?.user_id; - if (!senderId) { - logVerboseMessage("mattermost: drop post (missing sender id)"); - return; - } - if (senderId === botUserId) { - logVerboseMessage(`mattermost: drop post (self sender=${senderId})`); - return; - } - if (isSystemPost(post)) { - logVerboseMessage(`mattermost: drop post (system post type=${post.type ?? "unknown"})`); - return; - } - - const channelInfo = await resolveChannelInfo(channelId); - const channelType = - normalizeOptionalString(channelInfo?.type) ?? - normalizeOptionalString(payload.data?.channel_type); - if (!channelType) { - logVerboseMessage(`mattermost: drop post (cannot resolve channel type for ${channelId})`); - return; - } - const kind = resolveMattermostTrustedChatKind({ - channelType, - }); - const chatType = channelChatType(kind); - - const senderName = - normalizeOptionalString(payload.data?.sender_name) ?? - normalizeOptionalString((await resolveUserInfo(senderId))?.username) ?? - senderId; - const rawPostText = typeof post.message === "string" ? post.message : ""; - const rawText = normalizeOptionalString(rawPostText) ?? ""; - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ - cfg, - surface: "mattermost", - }); - const isControlCommand = - allowTextCommands && core.channel.commands.isControlCommandMessage(rawText, cfg); - const accessDecision = await resolveMattermostMonitorInboundAccess({ - account, - cfg, - senderId, - senderName, - channelId, - kind, - groupPolicy, - readStoreAllowFrom: pairing.readAllowFromStore, - allowTextCommands, - hasControlCommand: isControlCommand, - eventKind: "message", - mayPair: true, - }); - const commandAuthorized = accessDecision.commandAccess.authorized; - - if (accessDecision.ingress.decision !== "allow") { - if (kind === "direct") { - if (accessDecision.ingress.reasonCode === "dm_policy_disabled") { - logVerboseMessage(`mattermost: drop dm (dmPolicy=disabled sender=${senderId})`); - return; - } - if (accessDecision.ingress.decision === "pairing") { - const { code, created } = await pairing.upsertPairingRequest({ - id: senderId, - meta: { name: senderName }, - }); - logVerboseMessage(`mattermost: pairing request sender=${senderId} created=${created}`); - if (created) { - try { - await sendMessageMattermost( - `user:${senderId}`, - core.channel.pairing.buildPairingReply({ - channel: "mattermost", - idLine: `Your Mattermost user id: ${senderId}`, - code, - }), - { cfg, accountId: account.accountId }, - ); - opts.statusSink?.({ lastOutboundAt: Date.now() }); - } catch (err) { - logVerboseMessage(`mattermost: pairing reply failed for ${senderId}: ${String(err)}`); - } - } - return; - } - logVerboseMessage( - formatMattermostDirectMessageDropLog({ - senderId, - dmPolicy, - reasonCode: accessDecision.senderAccess.reasonCode, - }), - ); - return; - } - if (accessDecision.ingress.reasonCode === "group_policy_disabled") { - logVerboseMessage("mattermost: drop group message (groupPolicy=disabled)"); - return; - } - if (accessDecision.ingress.reasonCode === "group_policy_empty_allowlist") { - logVerboseMessage("mattermost: drop group message (no group allowlist)"); - return; - } - if (accessDecision.ingress.reasonCode === "group_policy_not_allowlisted") { - logVerboseMessage(`mattermost: drop group sender=${senderId} (not in groupAllowFrom)`); - return; - } - logVerboseMessage( - `mattermost: drop group message (groupPolicy=${groupPolicy} reason=${accessDecision.senderAccess.reasonCode})`, - ); - return; - } - - if (kind !== "direct" && accessDecision.commandAccess.shouldBlockControlCommand) { - logInboundDrop({ - log: logVerboseMessage, - channel: "mattermost", - reason: "control command (unauthorized)", - target: senderId, - }); - return; - } - - const teamId = payload.data?.team_id ?? channelInfo?.team_id ?? undefined; - const channelName = payload.data?.channel_name ?? channelInfo?.name ?? ""; - const channelDisplay = - payload.data?.channel_display_name ?? channelInfo?.display_name ?? channelName; - const roomLabel = channelName ? `#${channelName}` : channelDisplay || `#${channelId}`; - - const route = core.channel.routing.resolveAgentRoute({ - cfg, - channel: "mattermost", - accountId: account.accountId, - teamId, - peer: { - kind, - id: kind === "direct" ? senderId : channelId, - }, - }); - - const baseSessionKey = route.sessionKey; - const threadRootId = normalizeOptionalString(post.root_id); - const replyToMode = resolveMattermostReplyToMode(account, kind); - const threadContext = resolveMattermostThreadSessionContext({ - baseSessionKey, - kind, - postId: post.id, - replyToMode, - threadRootId, - }); - const { effectiveReplyToId, sessionKey, parentSessionKey } = threadContext; - const historyKey = resolveMattermostPendingHistoryKey({ kind, sessionKey }); - const fileIds = uniqueStrings(normalizeTrimmedStringList(post.file_ids ?? [])); - const nativeMedia = fileIds.map(() => ({})); - - const mentionRegexes = core.channel.mentions.buildMentionRegexes(cfg, route.agentId); - const wasMentioned = - kind !== "direct" && - ((botUsername - ? normalizeLowercaseStringOrEmpty(rawText).includes( - `@${normalizeLowercaseStringOrEmpty(botUsername)}`, - ) - : false) || - core.channel.mentions.matchesMentionPatterns(rawText, mentionRegexes)); - const pendingBody = formatMattermostPendingMediaText({ - body: rawText, - media: nativeMedia, - }); - const pendingSender = senderName; - const recordPendingHistory = () => { - const trimmed = pendingBody.trim(); - createChannelHistoryWindow({ historyMap: channelHistories }).record({ - limit: historyLimit, - historyKey: historyKey ?? "", - entry: - historyKey && trimmed - ? { - sender: pendingSender, - body: trimmed, - timestamp: typeof post.create_at === "number" ? post.create_at : undefined, - messageId: post.id ?? undefined, - } - : null, - }); - }; - - const oncharEnabled = account.chatmode === "onchar" && kind !== "direct"; - const oncharPrefixes = oncharEnabled ? resolveOncharPrefixes(account.oncharPrefixes) : []; - const oncharResult = oncharEnabled - ? stripOncharPrefix(rawText, oncharPrefixes) - : { triggered: false, stripped: rawText }; - const oncharTriggered = oncharResult.triggered; - const canDetectMention = Boolean(botUsername) || mentionRegexes.length > 0; - // Threads the bot already replied in auto-engage: follow-ups resume without - // a re-mention even under requireMention. Keyed by the thread root id. - const threadAlreadyEngaged = - kind !== "direct" && effectiveReplyToId - ? await hasMattermostThreadParticipationWithPersistence({ - accountId: account.accountId, - channelId, - threadRootId: effectiveReplyToId, - }) - : false; - const shouldRequireMention = - kind !== "direct" && - core.channel.groups.resolveRequireMention({ - cfg, - channel: "mattermost", - accountId: account.accountId, - groupId: channelId, - requireMentionOverride: account.requireMention, - }); - const implicitMentionKinds = implicitMentionKindWhen( - "bot_thread_participant", - threadAlreadyEngaged, - ); - const mentionDecision = resolveMattermostInboundMentionDecision({ - cfg, - accountId: account.accountId, - kind, - requireMention: shouldRequireMention || oncharEnabled, - canDetectMention: canDetectMention || oncharEnabled, - wasMentioned: wasMentioned || oncharTriggered, - implicitMentionKinds, - allowTextCommands, - hasControlCommand: isControlCommand, - commandAuthorized, - }); - const { shouldBypassMention } = mentionDecision; - - if ( - mentionDecision.shouldSkip && - oncharEnabled && - !oncharTriggered && - !wasMentioned && - !shouldBypassMention - ) { - logVerboseMessage( - `mattermost: drop group message (onchar not triggered channel=${channelId} sender=${senderId})`, - ); - recordPendingHistory(); - return; - } - - if (mentionDecision.shouldSkip) { - logVerboseMessage( - `mattermost: drop group message (missing mention channel=${channelId} sender=${senderId} requireMention=${shouldRequireMention} bypass=${shouldBypassMention} canDetectMention=${canDetectMention})`, - ); - recordPendingHistory(); - return; - } - const mediaList = await resolveMattermostMedia(fileIds); - const bodySource = oncharTriggered ? oncharResult.stripped : rawText; - const baseText = formatMattermostInboundMediaText({ - body: bodySource, - nativeMedia, - materializedMedia: mediaList, - }); - const bodyText = normalizeMention(baseText, botUsername); - if ( - mediaList.length === 0 && - shouldDropEmptyMattermostBody({ bodyText, rawText: rawPostText, botUsername }) - ) { - logVerboseMessage( - `mattermost: drop message (empty body after normalization channel=${channelId} sender=${senderId} wasMentioned=${wasMentioned})`, - ); - return; - } - // Mention-only turns need non-empty agent text; the shared reply runner rejects empty - // bodies before model invocation. The guard above ensures this fallback is a bot mention. - const bodyForAgent = bodyText || rawText.trim(); - - core.channel.activity.record({ - channel: "mattermost", - accountId: account.accountId, - direction: "inbound", - }); - - const fromLabel = formatInboundFromLabel({ - isGroup: kind !== "direct", - groupLabel: channelDisplay || roomLabel, - groupId: channelId, - groupFallback: roomLabel || "Channel", - directLabel: senderName, - directId: senderId, - }); - - const textWithId = `${bodyText}\n[mattermost message id: ${post.id ?? "unknown"} channel: ${channelId}]`; - const body = formatInboundEnvelope({ - channel: "Mattermost", - from: fromLabel, - timestamp: typeof post.create_at === "number" ? post.create_at : undefined, - body: textWithId, - chatType, - sender: { name: senderName, id: senderId }, - }); - let combinedBody = body; - if (historyKey) { - const channelHistory = createChannelHistoryWindow({ historyMap: channelHistories }); - combinedBody = channelHistory.buildPendingContext({ - historyKey, - limit: historyLimit, - currentMessage: combinedBody, - formatEntry: (entry) => - formatInboundEnvelope({ - channel: "Mattermost", - from: fromLabel, - timestamp: entry.timestamp, - body: `${entry.body}${ - entry.messageId ? ` [id:${entry.messageId} channel:${channelId}]` : "" - }`, - chatType, - senderLabel: entry.sender, - }), - }); - } - - const to = kind === "direct" ? `user:${senderId}` : `channel:${channelId}`; - const mediaPayload = buildMattermostInboundMediaPayload(mediaList); - const commandBody = rawText.trim(); - const inboundHistory = - historyKey && historyLimit > 0 - ? createChannelHistoryWindow({ historyMap: channelHistories }).buildInboundHistory({ - historyKey, - limit: historyLimit, - }) - : undefined; - const ctxPayload = finalizeInboundContext({ - Body: combinedBody, - BodyForAgent: bodyForAgent, - InboundHistory: inboundHistory, - RawBody: commandBody, - CommandBody: commandBody, - BodyForCommands: commandBody, - From: - kind === "direct" - ? `mattermost:${senderId}` - : kind === "group" - ? `mattermost:group:${channelId}` - : `mattermost:channel:${channelId}`, - To: to, - SessionKey: sessionKey, - DmScope: route.dmScope, - ParentSessionKey: parentSessionKey, - AccountId: route.accountId, - ChatType: chatType, - ConversationLabel: fromLabel, - GroupSubject: kind !== "direct" ? channelDisplay || roomLabel : undefined, - GroupChannel: channelName ? `#${channelName}` : undefined, - GroupSpace: teamId, - SenderName: senderName, - SenderId: senderId, - Provider: "mattermost" as const, - Surface: "mattermost" as const, - MessageSid: post.id ?? undefined, - MessageSids: allMessageIds.length > 1 ? allMessageIds : undefined, - MessageSidFirst: allMessageIds.length > 1 ? allMessageIds[0] : undefined, - MessageSidLast: - allMessageIds.length > 1 ? allMessageIds[allMessageIds.length - 1] : undefined, - ReplyToId: effectiveReplyToId, - MessageThreadId: effectiveReplyToId, - Timestamp: typeof post.create_at === "number" ? post.create_at : undefined, - WasMentioned: kind !== "direct" ? mentionDecision.effectiveWasMentioned : undefined, - CommandAuthorized: commandAuthorized, - // Tag typed text-slash control commands (e.g. ` /new`, ` /reset` sent via the regular - // post path rather than Mattermost's native slash UI) so the explicit-command turn - // exception in source-reply-delivery-mode.ts surfaces their acknowledgements under - // message_tool_only delivery modes (e.g. Codex harness DMs). Mirrors iMessage #82642. - CommandSource: commandAuthorized && isControlCommand ? ("text" as const) : undefined, - OriginatingChannel: "mattermost" as const, - OriginatingTo: to, - ...mediaPayload, - }); - const pinnedMainDmOwner = - kind === "direct" - ? resolvePinnedMainDmOwnerFromAllowlist({ - dmScope: cfg.session?.dmScope, - allowFrom: account.config.allowFrom, - normalizeEntry: normalizeMattermostAllowEntry, - }) - : null; - - const previewLine = truncateUtf16Safe(bodyText, 200).replace(/\n/g, "\\n"); - logVerboseMessage( - `mattermost inbound: from=${ctxPayload.From} len=${bodyText.length} preview="${previewLine}"`, - ); - - const textLimit = core.channel.text.resolveTextChunkLimit( - cfg, - "mattermost", - account.accountId, - { - fallbackLimit: account.textChunkLimit ?? 4000, - }, - ); - const tableMode = core.channel.text.resolveMarkdownTableMode({ - cfg, - channel: "mattermost", - accountId: account.accountId, - }); - const chunkMode = core.channel.text.resolveChunkMode(cfg, "mattermost", account.accountId); - - const { onModelSelected, typingCallbacks, resolveResponsePrefix, ...replyPipeline } = - createChannelMessageReplyPipeline({ - cfg, - agentId: route.agentId, - channel: "mattermost", - accountId: account.accountId, - typing: { - start: () => sendTypingIndicator(channelId, effectiveReplyToId), - onStartError: (err) => { - logTypingFailure({ - log: (message) => logger.debug?.(message), - channel: "mattermost", - target: channelId, - error: err, - }); - }, - }, - }); - const draftPreviewEnabled = account.streamingMode !== "off"; - const draftToolProgressEnabled = shouldUpdateMattermostDraftToolProgress(account); - const suppressDefaultToolProgressMessages = - shouldSuppressMattermostDefaultToolProgressMessages(account); - const draftStream = draftPreviewEnabled - ? createMattermostDraftStream({ - client, - channelId, - rootId: effectiveReplyToId, - throttleMs: 1200, - chunkText: (value) => - core.channel.text.chunkMarkdownTextWithMode( - core.channel.text.convertMarkdownTables(value, tableMode), - textLimit, - chunkMode, - ), - log: logVerboseMessage, - warn: logVerboseMessage, - }) - : createDisabledMattermostDraftStream(); - const previewBoundaryController = createMattermostDraftPreviewBoundaryController({ - enabled: draftPreviewEnabled && account.streamingMode === "block", - forceNewMessage: async () => { - await draftStream.forceNewMessage(); - }, - }); - let lastPartialText = ""; - let firstAssistantPreviewPrefix: string | undefined; - let firstAssistantPreviewPrefixPending = true; - let currentAssistantPreviewUsesPrefix = false; - let blockPreviewActivity: "none" | "reasoning" | "text" | "tool" = "none"; - let blockPreviewAssistantMessagePending = false; - const progressDraft = createChannelProgressDraftCompositor({ - entry: account.config, - mode: account.streamingMode, - active: draftPreviewEnabled, - seed: `${account.accountId}:${channelId}`, - update: async (previewText, options) => { - draftStream.update(previewText); - if (options?.flush) { - await draftStream.flush(); - } - }, - }); - const enterBlockPreviewActivity = (activity: "reasoning" | "text" | "tool") => { - if (account.streamingMode !== "block") { - return undefined; - } - const continuingToolActivity = activity === "tool" && blockPreviewActivity === "tool"; - const continuingTextActivity = - activity === "text" && - blockPreviewActivity === "text" && - !blockPreviewAssistantMessagePending; - const continuingReasoningActivity = - activity === "reasoning" && blockPreviewActivity === "reasoning"; - const continuesCurrentActivity = - continuingToolActivity || continuingTextActivity || continuingReasoningActivity; - // Reasoning placeholders are transient. A visible successor replaces the same draft; - // only entering reasoning from a durable text/tool block rotates generations. - const reusesReasoningGeneration = blockPreviewActivity === "reasoning"; - const startsNewGeneration = !continuesCurrentActivity && !reusesReasoningGeneration; - if (startsNewGeneration) { - currentAssistantPreviewUsesPrefix = false; - } - const boundarySettled = startsNewGeneration - ? previewBoundaryController.noteBoundary() - : undefined; - // Message-start is only a candidate boundary: consecutive tool-only turns stay in the - // same activity post, while the first visible text or reasoning starts a new block. - if (!continuesCurrentActivity) { - progressDraft.reset(); - } - blockPreviewActivity = activity; - blockPreviewAssistantMessagePending = false; - if (activity === "tool") { - lastPartialText = ""; - } - return boundarySettled; - }; - const previewState: MattermostDraftPreviewState = { - finalizedViaPreviewPost: false, - }; - - const resolveFinalDeliveryText = (text?: string) => { - if (typeof text !== "string") { - return undefined; - } - const resolution = draftStream.resolveFinalText(text); - return resolution.kind === "already-delivered" ? "" : resolution.text; - }; - - const resolvePreviewFinalText = (text?: string) => { - const deliveryText = resolveFinalDeliveryText(text); - if (typeof deliveryText !== "string") { - return undefined; - } - const formatted = core.channel.text.convertMarkdownTables(deliveryText, tableMode); - const chunks = core.channel.text.chunkMarkdownTextWithMode(formatted, textLimit, chunkMode); - if (!chunks.length && formatted) { - chunks.push(formatted); - } - if (chunks.length != 1) { - return undefined; - } - const trimmed = chunks[0]?.trim(); - if (!trimmed) { - return undefined; - } - if ( - lastPartialText && - lastPartialText.startsWith(trimmed) && - trimmed.length < lastPartialText.length - ) { - return undefined; - } - return trimmed; - }; - - const updateDraftFromPartial = (text?: string) => { - const cleaned = text?.trim(); - if (!cleaned) { - return undefined; - } - if (cleaned === lastPartialText) { - return undefined; - } - if ( - lastPartialText && - lastPartialText.startsWith(cleaned) && - cleaned.length < lastPartialText.length - ) { - return undefined; - } - const boundarySettled = enterBlockPreviewActivity("text"); - lastPartialText = cleaned; - if (firstAssistantPreviewPrefixPending) { - firstAssistantPreviewPrefix = resolveResponsePrefix?.(); - firstAssistantPreviewPrefixPending = false; - currentAssistantPreviewUsesPrefix = Boolean(firstAssistantPreviewPrefix); - } - const previewText = - currentAssistantPreviewUsesPrefix && firstAssistantPreviewPrefix - ? cleaned.startsWith(firstAssistantPreviewPrefix) - ? cleaned - : `${firstAssistantPreviewPrefix} ${cleaned}` - : cleaned; - draftStream.updateAssistantText(previewText); - previewBoundaryController.noteUpdate(); - return boundarySettled; - }; - - const deliveryBarrier = createMattermostReplyDeliveryBarrier({ - isDirect: kind === "direct", - dmRetryOptions: account.config.dmChannelRetry, - }); - const dispatcherOptions: NonNullable = { - ...replyPipeline, - resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, - onDeliverySettled: deliveryBarrier.markDeliverySettled, - humanDelay: resolveHumanDelayConfig(cfg, route.agentId), - typingCallbacks, - }; - const delivery: ChannelInboundTurnPlan["delivery"] = { - deliver: async (payloadEntry: ReplyPayload, info) => { - if (info.kind === "final") { - await enterBlockPreviewActivity("text"); - // Final text resolution uses only generations confirmed visible. Join prior - // boundary work before the synchronous final-edit decision. - await draftStream.settleBoundaries(); - progressDraft.markFinalReplyStarted(); - } - // A visible same-thread final arrives either via a normal send or by editing - // the draft preview in place; record participation on whichever path fires. - const markThreadParticipation = () => { - if (kind !== "direct" && effectiveReplyToId) { - recordMattermostThreadParticipation(account.accountId, channelId, effectiveReplyToId, { - agentId: route.agentId, - }); - } - }; - await deliverMattermostReplyWithDraftPreview({ - payload: payloadEntry, - info, - kind, - client, - draftStream, - effectiveReplyToId, - resolvePreviewFinalText, - previewState, - logVerboseMessage, - recordThreadParticipation: markThreadParticipation, - deliverPayload: async (payloadToDeliver) => { - const finalTextResolution = - info.kind === "final" && - !payloadToDeliver.isError && - typeof payloadToDeliver.text === "string" - ? draftStream.resolveFinalText(payloadToDeliver.text) - : undefined; - const resolvedPayload = finalTextResolution - ? { - ...payloadToDeliver, - text: - finalTextResolution.kind === "already-delivered" - ? "" - : finalTextResolution.text, - } - : payloadToDeliver; - const outcome = await deliverMattermostReplyPayload({ - core, - cfg, - payload: resolvedPayload, - to, - accountId: account.accountId, - agentId: route.agentId, - replyToId: resolveMattermostReplyRootId({ - kind, - threadRootId: effectiveReplyToId, - replyToId: payloadToDeliver.replyToId, - }), - textLimit, - tableMode, - sendMessage: sendMessageMattermost, - onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, - }); - // Record only on a visible send so threads we merely observed - // (reasoning-only/empty/suppressed) do not auto-engage later. - if (outcome === "text" || outcome === "media") { - markThreadParticipation(); - } else if (outcome === "empty" && finalTextResolution?.kind === "already-delivered") { - // The terminal payload confirms the already-published assistant block as - // the visible final reply even though this delivery has no remaining text. - markThreadParticipation(); - } - const deliveryLog = formatMattermostFinalDeliveryOutcomeLog({ - outcome, - payload: resolvedPayload, - to, - accountId: account.accountId, - agentId: route.agentId, - }); - if (deliveryLog) { - runtime.log?.(deliveryLog); - } - }, - }); - if (info.kind === "final") { - progressDraft.markFinalReplyDelivered(); - } - }, - onError: (err, info) => { - runtime.error?.(`mattermost ${info.kind} reply failed: ${String(err)}`); - }, - }; - - const inboundLastRouteSessionKey = resolveInboundLastRouteSessionKey({ - route, - sessionKey: route.sessionKey, - }); - - try { - await core.channel.inbound.run({ - channel: "mattermost", - accountId: route.accountId, - raw: post, - adapter: { - ingest: () => ({ - id: post.id ?? `${to}:${Date.now()}`, - timestamp: post.create_at ?? undefined, - rawText, - textForAgent: ctxPayload.BodyForAgent, - textForCommands: ctxPayload.CommandBody, - raw: post, - }), - resolveTurn: () => ({ - cfg, - channel: "mattermost", - accountId: route.accountId, - route: { - agentId: route.agentId, - dmScope: route.dmScope, - sessionKey: route.sessionKey, - }, - ctxPayload, - record: { - updateLastRoute: - kind === "direct" - ? { - sessionKey: inboundLastRouteSessionKey, - channel: "mattermost", - to, - accountId: route.accountId, - mainDmOwnerPin: - inboundLastRouteSessionKey === route.mainSessionKey && pinnedMainDmOwner - ? { - ownerRecipient: pinnedMainDmOwner, - senderRecipient: normalizeMattermostAllowEntry(senderId), - onSkip: ({ - ownerRecipient, - senderRecipient, - }: { - ownerRecipient: string; - senderRecipient: string; - }) => { - logVerboseMessage( - `mattermost: skip main-session last route for ${senderRecipient} (pinned owner ${ownerRecipient})`, - ); - }, - } - : undefined, - } - : undefined, - onRecordError: (err) => { - logVerboseMessage( - `mattermost: failed updating session meta id=${post.id ?? "unknown"}: ${String(err)}`, - ); - }, - }, - history: { - isGroup: Boolean(historyKey), - historyKey: historyKey ?? undefined, - historyMap: channelHistories, - limit: historyLimit, - }, - dispatcherOptions, - delivery, - replyOptions: { - ...(turnAdoptionLifecycle - ? bindIngressLifecycleToReplyOptions(turnAdoptionLifecycle) - : {}), - allowProgressCallbacksWhenSourceDeliverySuppressed: draftToolProgressEnabled - ? true - : undefined, - preserveProgressCallbackStartOrder: draftPreviewEnabled ? true : undefined, - onObservedReplyDelivery: draftToolProgressEnabled - ? () => draftStream.clear() - : undefined, - disableBlockStreaming: draftPreviewEnabled - ? true - : typeof account.blockStreaming === "boolean" - ? !account.blockStreaming - : undefined, - ...(suppressDefaultToolProgressMessages - ? { suppressDefaultToolProgressMessages: true } - : {}), - onModelSelected, - onPartialReply: (payloadResult) => { - if (account.streamingMode !== "progress") { - return updateDraftFromPartial(payloadResult.text); - } - return undefined; - }, - onAssistantMessageStart: () => { - lastPartialText = ""; - progressDraft.resetReasoningProgress(); - if (account.streamingMode === "block") { - blockPreviewAssistantMessagePending = true; - return; - } - if (account.streamingMode !== "progress") { - progressDraft.reset(); - } - }, - onReasoningEnd: () => { - // Hidden reasoning has no visible boundary. Only transitions that - // actually render text, reasoning, or tools rotate preview posts. - lastPartialText = ""; - progressDraft.resetReasoningProgress(); - if (account.streamingMode !== "block" && account.streamingMode !== "progress") { - progressDraft.reset(); - } - }, - onReasoningStream: async (payloadResult) => { - if (account.streamingMode === "progress") { - await progressDraft.pushReasoningProgress(payloadResult.text || "Thinking…", { - snapshot: payloadResult.isReasoningSnapshot === true, - }); - return; - } - if (!lastPartialText) { - const boundarySettled = enterBlockPreviewActivity("reasoning"); - draftStream.update("Thinking…"); - previewBoundaryController.noteUpdate(); - await boundarySettled; - } - }, - onToolStart: async (payloadValue) => { - if (!draftToolProgressEnabled) { - return; - } - const boundarySettled = enterBlockPreviewActivity("tool"); - // Boundary detach and progress staging both happen synchronously before - // their first await; agent callbacks may be dispatched fire-and-forget. - const progressSettled = progressDraft.pushToolProgress( - buildChannelProgressDraftLineForEntry( - account.config, - { - event: "tool", - itemId: payloadValue.itemId, - toolCallId: payloadValue.toolCallId, - name: payloadValue.name, - phase: payloadValue.phase, - args: payloadValue.args, - }, - payloadValue.detailMode ? { detailMode: payloadValue.detailMode } : undefined, - ), - { startImmediately: true }, - ); - previewBoundaryController.noteUpdate(); - await Promise.all([boundarySettled, progressSettled]); - }, - onItemEvent: async (payloadLocal) => { - if (!draftToolProgressEnabled) { - return; - } - const boundarySettled = enterBlockPreviewActivity("tool"); - const progressSettled = progressDraft.pushToolProgress( - buildChannelProgressDraftLineForEntry(account.config, { - event: "item", - itemId: payloadLocal.itemId, - itemKind: payloadLocal.kind, - title: payloadLocal.title, - name: payloadLocal.name, - phase: payloadLocal.phase, - status: payloadLocal.status, - summary: payloadLocal.summary, - progressText: payloadLocal.progressText, - meta: payloadLocal.meta, - }), - { startImmediately: true }, - ); - previewBoundaryController.noteUpdate(); - await Promise.all([boundarySettled, progressSettled]); - }, - }, - }), - }, - }); - } finally { - try { - await draftStream.stop(); - } catch (err) { - logVerboseMessage(`mattermost draft preview cleanup failed: ${String(err)}`); - } - } - }; - - const handleReactionEvent = async (payload: MattermostEventPayload) => { - const reactionData = payload.data?.reaction; - if (!reactionData) { - return; - } - let reaction: MattermostReaction | null = null; - if (typeof reactionData === "string") { - try { - reaction = JSON.parse(reactionData) as MattermostReaction; - } catch { - return; - } - } else if (typeof reactionData === "object") { - reaction = reactionData as MattermostReaction; - } - if (!reaction) { - return; - } - - const userId = reaction.user_id?.trim(); - const postId = reaction.post_id?.trim(); - const emojiName = reaction.emoji_name?.trim(); - if (!userId || !postId || !emojiName) { - return; - } - - // Skip reactions from the bot itself - if (userId === botUserId) { - return; - } - - const isRemoved = payload.event === "reaction_removed"; - const action = isRemoved ? "removed" : "added"; - - const senderInfo = await resolveUserInfo(userId); - const senderName = normalizeOptionalString(senderInfo?.username) ?? userId; - - // Resolve the channel from broadcast or post to route to the correct agent session - const channelId = resolveMattermostReactionChannelId(payload); - if (!channelId) { - // Without a channel id we cannot verify DM/group policies — drop to be safe - logVerboseMessage( - `mattermost: drop reaction (no channel_id in broadcast, cannot enforce policy)`, - ); - return; - } - const channelInfo = await resolveChannelInfo(channelId); - if (!channelInfo?.type) { - // Cannot determine channel type — drop to avoid policy bypass - logVerboseMessage(`mattermost: drop reaction (cannot resolve channel type for ${channelId})`); - return; - } - const kind = mapMattermostChannelTypeToChatType(channelInfo.type); - - // Enforce DM/group policy and allowlist checks (same as normal messages). - const reactionAccess = await resolveMattermostMonitorInboundAccess({ - account, - cfg, - senderId: userId, - senderName, - channelId, - kind, - groupPolicy, - readStoreAllowFrom: pairing.readAllowFromStore, - allowTextCommands: false, - hasControlCommand: false, - eventKind: "reaction", - mayPair: false, - }); - if (reactionAccess.ingress.decision !== "allow") { - if (kind === "direct") { - logVerboseMessage( - `mattermost: drop reaction (dmPolicy=${dmPolicy} sender=${userId} reason=${reactionAccess.senderAccess.reasonCode})`, - ); - } else { - logVerboseMessage( - `mattermost: drop reaction (groupPolicy=${groupPolicy} sender=${userId} reason=${reactionAccess.senderAccess.reasonCode} channel=${channelId})`, - ); - } - return; - } - - const teamId = channelInfo?.team_id ?? undefined; - const route = core.channel.routing.resolveAgentRoute({ - cfg, - channel: "mattermost", - accountId: account.accountId, - teamId, - peer: { - kind, - id: kind === "direct" ? userId : channelId, - }, - }); - const sessionKey = route.sessionKey; - - const eventText = `Mattermost reaction ${action}: :${emojiName}: by @${senderName} on post ${postId} in channel ${channelId}`; - - core.system.enqueueSystemEvent(eventText, { - sessionKey, - contextKey: `mattermost:reaction:${postId}:${emojiName}:${userId}:${action}`, - }); - - logVerboseMessage( - `mattermost reaction: ${action} :${emojiName}: by ${senderName} on ${postId}`, - ); - }; - - const inboundDebounceMs = core.channel.debounce.resolveInboundDebounceMs({ + const monitor: MattermostMonitorContext = { + core, + runtime, cfg, - channel: "mattermost", + account, + client, + pairing, + botUserId, + botUsername, + groupPolicy, + resources, + logDebugMessage: (message) => logger.debug?.(message), + logVerboseMessage, + statusSink: opts.statusSink, + }; + const unregisterInteractions = registerMattermostInteractions({ + monitor, + interactionPath, + allowedSourceIps: + allowedInteractionSourceIps.length > 0 ? allowedInteractionSourceIps : ["127.0.0.1", "::1"], + handleModelPickerInteraction: createMattermostModelPickerInteractionHandler(monitor), }); + const handlePost = createMattermostPostHandler(monitor); + const handleReactionEvent = createMattermostReactionHandler(monitor); + const debouncer = core.channel.debounce.createInboundDebouncer<{ post: MattermostPost; payload: MattermostEventPayload; turnAdoptionLifecycle: MattermostIngressLifecycle; }>({ - debounceMs: inboundDebounceMs, + debounceMs: core.channel.debounce.resolveInboundDebounceMs({ + cfg, + channel: "mattermost", + }), buildKey: (entry) => { const channelId = entry.post.channel_id ?? @@ -1963,18 +240,14 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} return null; } const threadId = normalizeOptionalString(entry.post.root_id); - const threadKey = threadId ? `thread:${threadId}` : "channel"; - return `mattermost:${account.accountId}:${channelId}:${threadKey}`; + return `mattermost:${account.accountId}:${channelId}:${threadId ? `thread:${threadId}` : "channel"}`; }, shouldDebounce: (entry) => { - if (entry.post.file_ids && entry.post.file_ids.length > 0) { + if (entry.post.file_ids?.length) { return false; } const text = normalizeOptionalString(entry.post.message) ?? ""; - if (!text) { - return false; - } - return !core.channel.commands.isControlCommandMessage(text, cfg); + return Boolean(text) && !core.channel.commands.isControlCommandMessage(text, cfg); }, onFlush: async (entries) => { const last = entries.at(-1); @@ -1990,13 +263,12 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} await settle(); return; } - const combinedText = entries - .map((entry) => normalizeOptionalString(entry.post.message) ?? "") - .filter(Boolean) - .join("\n"); const mergedPost: MattermostPost = { ...last.post, - message: combinedText, + message: entries + .map((entry) => normalizeOptionalString(entry.post.message) ?? "") + .filter(Boolean) + .join("\n"), file_ids: [], }; await handlePost( @@ -2015,66 +287,49 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} runtime.error?.(`mattermost debounce flush failed: ${String(err)}`); }, }); - const ingress = createMattermostIngressMonitor({ accountId: account.accountId, runtime, abortSignal: opts.abortSignal, dispatch: async (post, payload, turnAdoptionLifecycle) => { // Deferred claims settle through lifecycle callbacks, so terminal flush - // errors (401/403 included) abandon rather than hit the non-retryable - // classifier; the drain's attempt/age retry policy still dead-letters - // them — auth failures just spend the bounded retry budget first. - // Accepted tradeoff over threading a fail() channel through deferral. + // errors spend the drain's bounded retry budget before dead-lettering. await debouncer.enqueue({ post, payload, turnAdoptionLifecycle }); return { kind: "deferred" }; }, }); - - const wsUrl = buildMattermostWsUrl(baseUrl); - let seq = 1; + let sequence = 1; const connectOnce = createMattermostConnectOnce({ - wsUrl, + wsUrl: `${baseUrl.replace(/^http/i, "ws")}/api/v4/websocket`, botToken, abortSignal: opts.abortSignal, statusSink: opts.statusSink, runtime, webSocketFactory: opts.webSocketFactory, - nextSeq: () => seq++, - getBotUpdateAt: async () => { - const me = await fetchMattermostMe(client); - return me.update_at ?? 0; - }, + nextSeq: () => sequence++, + getBotUpdateAt: async () => (await fetchMattermostMe(client)).update_at ?? 0, onPosted: ingress.receive, - onReaction: async (payload) => { - await handleReactionEvent(payload); - }, + onReaction: handleReactionEvent, }); let slashShutdownCleanup: Promise | null = null; - - // Clean up slash commands on shutdown if (slashEnabled) { const runAbortCleanup = () => { if (slashShutdownCleanup) { return; } - // Snapshot registered commands before deactivating state. - // This listener may run concurrently with startup in a new process, so we keep - // monitor shutdown alive until the remote cleanup completes. + // Snapshot registered commands before deactivation and keep monitor shutdown + // alive until the remote cleanup completes. const commands = getSlashCommandState(account.accountId)?.registeredCommands ?? []; - // Deactivate state immediately to prevent new local dispatches during teardown. deactivateSlashCommands(account.accountId); - slashShutdownCleanup = cleanupSlashCommands({ client, commands, - log: (msg) => runtime.log?.(msg), + log: (message) => runtime.log?.(message), }).catch((err: unknown) => { runtime.error?.(`mattermost: slash cleanup failed: ${String(err)}`); }); }; - if (opts.abortSignal?.aborted) { runAbortCleanup(); } else { @@ -2098,10 +353,8 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} await ingress.stop(); unregisterInteractions?.(); } - const slashShutdownCleanupPromise = slashShutdownCleanup; if (slashShutdownCleanupPromise) { await Promise.resolve(slashShutdownCleanupPromise); } } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/channels/turn/message-turn-guardrails.test.ts b/src/channels/turn/message-turn-guardrails.test.ts index 7154ca87e03c..d2641159d6a2 100644 --- a/src/channels/turn/message-turn-guardrails.test.ts +++ b/src/channels/turn/message-turn-guardrails.test.ts @@ -15,7 +15,7 @@ const migratedMessageTurnFiles = [ "extensions/imessage/src/monitor/inbound-processing.ts", "extensions/line/src/bot-handlers.ts", "extensions/line/src/bot-message-context.ts", - "extensions/mattermost/src/mattermost/monitor.ts", + "extensions/mattermost/src/mattermost/monitor-posts.ts", "extensions/msteams/src/monitor-handler/message-handler.ts", "extensions/signal/src/monitor/event-handler.ts", "extensions/slack/src/monitor/message-handler/prepare.ts", @@ -33,7 +33,7 @@ const historyWindowFiles = [ "extensions/imessage/src/monitor/inbound-processing.ts", "extensions/line/src/bot-handlers.ts", "extensions/line/src/group-history.ts", - "extensions/mattermost/src/mattermost/monitor.ts", + "extensions/mattermost/src/mattermost/monitor-posts.ts", "extensions/msteams/src/monitor-handler/message-handler.ts", "extensions/qqbot/src/bridge/sdk-adapter.ts", "extensions/signal/src/monitor/event-handler.ts",