diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index e9b54d6786eb..a014d3d1ead1 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -266,7 +266,6 @@ extensions/slack/src/monitor/events/interactions.block-actions.ts extensions/slack/src/monitor/events/interactions.test.ts extensions/slack/src/monitor/media.test.ts extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts -extensions/slack/src/monitor/message-handler/dispatch.ts extensions/slack/src/monitor/message-handler/prepare.test.ts extensions/slack/src/monitor/message-handler/prepare.ts extensions/slack/src/monitor/provider.ts diff --git a/extensions/slack/src/monitor/message-handler/dispatch-helpers.ts b/extensions/slack/src/monitor/message-handler/dispatch-helpers.ts new file mode 100644 index 000000000000..04d230ebbbc4 --- /dev/null +++ b/extensions/slack/src/monitor/message-handler/dispatch-helpers.ts @@ -0,0 +1,274 @@ +import type { ChannelBotLoopProtectionFacts } from "openclaw/plugin-sdk/channel-inbound"; +import { resolveChannelProgressDraftConfig } from "openclaw/plugin-sdk/channel-outbound"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { mergePairLoopGuardConfig } from "openclaw/plugin-sdk/pair-loop-guard-runtime"; +import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload"; +import type { ReplyDispatchKind, ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; +import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { resolveSlackReplyRenderPlan } from "../../reply-blocks.js"; +import type { SlackMessageEvent } from "../../types.js"; +import { readSlackReplyBlocks, resolveSlackThreadTs } from "../replies.js"; +import { resolveSlackTimestampMs } from "./timestamp.js"; +import type { PreparedSlackMessage } from "./types.js"; + +function resolveSlackMessageTimestampMs(message: SlackMessageEvent): number | undefined { + const ts = message.event_ts ?? message.ts; + return resolveSlackTimestampMs(ts); +} + +export function resolveSlackBotLoopProtection( + prepared: PreparedSlackMessage, +): ChannelBotLoopProtectionFacts | undefined { + const senderBotId = prepared.message.bot_id; + if (!senderBotId) { + return undefined; + } + const receiverBotId = prepared.ctx.botId || prepared.ctx.botUserId; + if ( + !receiverBotId || + senderBotId === prepared.ctx.botId || + prepared.message.user === prepared.ctx.botUserId + ) { + return undefined; + } + return { + scopeId: prepared.route.accountId, + conversationId: prepared.message.channel, + senderId: senderBotId, + receiverId: receiverBotId, + config: mergePairLoopGuardConfig( + prepared.account.config.botLoopProtection, + prepared.channelConfig?.botLoopProtection, + ), + defaultsConfig: prepared.ctx.cfg.channels?.defaults?.botLoopProtection, + defaultEnabled: true, + nowMs: resolveSlackMessageTimestampMs(prepared.message), + }; +} + +export function isSlackStreamingEnabled(params: { + mode: "off" | "partial" | "block" | "progress"; + nativeStreaming: boolean; + nativeProgressTaskCards?: boolean; +}): boolean { + if (params.mode === "partial") { + return params.nativeStreaming; + } + if (params.mode === "progress") { + return params.nativeStreaming && params.nativeProgressTaskCards === true; + } + return false; +} + +export function shouldEnableSlackPreviewStreaming(params: { + mode: "off" | "partial" | "block" | "progress"; +}): boolean { + return params.mode !== "off"; +} + +export function shouldInitializeSlackDraftStream(params: { + previewStreamingEnabled: boolean; + useStreaming: boolean; +}): boolean { + return params.previewStreamingEnabled && !params.useStreaming; +} + +export function resolveSlackDisableBlockStreaming(params: { + useStreaming: boolean; + shouldUseDraftStream: boolean; + blockStreamingEnabled: boolean | undefined; +}): boolean | undefined { + if (params.useStreaming || params.shouldUseDraftStream) { + return true; + } + return typeof params.blockStreamingEnabled === "boolean" + ? !params.blockStreamingEnabled + : undefined; +} + +export function resolveExplicitSlackProgressTitle( + entry: Parameters[0], +): string | undefined { + const label = resolveChannelProgressDraftConfig(entry).label; + if (typeof label !== "string") { + return undefined; + } + const trimmed = label.trim(); + return trimmed && trimmed.toLowerCase() !== "auto" ? trimmed : undefined; +} + +export function resolveSlackNativeProgressTaskCards( + entry: Parameters[0], +): boolean { + const streaming = entry?.streaming; + if (!streaming || typeof streaming !== "object" || Array.isArray(streaming)) { + return false; + } + const progressConfig = (streaming as Record).progress; + return ( + Boolean(progressConfig) && + typeof progressConfig === "object" && + !Array.isArray(progressConfig) && + (progressConfig as { nativeTaskCards?: unknown }).nativeTaskCards === true + ); +} + +export function resolveSlackStreamingThreadHint(params: { + replyToMode: "off" | "first" | "all" | "batched"; + incomingThreadTs: string | undefined; + messageTs: string | undefined; + isThreadReply?: boolean; +}): string | undefined { + return resolveSlackThreadTs({ + replyToMode: params.replyToMode, + incomingThreadTs: params.incomingThreadTs, + messageTs: params.messageTs, + hasReplied: false, + isThreadReply: params.isThreadReply, + }); +} + +export type SlackEventDeliveryAttempt = { + kind: ReplyDispatchKind; + payload: ReplyPayload; + threadTs?: string; + textOverride?: string; +}; + +const SLACK_STREAM_RECIPIENT_TEAM_CACHE_MAX = 2000; +const slackStreamRecipientTeamCaches = new WeakMap>(); + +function getSlackStreamRecipientTeamCache(client: object): Map { + const existing = slackStreamRecipientTeamCaches.get(client); + if (existing) { + return existing; + } + const cache = new Map(); + slackStreamRecipientTeamCaches.set(client, cache); + return cache; +} + +function buildSlackEventDeliveryKey(params: SlackEventDeliveryAttempt): string | null { + const reply = resolveSendableOutboundReplyParts(params.payload, { + text: params.textOverride, + }); + const renderPlan = resolveSlackReplyRenderPlan( + params.payload, + params.textOverride ?? params.payload.text, + ); + const plannedBlocks = + renderPlan.mode === "single" ? renderPlan.blocks : renderPlan.blockPart?.blocks; + const slackBlocks = readSlackReplyBlocks(params.payload) ?? plannedBlocks; + const renderedText = renderPlan.mode === "single" ? renderPlan.text : renderPlan.fallbackText; + if (!reply.hasContent && !slackBlocks?.length && !renderedText.trim()) { + return null; + } + return JSON.stringify({ + kind: params.kind, + threadTs: params.threadTs ?? "", + replyToId: params.payload.replyToId ?? null, + text: renderedText || reply.trimmedText, + mediaUrls: reply.mediaUrls, + blocks: slackBlocks ?? null, + }); +} + +function readSlackStreamRecipientTeamCache(params: { + client: object; + fallbackTeamId?: string; + userId?: string; +}): string | undefined { + if (!params.fallbackTeamId || !params.userId) { + return undefined; + } + const cacheKey = `${params.fallbackTeamId}:${params.userId}`; + const cache = getSlackStreamRecipientTeamCache(params.client); + const cached = cache.get(cacheKey); + if (!cached) { + return undefined; + } + cache.delete(cacheKey); + cache.set(cacheKey, cached); + return cached; +} + +function rememberSlackStreamRecipientTeam(params: { + client: object; + fallbackTeamId?: string; + userId?: string; + teamId: string; +}): void { + if (!params.fallbackTeamId || !params.userId) { + return; + } + const cacheKey = `${params.fallbackTeamId}:${params.userId}`; + const cache = getSlackStreamRecipientTeamCache(params.client); + if (cache.has(cacheKey)) { + cache.delete(cacheKey); + } + cache.set(cacheKey, params.teamId); + if (cache.size > SLACK_STREAM_RECIPIENT_TEAM_CACHE_MAX) { + const oldest = cache.keys().next().value; + if (oldest) { + cache.delete(oldest); + } + } +} + +export function createSlackEventDeliveryTracker() { + const deliveredKeys = new Set(); + return { + hasDelivered(params: SlackEventDeliveryAttempt) { + const key = buildSlackEventDeliveryKey(params); + return key ? deliveredKeys.has(key) : false; + }, + markDelivered(params: SlackEventDeliveryAttempt) { + const key = buildSlackEventDeliveryKey(params); + if (key) { + deliveredKeys.add(key); + } + }, + }; +} + +export function shouldUseStreaming(params: { + streamingEnabled: boolean; + threadTs: string | undefined; +}): boolean { + if (!params.streamingEnabled) { + return false; + } + if (!params.threadTs) { + logVerbose("slack-stream: streaming disabled — no reply thread target available"); + return false; + } + return true; +} + +export async function resolveSlackStreamRecipientTeamId(params: { + client: Pick; + token: string; + userId?: PreparedSlackMessage["message"]["user"]; + fallbackTeamId?: string; +}): Promise { + const cachedTeamId = readSlackStreamRecipientTeamCache(params); + if (cachedTeamId) { + return cachedTeamId; + } + if (params.userId) { + try { + const info = await params.client.users.info({ + token: params.token, + user: params.userId, + }); + const teamId = info.user?.team_id ?? info.user?.profile?.team; + if (teamId) { + rememberSlackStreamRecipientTeam({ ...params, teamId }); + return teamId; + } + } catch (err) { + logVerbose(`slack-stream: users.info team lookup failed (${formatErrorMessage(err)})`); + } + } + return params.fallbackTeamId; +} diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress-io.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress-io.ts new file mode 100644 index 000000000000..717fa976c040 --- /dev/null +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress-io.ts @@ -0,0 +1,23 @@ +import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { formatSlackError } from "../../errors.js"; +import type { SlackDispatchSetup } from "./dispatch-setup.js"; +import { finalizeSlackPreviewEdit } from "./preview-finalize.js"; + +export async function collapseSlackProgressReceipt(params: { + setup: Pick; + edit: Omit[0], "client" | "token" | "accountId">; +}): Promise { + const { account, ctx, slackClient } = params.setup; + try { + await finalizeSlackPreviewEdit({ + client: slackClient, + token: ctx.botToken, + accountId: account.accountId, + ...params.edit, + }); + return true; + } catch (err) { + logVerbose(`slack: progress receipt edit failed (${formatSlackError(err)})`); + return false; + } +} diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress-render.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress-render.ts new file mode 100644 index 000000000000..b92c3550646f --- /dev/null +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress-render.ts @@ -0,0 +1,83 @@ +import { + buildChannelProgressDraftLine, + type AgentPlanStep, + type ChannelProgressDraftCompositorLine, + type ChannelProgressDraftCompositorSnapshot, + type ChannelProgressDraftLine, +} from "openclaw/plugin-sdk/channel-outbound"; +import { + buildSlackProgressStreamStartChunks, + buildSlackProgressStreamUpdateChunks, +} from "../../progress-blocks.js"; + +export function resolveStructuredProgressLines( + lines: readonly ChannelProgressDraftCompositorLine[], +): ChannelProgressDraftLine[] { + return lines.map((line) => { + if (typeof line !== "string") { + return line; + } + const reasoning = line.startsWith("🧠 "); + const text = line + .replace(/^(?:🧠|💬)\s+/u, "") + .replace(/^_(.*)_$/su, "$1") + .trim(); + return { + // Reasoning snapshots replace one rolling row; text-based ids would orphan it each delta. + ...(reasoning ? { id: "reasoning" } : {}), + kind: "item", + text, + label: reasoning ? "Reasoning" : "Update", + prefix: false, + }; + }); +} + +// Native cards derive from the compositor snapshot. Empty plans fall back +// to line tasks, and reconciliation retires rows from the prior source. +export function resolveNativeProgressPlan( + snapshot: ChannelProgressDraftCompositorSnapshot, +): readonly AgentPlanStep[] | undefined { + return snapshot.plan?.length ? snapshot.plan : undefined; +} + +export function resolveNativeProgressLines( + snapshot: ChannelProgressDraftCompositorSnapshot, +): ChannelProgressDraftLine[] { + const lines = resolveStructuredProgressLines(snapshot.lines); + if (snapshot.plan?.length || !snapshot.planExplanation) { + return lines; + } + const explanationLine = buildChannelProgressDraftLine({ + event: "plan", + phase: "update", + explanation: snapshot.planExplanation, + }); + return explanationLine ? [...lines, explanationLine] : lines; +} + +export function combineProgressHeadlineAndExplanation( + headline: string | undefined, + explanation: string | undefined, +): string | undefined { + return headline && explanation && headline !== explanation + ? `${headline} — ${explanation}` + : (headline ?? explanation); +} + +export function buildNativeProgressChunks(params: { + snapshot: ChannelProgressDraftCompositorSnapshot; + streamStarted: boolean; + title?: string; + maxLineChars?: number; +}) { + const input = { + title: params.title, + lines: resolveNativeProgressLines(params.snapshot), + plan: resolveNativeProgressPlan(params.snapshot), + maxLineChars: params.maxLineChars, + }; + return params.streamStarted + ? buildSlackProgressStreamUpdateChunks(input) + : buildSlackProgressStreamStartChunks(input); +} diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts new file mode 100644 index 000000000000..6204a35f6be5 --- /dev/null +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts @@ -0,0 +1,671 @@ +import type { AgentPlanStep } from "openclaw/plugin-sdk/channel-outbound"; +import { + createChannelProgressDraftCompositor, + createChannelProgressReceiptTracker, + formatChannelProgressDraftText, + isChannelProgressDraftWorkToolName, + mergeChannelProgressDraftLine, + resolveChannelProgressDraftMaxLines, + resolveChannelProgressDraftMaxLineChars, + resolveChannelProgressDraftRender, + resolveChannelStreamingPreviewToolProgress, + resolveChannelStreamingSuppressDefaultToolProgressMessages, + type ChannelProgressDraftCompositorSnapshot, + type ChannelProgressDraftLine, +} from "openclaw/plugin-sdk/channel-outbound"; +import { danger, logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { createSlackDraftStream } from "../../draft-stream.js"; +import { formatSlackError } from "../../errors.js"; +import { SLACK_TEXT_LIMIT } from "../../limits.js"; +import { + buildSlackProgressDraftBlocks, + buildSlackProgressStreamCompletionChunks, + buildSlackProgressStreamStartChunks, + buildSlackProgressStreamUpdateChunks, + reconcileSlackNativeTaskChunks, + type SlackNativeTaskSnapshot, +} from "../../progress-blocks.js"; +import { applyAppendOnlyStreamUpdate } from "../../stream-mode.js"; +import { + appendSlackStream, + startSlackStream, + stopSlackStream, + type SlackStreamSession, +} from "../../streaming.js"; +import { escapeSlackMrkdwn } from "../mrkdwn.js"; +import { + resolveExplicitSlackProgressTitle, + resolveSlackStreamRecipientTeamId, +} from "./dispatch-helpers.js"; +import { collapseSlackProgressReceipt } from "./dispatch-progress-io.js"; +import { + buildNativeProgressChunks as buildRenderedNativeProgressChunks, + combineProgressHeadlineAndExplanation, + resolveNativeProgressLines, + resolveNativeProgressPlan, + resolveStructuredProgressLines, +} from "./dispatch-progress-render.js"; +import type { SlackDispatchSetup } from "./dispatch-setup.js"; +import type { SlackStreamingDeliveryRuntime } from "./dispatch-streaming.js"; + +export function createSlackProgressRuntime(runtimeParams: { + setup: SlackDispatchSetup; + delivery: SlackStreamingDeliveryRuntime; + resetPreviewDeliveryState: () => void; +}) { + const { setup, delivery, resetPreviewDeliveryState } = runtimeParams; + const { + account, + cfg, + ctx, + message, + prepared, + replyPlan, + runtime, + slackClient, + slackIdentity, + slackMessageMetadata, + slackStreaming, + slackStreamFallbackTeamId, + shouldUseDraftStream, + useStreaming, + previewStreamingEnabled, + } = setup; + const draftStream = shouldUseDraftStream + ? createSlackDraftStream({ + target: prepared.replyTarget, + cfg, + token: ctx.botToken, + accountId: account.accountId, + ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), + identity: slackIdentity, + ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), + maxChars: Math.min(ctx.textLimit, SLACK_TEXT_LIMIT), + resolveThreadTs: () => { + const ts = replyPlan.peekThreadTs(); + if (ts) { + delivery.usedReplyThreadTs ??= ts; + } + return ts; + }, + log: logVerbose, + warn: logVerbose, + }) + : undefined; + let hasStreamedMessage = false; + const streamMode = slackStreaming.draftMode; + const useNativeProgressStreaming = useStreaming && slackStreaming.mode === "progress"; + const progressDraftActive = Boolean(draftStream) || useNativeProgressStreaming; + const previewToolProgressEnabled = + progressDraftActive && resolveChannelStreamingPreviewToolProgress(account.config); + let shouldYieldDraftProgress: () => boolean = () => false; + const suppressDefaultToolProgressMessages = + resolveChannelStreamingSuppressDefaultToolProgressMessages(account.config, { + draftStreamActive: Boolean(draftStream) || useNativeProgressStreaming, + previewToolProgressEnabled, + previewStreamingEnabled, + }); + let previewToolProgressSuppressed = false; + let legacyPreviewToolProgressLines: ChannelProgressDraftLine[] = []; + // Last task rows emitted to the native stream; reconciliation terminalizes + // ids that drop out (plan shrinks, tool-line <-> plan source switches). + let nativeTaskState: SlackNativeTaskSnapshot = new Map(); + let appendRenderedText = ""; + let appendSourceText = ""; + let nativeProgressCompletionSent = false; + // Terminal status of the turn's final payload; completion retries and + // queued rotation must not repaint an errored turn as complete. + let nativeProgressTerminalStatus: "complete" | "error" = "complete"; + let nativeProgressChunkKey: string | undefined; + const progressReceipt = createChannelProgressReceiptTracker(); + let progressReceiptCollapsed = false; + let pendingNativeProgressReceipt: string | undefined; + const progressSeed = `${account.accountId}:${message.channel}`; + const useRichProgressDraft = + streamMode === "status_final" && resolveChannelProgressDraftRender(account.config) === "rich"; + const explicitProgressTitle = resolveExplicitSlackProgressTitle(account.config); + const progressDraftMaxLineChars = resolveChannelProgressDraftMaxLineChars(account.config); + + const waitForNativeProgressStreamStart = async (): Promise => { + if (delivery.streamSession || !delivery.nativeProgressStreamStartPromise) { + return true; + } + try { + await delivery.nativeProgressStreamStartPromise; + } catch { + delivery.streamFailed = true; + return false; + } + return !delivery.streamFailed; + }; + + const appendNativeProgressCompletion = async (isError: boolean) => { + const session = delivery.streamSession; + if (isError) { + nativeProgressTerminalStatus = "error"; + } + if (!session || nativeProgressCompletionSent) { + return; + } + const chunks = buildNativeProgressCompletionChunks(isError ? "error" : "complete"); + if (!chunks?.length) { + return; + } + try { + await appendSlackStream({ session, chunks }); + nativeProgressCompletionSent = true; + delivery.observedReplyDelivery ||= session.delivered; + } catch (err) { + delivery.streamFailed = true; + runtime.error?.( + danger(`slack-stream: native progress completion failed: ${formatSlackError(err)}`), + ); + } + }; + + const buildNativeProgressChunks = (snapshot: ChannelProgressDraftCompositorSnapshot) => + buildRenderedNativeProgressChunks({ + snapshot, + streamStarted: Boolean(delivery.streamSession), + title: combineProgressHeadlineAndExplanation( + explicitProgressTitle ?? snapshot.statusHeadline, + snapshot.planExplanation, + ), + maxLineChars: progressDraftMaxLineChars, + }); + + const resolveNativeProgressTitle = (snapshot: ChannelProgressDraftCompositorSnapshot) => + combineProgressHeadlineAndExplanation( + explicitProgressTitle ?? snapshot.statusHeadline, + snapshot.planExplanation, + ); + + const markNativeProgressDelivered = (session: SlackStreamSession, threadTs?: string) => { + if (session.delivered) { + delivery.observedReplyDelivery = true; + } + if (threadTs) { + delivery.usedReplyThreadTs ??= threadTs; + delivery.rememberDeliveredThreadTs("block", threadTs); + } + }; + + const startNativeProgressStream = async ( + chunks: NonNullable>, + chunkKey: string, + ) => { + const streamThreadTs = replyPlan.nextThreadTs(); + if (!streamThreadTs) { + logVerbose( + "slack-stream: no reply thread target for native progress stream start, falling back", + ); + delivery.streamFailed = true; + return; + } + delivery.nativeProgressStreamThreadTs = streamThreadTs; + const startPromise = (async () => { + const session = await startSlackStream({ + client: slackClient, + channel: message.channel, + threadTs: streamThreadTs, + chunks, + taskDisplayMode: "plan", + ...(slackIdentity ? { identity: slackIdentity } : {}), + teamId: await resolveSlackStreamRecipientTeamId({ + client: slackClient, + token: ctx.botToken, + userId: message.user, + fallbackTeamId: slackStreamFallbackTeamId, + }), + userId: message.user, + }); + delivery.streamSession = session; + return session; + })(); + delivery.nativeProgressStreamStartPromise = startPromise; + let startedSession: SlackStreamSession | null; + try { + startedSession = await startPromise; + } finally { + if (delivery.nativeProgressStreamStartPromise === startPromise) { + delivery.nativeProgressStreamStartPromise = null; + } + } + if (startedSession) { + markNativeProgressDelivered(startedSession, streamThreadTs); + } + nativeProgressChunkKey = chunkKey; + replyPlan.markSent(); + }; + + const appendNativeProgressStream = async ( + chunks: NonNullable>, + chunkKey: string, + ) => { + if (!delivery.streamSession) { + return; + } + await appendSlackStream({ session: delivery.streamSession, chunks }); + markNativeProgressDelivered(delivery.streamSession); + nativeProgressChunkKey = chunkKey; + }; + + const updateNativeProgressStream = async () => { + const snapshot = progressDraft.getSnapshot(); + const progressLines = resolveNativeProgressLines(snapshot); + const hasRetirableNativeTasks = [...nativeTaskState.values()].some( + (task) => task.status !== "complete" && task.status !== "error", + ); + if ( + !useNativeProgressStreaming || + delivery.streamFailed || + (progressLines.length === 0 && + !snapshot.plan?.length && + !snapshot.statusHeadline && + !explicitProgressTitle && + !hasRetirableNativeTasks) + ) { + return; + } + const canContinue = await waitForNativeProgressStreamStart(); + if (!canContinue) { + return; + } + const reconciled = reconcileSlackNativeTaskChunks({ + previousTasks: nativeTaskState, + chunks: buildNativeProgressChunks(snapshot), + }); + const chunks = reconciled.chunks; + if (!chunks?.length) { + return; + } + const chunkKey = JSON.stringify(chunks); + if (chunkKey === nativeProgressChunkKey) { + return; + } + try { + if (!delivery.streamSession) { + await startNativeProgressStream(chunks, chunkKey); + } else { + await appendNativeProgressStream(chunks, chunkKey); + } + // Commit only after Slack accepted the chunks; a failed emit must retry + // the same reconciliation against the previous snapshot. + nativeTaskState = reconciled.tasks; + } catch (err) { + runtime.error?.( + danger( + `slack-stream: native progress stream failed: ${formatSlackError(err)}, falling back`, + ), + ); + delivery.streamFailed = true; + } + }; + + const resetProgressTurnState = () => { + progressReceipt.reset(); + progressReceiptCollapsed = false; + pendingNativeProgressReceipt = undefined; + }; + + const collapseProgressReceipt = async ( + params: Parameters[0]["edit"], + ) => { + progressReceiptCollapsed = await collapseSlackProgressReceipt({ setup, edit: params }); + }; + + const progressDraft = createChannelProgressDraftCompositor({ + entry: account.config, + mode: slackStreaming.mode, + active: progressDraftActive && streamMode === "status_final", + seed: progressSeed, + formatLine: escapeSlackMrkdwn, + reasoningLinePrefix: "🧠 ", + commentaryLinePrefix: "💬 ", + reasoningGate: previewToolProgressEnabled, + commentaryItalics: false, + updateOnLineChange: useNativeProgressStreaming || useRichProgressDraft, + update: async (previewText, options) => { + if (useNativeProgressStreaming) { + await updateNativeProgressStream(); + return; + } + if (!draftStream) { + return; + } + const snapshot = progressDraft.getSnapshot(); + const structuredLines = resolveStructuredProgressLines(options?.lines ?? snapshot.lines); + const richNarration = combineProgressHeadlineAndExplanation( + snapshot.statusHeadline, + snapshot.planExplanation, + ); + const richProgressBlocks = useRichProgressDraft + ? buildSlackProgressDraftBlocks({ + title: explicitProgressTitle, + lines: structuredLines, + plan: snapshot.plan, + narration: richNarration, + maxLineChars: progressDraftMaxLineChars, + }) + : undefined; + draftStream.update( + useRichProgressDraft && richProgressBlocks + ? { text: previewText, blocks: richProgressBlocks } + : previewText, + ); + hasStreamedMessage = true; + if (options?.flush) { + await draftStream.flush(); + } + }, + }); + const commentaryProgressEnabled = progressDraft.commentaryProgressEnabled; + + const buildNativeProgressCompletionChunks = (finalInProgressStatus: "complete" | "error") => { + const snapshot = progressDraft.getSnapshot(); + const lines = resolveNativeProgressLines(snapshot); + const hasRetirableNativeTasks = [...nativeTaskState.values()].some( + (task) => task.status !== "complete" && task.status !== "error", + ); + if (lines.length === 0 && !snapshot.plan?.length && !hasRetirableNativeTasks) { + return undefined; + } + return reconcileSlackNativeTaskChunks({ + previousTasks: nativeTaskState, + chunks: buildSlackProgressStreamCompletionChunks({ + title: resolveNativeProgressTitle(snapshot), + lines, + plan: resolveNativeProgressPlan(snapshot), + maxLineChars: progressDraftMaxLineChars, + finalInProgressStatus, + }), + }).chunks; + }; + + const finishNativeProgressTurn = async ( + completionChunks: ReturnType, + ) => { + if (delivery.nativeProgressStreamStartPromise) { + await delivery.nativeProgressStreamStartPromise.catch(() => null); + } + const session = delivery.streamSession; + if (session && !session.stopped) { + try { + if (completionChunks?.length) { + nativeProgressCompletionSent = true; + } + const stopResult = await stopSlackStream({ + session, + ...(completionChunks?.length ? { chunks: completionChunks } : {}), + ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), + }); + delivery.acknowledgeStoppedStreamedDeliveries(session, stopResult?.messageId); + if (pendingNativeProgressReceipt && stopResult?.messageId) { + await collapseProgressReceipt({ + channelId: session.channel, + messageId: stopResult.messageId, + text: pendingNativeProgressReceipt, + threadTs: session.threadTs, + }); + } + } catch (err) { + const error = formatSlackError(err); + // stopSlackStream makes the one-shot session terminal before throwing. + // Settle delivery bookkeeping before releasing that handle. + delivery.emitAcknowledgedStreamedDeliveries(); + delivery.emitFailedPendingStreamedDeliveries(error); + logVerbose(`slack-stream: failed to rotate native progress stream (${error})`); + } + } + delivery.streamSession = null; + delivery.nativeProgressStreamStartPromise = null; + delivery.nativeProgressStreamThreadTs = undefined; + delivery.streamFailed = false; + }; + + const pushPlanProgress = async (steps?: AgentPlanStep[], explanation?: string) => { + if (streamMode === "status_final") { + await progressDraft.pushPlanProgress(steps, { explanation }); + return; + } + if (previewToolProgressSuppressed || !draftStream) { + return; + } + const text = formatChannelProgressDraftText({ + entry: account.config, + lines: legacyPreviewToolProgressLines, + seed: progressSeed, + formatLine: escapeSlackMrkdwn, + narration: explanation, + plan: steps, + }); + if (text) { + draftStream.update(text); + hasStreamedMessage = true; + } + }; + + const pushPreviewProgress = async ( + line?: ChannelProgressDraftLine, + options?: { toolName?: string }, + ) => { + if (!draftStream && !useNativeProgressStreaming) { + return; + } + if (options?.toolName !== undefined && !isChannelProgressDraftWorkToolName(options.toolName)) { + return; + } + const normalized = line?.text.replace(/\s+/g, " ").trim(); + if (streamMode === "status_final") { + if (!line || !normalized) { + await progressDraft.noteActivity(); + return; + } + await progressDraft.pushToolProgress(line, options); + return; + } + if ( + !line || + !normalized || + !draftStream || + !previewToolProgressEnabled || + previewToolProgressSuppressed + ) { + return; + } + const nextLines = mergeChannelProgressDraftLine(legacyPreviewToolProgressLines, line, { + maxLines: resolveChannelProgressDraftMaxLines(account.config), + }); + if (nextLines === legacyPreviewToolProgressLines) { + return; + } + legacyPreviewToolProgressLines = nextLines; + draftStream.update( + formatChannelProgressDraftText({ + entry: account.config, + lines: legacyPreviewToolProgressLines, + seed: progressSeed, + formatLine: escapeSlackMrkdwn, + }), + ); + hasStreamedMessage = true; + }; + + const updateDraftFromPartial = (text?: string) => { + const trimmed = text?.trimEnd(); + if (!trimmed) { + return; + } + + if (streamMode === "append") { + previewToolProgressSuppressed = true; + legacyPreviewToolProgressLines = []; + const next = applyAppendOnlyStreamUpdate({ + incoming: trimmed, + rendered: appendRenderedText, + source: appendSourceText, + }); + appendRenderedText = next.rendered; + appendSourceText = next.source; + if (!next.changed) { + return; + } + draftStream?.update(next.rendered); + hasStreamedMessage = true; + return; + } + + if (streamMode === "status_final") { + return; + } + + previewToolProgressSuppressed = true; + legacyPreviewToolProgressLines = []; + draftStream?.update(trimmed); + hasStreamedMessage = true; + }; + const pushReasoningProgress = async (payload?: { + text?: string; + isReasoningSnapshot?: boolean; + }) => { + if (!payload?.text) { + return; + } + if (streamMode !== "status_final") { + const normalized = progressDraft + .mergeReasoningProgress(payload.text, { + snapshot: payload.isReasoningSnapshot === true, + }) + .replace(/^_(.*)_$/su, "$1") + .trim(); + if (!normalized) { + return; + } + await pushPreviewProgress({ + id: "reasoning", + kind: "item", + text: normalized, + label: "Reasoning", + }); + return; + } + progressReceipt.noteReasoning(); + await progressDraft.pushReasoningProgress(payload.text, { + snapshot: payload.isReasoningSnapshot === true, + }); + }; + const resetDraftDeliveryState = () => { + hasStreamedMessage = false; + appendRenderedText = ""; + appendSourceText = ""; + }; + const resetDraftProgressState = () => { + progressDraft.resetReasoningProgress(); + previewToolProgressSuppressed = false; + legacyPreviewToolProgressLines = []; + }; + const beginNewProgressTurn = async (options?: { force?: boolean }) => { + const completionChunks = + useNativeProgressStreaming && !nativeProgressCompletionSent + ? buildNativeProgressCompletionChunks(nativeProgressTerminalStatus) + : undefined; + if (!progressDraft.beginNewTurn(options)) { + return false; + } + // Native messages are one-shot streams. Stop the prior turn before the + // reset compositor can publish the queued turn's first snapshot. + if (useNativeProgressStreaming) { + await finishNativeProgressTurn(completionChunks); + } else { + draftStream?.forceNewMessage(); + } + resetProgressTurnState(); + nativeTaskState = new Map(); + nativeProgressCompletionSent = false; + nativeProgressTerminalStatus = "complete"; + nativeProgressChunkKey = undefined; + // A re-armed turn is a new visible reply: it must not dedupe against or + // inherit delivery state from the settled turn (mirrors queued admission). + resetPreviewDeliveryState(); + delivery.resetDeliveryTracker(); + progressReceiptCollapsed = false; + return true; + }; + const onDraftBoundary = + !shouldUseDraftStream && !useNativeProgressStreaming + ? undefined + : async () => { + if (streamMode === "status_final") { + await beginNewProgressTurn(); + return; + } + if (hasStreamedMessage) { + draftStream?.forceNewMessage(); + } + resetDraftDeliveryState(); + resetDraftProgressState(); + }; + + const onQueuedFollowupAdmitted = + !shouldUseDraftStream && !useNativeProgressStreaming + ? undefined + : async () => { + // A queued input is a new visible reply even though it drains through + // this turn's callbacks. Do not let it edit or dedupe against this run. + await draftStream?.flush(); + resetPreviewDeliveryState(); + if (streamMode === "status_final") { + await beginNewProgressTurn({ force: true }); + } else { + draftStream?.forceNewMessage(); + } + delivery.resetDeliveryTracker(); + resetDraftDeliveryState(); + resetDraftProgressState(); + }; + + return { + draftStream, + streamMode, + useNativeProgressStreaming, + progressDraftActive, + previewToolProgressEnabled, + suppressDefaultToolProgressMessages, + progressDraft, + commentaryProgressEnabled, + progressReceipt, + get progressReceiptCollapsed() { + return progressReceiptCollapsed; + }, + get pendingNativeProgressReceipt() { + return pendingNativeProgressReceipt; + }, + set pendingNativeProgressReceipt(value: string | undefined) { + pendingNativeProgressReceipt = value; + }, + get nativeProgressCompletionSent() { + return nativeProgressCompletionSent; + }, + set nativeProgressCompletionSent(value: boolean) { + nativeProgressCompletionSent = value; + }, + get nativeProgressTerminalStatus() { + return nativeProgressTerminalStatus; + }, + appendNativeProgressCompletion, + beginNewProgressTurn, + buildNativeProgressCompletionChunks, + collapseProgressReceipt, + onDraftBoundary, + onQueuedFollowupAdmitted, + pushPlanProgress, + pushPreviewProgress, + pushReasoningProgress, + updateDraftFromPartial, + waitForNativeProgressStreamStart, + setShouldYieldDraftProgress: (value: () => boolean) => { + shouldYieldDraftProgress = value; + }, + shouldYieldDraftProgress: () => shouldYieldDraftProgress(), + }; +} diff --git a/extensions/slack/src/monitor/message-handler/dispatch-setup.ts b/extensions/slack/src/monitor/message-handler/dispatch-setup.ts new file mode 100644 index 000000000000..2dfa7524f3ce --- /dev/null +++ b/extensions/slack/src/monitor/message-handler/dispatch-setup.ts @@ -0,0 +1,359 @@ +import { + createStatusReactionController, + DEFAULT_TIMING, + logAckFailure, + logTypingFailure, + type StatusReactionAdapter, +} from "openclaw/plugin-sdk/channel-feedback"; +import { + createChannelMessageReplyPipeline, + resolveAgentOutboundIdentity, + resolveChannelMessageSourceReplyDeliveryMode, + resolveChannelStreamingBlockEnabled, + resolveChannelStreamingNativeTransport, +} from "openclaw/plugin-sdk/channel-outbound"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { resolveInboundLastRouteSessionKey } from "openclaw/plugin-sdk/routing"; +import { danger, logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { resolvePinnedMainDmOwnerFromAllowlist } from "openclaw/plugin-sdk/security-runtime"; +import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { reactSlackMessage, removeSlackReaction } from "../../actions.js"; +import { formatSlackError } from "../../errors.js"; +import { + compileSlackInteractiveReplies, + isSlackInteractiveRepliesEnabled, +} from "../../interactive-replies.js"; +import { resolveSlackStreamingConfig } from "../../stream-mode.js"; +import { resolveSlackThreadTargets } from "../../threading.js"; +import { normalizeSlackAllowOwnerEntry } from "../allow-list.js"; +import { resolveStorePath, updateLastRoute } from "../config.runtime.js"; +import { createSlackReplyDeliveryPlan } from "../replies.js"; +import { + isSlackStreamingEnabled, + resolveSlackDisableBlockStreaming, + resolveSlackNativeProgressTaskCards, + resolveSlackStreamingThreadHint, + shouldEnableSlackPreviewStreaming, + shouldInitializeSlackDraftStream, + shouldUseStreaming, +} from "./dispatch-helpers.js"; +import type { PreparedSlackMessage } from "./types.js"; + +export async function createSlackDispatchSetup(prepared: PreparedSlackMessage) { + const { ctx, account, message, route } = prepared; + const slackClient = prepared.eventScope?.client ?? ctx.app.client; + const slackStreamFallbackTeamId = prepared.eventScope?.teamId ?? ctx.teamId; + const cfg = ctx.cfg; + const runtime = ctx.runtime; + + // Resolve agent identity for Slack chat:write.customize overrides. + const outboundIdentity = resolveAgentOutboundIdentity(cfg, route.agentId); + const slackIdentity = outboundIdentity + ? { + username: outboundIdentity.name, + iconUrl: outboundIdentity.avatarUrl, + iconEmoji: outboundIdentity.emoji, + } + : prepared.relayIdentity; + + if (prepared.isDirectMessage) { + const sessionCfg = cfg.session; + const storePath = resolveStorePath(sessionCfg?.store, { + agentId: route.agentId, + }); + const pinnedMainDmOwner = resolvePinnedMainDmOwnerFromAllowlist({ + dmScope: cfg.session?.dmScope, + allowFrom: ctx.allowFrom, + normalizeEntry: normalizeSlackAllowOwnerEntry, + }); + const senderRecipient = normalizeOptionalLowercaseString(message.user); + const inboundLastRouteSessionKey = resolveInboundLastRouteSessionKey({ + route, + sessionKey: prepared.ctxPayload.SessionKey ?? route.sessionKey, + }); + const skipMainUpdate = + inboundLastRouteSessionKey === route.mainSessionKey && + pinnedMainDmOwner && + senderRecipient && + normalizeOptionalLowercaseString(pinnedMainDmOwner) !== senderRecipient; + if (skipMainUpdate) { + logVerbose( + `slack: skip main-session last route for ${senderRecipient} (pinned owner ${pinnedMainDmOwner})`, + ); + } else { + await updateLastRoute({ + storePath, + sessionKey: inboundLastRouteSessionKey, + deliveryContext: { + channel: "slack", + to: `user:${message.user}`, + accountId: route.accountId, + threadId: prepared.ctxPayload.MessageThreadId ?? prepared.ctxPayload.TransportThreadId, + }, + ctx: prepared.ctxPayload, + }); + } + } + + const threadTargets = resolveSlackThreadTargets({ + message, + replyToMode: prepared.replyToMode, + }); + const forcedReplyThreadTs = prepared.forcedReplyThreadTs; + const slackMessageMetadata = prepared.slackMessageMetadata; + const statusThreadTs = forcedReplyThreadTs ?? threadTargets.statusThreadTs; + const isThreadReply = threadTargets.isThreadReply; + const replyDeliveryMode = forcedReplyThreadTs ? "off" : prepared.replyToMode; + const sourceReplyDeliveryMode = resolveChannelMessageSourceReplyDeliveryMode({ + cfg, + ctx: prepared.ctxPayload, + }); + const sourceRepliesAreToolOnly = sourceReplyDeliveryMode === "message_tool_only"; + const suppressRoomEventTyping = prepared.ctxPayload.InboundEventKind === "room_event"; + + // Shared context for the `message_sent` plugin hook emitted on each delivered + // reply (both the `deliverReplies` paths and the native-streaming finalizer). + const messageSentHookTarget = + prepared.ctxPayload.OriginatingTo ?? prepared.ctxPayload.To ?? prepared.replyTarget; + const messageSentHookContext = { + sessionKeyForInternalHooks: prepared.ctxPayload.SessionKey ?? route.sessionKey, + isGroup: prepared.isRoomish, + groupId: prepared.isRoomish ? message.channel : undefined, + }; + const messageSentDeliveryHookContext = { + ...messageSentHookContext, + messageSentHookTarget, + }; + + const reactionMessageTs = prepared.ackReactionMessageTs; + const messageTs = message.ts ?? message.event_ts; + const incomingThreadTs = message.thread_ts; + let didSetStatus = false; + const statusReactionsEnabled = + prepared.ctxPayload.InboundEventKind !== "room_event" && + Boolean(prepared.ackReactionPromise) && + Boolean(reactionMessageTs) && + cfg.messages?.statusReactions?.enabled === true; + const slackStatusAdapter: StatusReactionAdapter = { + setReaction: async (emoji) => { + await reactSlackMessage(message.channel, reactionMessageTs ?? "", emoji, { + token: ctx.botToken, + client: slackClient, + }).catch((err: unknown) => { + if (formatErrorMessage(err).includes("already_reacted")) { + return; + } + throw err; + }); + }, + removeReaction: async (emoji) => { + await removeSlackReaction(message.channel, reactionMessageTs ?? "", emoji, { + token: ctx.botToken, + client: slackClient, + }).catch((err: unknown) => { + if (formatErrorMessage(err).includes("no_reaction")) { + return; + } + throw err; + }); + }, + }; + const statusReactions = createStatusReactionController({ + enabled: statusReactionsEnabled, + adapter: slackStatusAdapter, + initialEmoji: prepared.ackReactionValue || "eyes", + emojis: undefined, + timing: DEFAULT_TIMING, + onError: (err) => { + logAckFailure({ + log: logVerbose, + channel: "slack", + target: `${message.channel}/${message.ts}`, + error: err, + }); + }, + }); + + if (statusReactionsEnabled) { + void statusReactions.setQueued(); + } + + // Shared mutable ref for "replyToMode=first". Both tool + auto-reply flows + // mark this to ensure only the first reply is threaded. + const hasRepliedRef = { value: false }; + const replyPlan = createSlackReplyDeliveryPlan({ + replyToMode: replyDeliveryMode, + incomingThreadTs: forcedReplyThreadTs ?? incomingThreadTs, + messageTs, + hasRepliedRef, + isThreadReply: Boolean(forcedReplyThreadTs) || isThreadReply, + }); + + const typingTarget = statusThreadTs ? `${message.channel}/${statusThreadTs}` : message.channel; + const typingReaction = ctx.typingReaction; + const { onModelSelected, ...replyPipeline } = createChannelMessageReplyPipeline({ + cfg, + agentId: route.agentId, + channel: "slack", + accountId: route.accountId, + transformReplyPayload: (payload) => { + if (payload.isReasoning === true) { + return null; + } + return isSlackInteractiveRepliesEnabled({ cfg, accountId: route.accountId }) + ? compileSlackInteractiveReplies(payload) + : payload; + }, + typing: { + start: async () => { + didSetStatus = true; + await ctx.setSlackThreadStatus({ + channelId: message.channel, + threadTs: statusThreadTs, + status: "is typing...", + eventScope: prepared.eventScope, + }); + if (typingReaction && message.ts) { + await reactSlackMessage(message.channel, message.ts, typingReaction, { + token: ctx.botToken, + client: slackClient, + }).catch((err: unknown) => { + logVerbose(`slack send: typing reaction failed: ${formatSlackError(err)}`); + }); + } + }, + stop: async () => { + if (!didSetStatus) { + return; + } + didSetStatus = false; + await ctx.setSlackThreadStatus({ + channelId: message.channel, + threadTs: statusThreadTs, + status: "", + eventScope: prepared.eventScope, + }); + if (typingReaction && message.ts) { + await removeSlackReaction(message.channel, message.ts, typingReaction, { + token: ctx.botToken, + client: slackClient, + }).catch((err: unknown) => { + logVerbose(`slack send: typing reaction removal failed: ${formatSlackError(err)}`); + }); + } + }, + onStartError: (err) => { + logTypingFailure({ + log: (messageValue) => runtime.error?.(danger(messageValue)), + channel: "slack", + action: "start", + target: typingTarget, + error: err, + }); + }, + onStopError: (err) => { + logTypingFailure({ + log: (messageLocal) => runtime.error?.(danger(messageLocal)), + channel: "slack", + action: "stop", + target: typingTarget, + error: err, + }); + }, + }, + }); + + const slackStreaming = resolveSlackStreamingConfig({ + streaming: account.config.streaming, + nativeStreaming: resolveChannelStreamingNativeTransport(account.config), + }); + const streamThreadHint = + forcedReplyThreadTs ?? + resolveSlackStreamingThreadHint({ + replyToMode: replyDeliveryMode, + incomingThreadTs, + messageTs, + isThreadReply, + }); + const previewStreamingEnabled = + !sourceRepliesAreToolOnly && + shouldEnableSlackPreviewStreaming({ + mode: slackStreaming.mode, + }); + const hasSlackCustomIdentity = Boolean( + slackIdentity?.username || slackIdentity?.iconUrl || slackIdentity?.iconEmoji, + ); + const streamingEnabled = + !sourceRepliesAreToolOnly && + isSlackStreamingEnabled({ + mode: slackStreaming.mode, + nativeStreaming: slackStreaming.nativeStreaming, + nativeProgressTaskCards: resolveSlackNativeProgressTaskCards(account.config), + }); + const useStreaming = shouldUseStreaming({ + streamingEnabled, + threadTs: streamThreadHint, + }); + // chat.update cannot preserve custom authorship. Use native streaming when + // possible; otherwise keep identity intact with one final postMessage. + const shouldUseDraftStream = + !hasSlackCustomIdentity && + shouldInitializeSlackDraftStream({ + previewStreamingEnabled, + useStreaming, + }); + const blockStreamingEnabled = resolveChannelStreamingBlockEnabled(account.config); + const disableBlockStreaming = sourceRepliesAreToolOnly + ? true + : resolveSlackDisableBlockStreaming({ + useStreaming, + shouldUseDraftStream, + blockStreamingEnabled, + }); + + const onSlackDeliveryError = (err: unknown, info: { kind: string }) => { + runtime.error?.(danger(`slack ${info.kind} reply failed: ${formatSlackError(err)}`)); + replyPipeline.typingCallbacks?.onIdle?.(); + }; + + return { + prepared, + ctx, + account, + message, + route, + slackClient, + slackStreamFallbackTeamId, + cfg, + runtime, + slackIdentity, + forcedReplyThreadTs, + slackMessageMetadata, + statusThreadTs, + isThreadReply, + replyDeliveryMode, + sourceReplyDeliveryMode, + sourceRepliesAreToolOnly, + suppressRoomEventTyping, + messageSentHookTarget, + messageSentHookContext, + messageSentDeliveryHookContext, + incomingThreadTs, + messageTs, + statusReactionsEnabled, + statusReactions, + hasRepliedRef, + replyPlan, + onModelSelected, + replyPipeline, + slackStreaming, + streamThreadHint, + previewStreamingEnabled, + shouldUseDraftStream, + disableBlockStreaming, + useStreaming, + onSlackDeliveryError, + }; +} + +export type SlackDispatchSetup = Awaited>; diff --git a/extensions/slack/src/monitor/message-handler/dispatch-streaming.ts b/extensions/slack/src/monitor/message-handler/dispatch-streaming.ts new file mode 100644 index 000000000000..b628f4a3ecec --- /dev/null +++ b/extensions/slack/src/monitor/message-handler/dispatch-streaming.ts @@ -0,0 +1,607 @@ +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload"; +import type { ReplyDispatchKind, ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; +import { danger, logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { formatSlackError } from "../../errors.js"; +import { emitSlackMessageSentHooks } from "../../message-sent-hook.js"; +import { resolveSlackReplyRenderPlan } from "../../reply-blocks.js"; +import { + appendSlackStream, + markSlackStreamFallbackDelivered, + SlackStreamNotDeliveredError, + startSlackStream, + stopSlackStream, + type SlackStreamSession, +} from "../../streaming.js"; +import { + deliverReplies, + readSlackReplyBlocks, + resolveDeliveredSlackReplyThreadTs, +} from "../replies.js"; +import { + createSlackEventDeliveryTracker, + resolveSlackStreamRecipientTeamId, + type SlackEventDeliveryAttempt, +} from "./dispatch-helpers.js"; +import type { SlackDispatchSetup } from "./dispatch-setup.js"; + +export function createSlackStreamingDeliveryRuntime(setup: SlackDispatchSetup) { + const { + account, + ctx, + forcedReplyThreadTs, + isThreadReply, + message, + messageSentDeliveryHookContext, + messageSentHookContext, + messageSentHookTarget, + prepared, + replyDeliveryMode, + replyPlan, + runtime, + slackClient, + slackIdentity, + slackMessageMetadata, + slackStreamFallbackTeamId, + } = setup; + const state = { + streamSession: null as SlackStreamSession | null, + nativeProgressStreamStartPromise: null as Promise | null, + nativeProgressStreamThreadTs: undefined as string | undefined, + streamFailed: false, + usedReplyThreadTs: undefined as string | undefined, + usedBlockReplyThreadTs: undefined as string | undefined, + observedReplyDelivery: false, + observedFinalReplyDelivery: false, + }; + // Reply payloads routed through the native text stream. Track Slack + // acknowledgement separately because a later buffered suffix can fall back + // after earlier payloads are already visible. + const streamedDeliveries: Array<{ + kind: ReplyDispatchKind; + content: string; + acknowledged: boolean; + outcome?: "success" | "failure"; + }> = []; + const streamedFailuresOwnedByDispatcher: Record = { + tool: 0, + block: 0, + final: 0, + }; + const refreshStreamedAcknowledgements = (session: SlackStreamSession) => { + if (session.pendingText.length === 0) { + for (const delivery of streamedDeliveries) { + delivery.acknowledged = true; + } + } + }; + const recordStreamedDelivery = (kind: ReplyDispatchKind, content: string) => { + const delivery: (typeof streamedDeliveries)[number] = { + kind, + content, + acknowledged: false, + }; + streamedDeliveries.push(delivery); + return delivery; + }; + const rememberStreamedDelivery = ( + kind: ReplyDispatchKind, + content: string, + session: SlackStreamSession, + ) => { + recordStreamedDelivery(kind, content); + refreshStreamedAcknowledgements(session); + }; + const emitAcknowledgedStreamedDeliveries = (messageId?: string) => { + for (const delivery of streamedDeliveries) { + if (!delivery.acknowledged || delivery.outcome) { + continue; + } + emitSlackMessageSentHooks({ + ...messageSentHookContext, + to: messageSentHookTarget, + accountId: account.accountId, + content: delivery.content, + success: true, + ...(messageId ? { messageId } : {}), + }); + delivery.outcome = "success"; + } + }; + const acknowledgeStoppedStreamedDeliveries = ( + session: SlackStreamSession, + messageId?: string, + ) => { + refreshStreamedAcknowledgements(session); + for (const delivery of streamedDeliveries) { + delivery.acknowledged = true; + } + emitAcknowledgedStreamedDeliveries(messageId); + }; + const emitFailedPendingStreamedDeliveries = (error: string) => { + for (const delivery of streamedDeliveries) { + if (delivery.acknowledged || delivery.outcome) { + continue; + } + emitSlackMessageSentHooks({ + ...messageSentHookContext, + to: messageSentHookTarget, + accountId: account.accountId, + content: delivery.content, + success: false, + error, + }); + delivery.outcome = "failure"; + } + }; + const emitSuccessfulPendingStreamedDeliveries = (messageId?: string) => { + for (const delivery of streamedDeliveries) { + if (delivery.acknowledged || delivery.outcome) { + continue; + } + emitSlackMessageSentHooks({ + ...messageSentHookContext, + to: messageSentHookTarget, + accountId: account.accountId, + content: delivery.content, + success: true, + ...(messageId ? { messageId } : {}), + }); + delivery.outcome = "success"; + } + }; + let deliveryTracker = createSlackEventDeliveryTracker(); + const markPreviewPayloadDelivered = (params: { + kind: ReplyDispatchKind; + payload: ReplyPayload; + threadTs: string | undefined; + }) => { + deliveryTracker.markDelivered(params); + // Single-use reply modes move later same-turn payloads off the preview + // thread, so protect both delivery keys from duplicates. + const nextThreadTs = replyPlan.peekThreadTs(); + if (nextThreadTs !== params.threadTs) { + deliveryTracker.markDelivered({ ...params, threadTs: nextThreadTs }); + } + }; + const resolveDeliveryThreadTs = (params: { + kind: ReplyDispatchKind; + forcedThreadTs?: string; + }): string | undefined => { + const plannedThreadTs = params.forcedThreadTs ? undefined : replyPlan.nextThreadTs(); + return ( + params.forcedThreadTs ?? + plannedThreadTs ?? + (params.kind === "block" ? state.usedBlockReplyThreadTs : undefined) + ); + }; + const rememberDeliveredThreadTs = ( + kind: ReplyDispatchKind, + deliveredThreadTs: string | undefined, + ) => { + if (!deliveredThreadTs) { + return; + } + state.usedReplyThreadTs ??= deliveredThreadTs; + if (kind === "block") { + state.usedBlockReplyThreadTs = deliveredThreadTs; + } + }; + const deliverPendingStreamFallback = async ( + session: SlackStreamSession, + err: SlackStreamNotDeliveredError, + ): Promise => { + let fallbackError = err; + if (!session.stopped) { + try { + const stopResult = await stopSlackStream({ + session, + ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), + }); + acknowledgeStoppedStreamedDeliveries(session, stopResult.messageId); + state.observedReplyDelivery = true; + state.usedReplyThreadTs ??= session.threadTs; + return true; + } catch (stopErr) { + if (stopErr instanceof SlackStreamNotDeliveredError) { + fallbackError = stopErr; + } else { + runtime.error?.( + danger( + `slack-stream: failed to finalize buffered text before fallback: ${formatSlackError(stopErr)}`, + ), + ); + } + } + } + emitAcknowledgedStreamedDeliveries(); + // The Slack SDK still owns this text in-memory; no streaming API call has + // acknowledged it. Route through deliverReplies so pendingText that + // exceeds Slack's per-message text limit still lands (a single + // chat.postMessage would have failed with msg_too_long), and so the + // fallback respects the configured replyToMode/identity the same way + // normal replies do. + const fallbackText = fallbackError.pendingText.trim(); + if (!fallbackText) { + return false; + } + try { + await deliverReplies({ + cfg: ctx.cfg, + replies: [{ text: fallbackText } as ReplyPayload], + target: prepared.replyTarget, + token: ctx.botToken, + accountId: account.accountId, + runtime, + textLimit: ctx.textLimit, + mediaMaxBytes: ctx.mediaMaxBytes, + replyThreadTs: session.threadTs, + replyToMode: replyDeliveryMode, + ...(slackIdentity ? { identity: slackIdentity } : {}), + ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), + ...messageSentDeliveryHookContext, + deferMessageSentHooks: true, + ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), + }); + markSlackStreamFallbackDelivered(session); + if (!session.stopped) { + try { + await stopSlackStream({ + session, + ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), + }); + } catch (finalizeErr) { + runtime.error?.( + danger( + `slack-stream: failed to finalize native stream after fallback delivery: ${formatSlackError(finalizeErr)}`, + ), + ); + } + } + // The combined fallback can span multiple logical payloads and Slack + // chunks, so no single message `ts` correctly identifies every event. + emitSuccessfulPendingStreamedDeliveries(); + state.observedReplyDelivery = true; + state.usedReplyThreadTs ??= session.threadTs; + logVerbose( + `slack-stream: streamed delivery failed (${fallbackError.slackCode}); delivered ${fallbackText.length} chars via deliverReplies fallback`, + ); + return true; + } catch (postErr) { + emitFailedPendingStreamedDeliveries(formatErrorMessage(postErr)); + runtime.error?.( + danger( + `slack-stream: fallback deliverReplies failed after ${fallbackError.slackCode}: ${formatErrorMessage(postErr)}`, + ), + ); + return false; + } + }; + + const deliverNormally = async (params: { + payload: ReplyPayload; + kind: ReplyDispatchKind; + forcedThreadTs?: string; + }): Promise => { + if (params.payload.isReasoning === true) { + return undefined; + } + const replyThreadTs = resolveDeliveryThreadTs(params); + const deliveryReplyThreadTs = + replyDeliveryMode === "off" && !forcedReplyThreadTs && !isThreadReply + ? undefined + : replyThreadTs; + if ( + deliveryTracker.hasDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: deliveryReplyThreadTs, + }) + ) { + logVerbose("slack: suppressed duplicate normal delivery within the same turn"); + return deliveryReplyThreadTs; + } + await deliverReplies({ + cfg: ctx.cfg, + replies: [params.payload], + target: prepared.replyTarget, + token: ctx.botToken, + accountId: account.accountId, + runtime, + textLimit: ctx.textLimit, + mediaMaxBytes: ctx.mediaMaxBytes, + replyThreadTs: deliveryReplyThreadTs, + replyToMode: replyDeliveryMode, + ...(slackIdentity ? { identity: slackIdentity } : {}), + ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), + ...messageSentDeliveryHookContext, + ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), + }); + state.observedReplyDelivery = true; + if (params.kind === "final") { + state.observedFinalReplyDelivery = true; + } + const deliveredThreadTs = resolveDeliveredSlackReplyThreadTs({ + replyToMode: replyDeliveryMode, + payloadReplyToId: params.payload.replyToId, + replyThreadTs: deliveryReplyThreadTs, + }); + // Record the thread ts only after confirmed delivery success. + rememberDeliveredThreadTs(params.kind, deliveredThreadTs); + replyPlan.markSent(); + deliveryTracker.markDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: deliveryReplyThreadTs, + }); + return deliveryReplyThreadTs; + }; + + const deliverBufferedStreamFallback = async (params: { + session: SlackStreamSession; + err: SlackStreamNotDeliveredError; + payload: ReplyPayload; + kind: ReplyDispatchKind; + textOverride: string; + }): Promise => { + const delivered = await deliverPendingStreamFallback(params.session, params.err); + if (!delivered) { + // The reply dispatcher will charge the currently executing payload as + // failed; earlier buffered payloads need separate reconciliation below. + streamedFailuresOwnedByDispatcher[params.kind] += 1; + return false; + } + replyPlan.markSent(); + if (params.kind === "final") { + state.observedFinalReplyDelivery = true; + } + deliveryTracker.markDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: params.session.threadTs, + textOverride: params.textOverride, + }); + rememberDeliveredThreadTs(params.kind, params.session.threadTs); + return true; + }; + + const deliverWithStreaming = async (params: { + payload: ReplyPayload; + kind: ReplyDispatchKind; + }): Promise => { + if (params.payload.isReasoning === true) { + return; + } + const reply = resolveSendableOutboundReplyParts(params.payload); + const renderPlan = resolveSlackReplyRenderPlan(params.payload); + const plannedBlocks = + renderPlan.mode === "single" ? renderPlan.blocks : renderPlan.blockPart?.blocks; + if ( + state.streamFailed || + reply.hasMedia || + renderPlan.mode === "split" || + Boolean(plannedBlocks?.length) || + readSlackReplyBlocks(params.payload)?.length || + !reply.hasText + ) { + await deliverNormally({ + payload: params.payload, + kind: params.kind, + forcedThreadTs: state.streamSession?.threadTs ?? state.nativeProgressStreamThreadTs, + }); + return; + } + + const text = reply.trimmedText; + let plannedThreadTs: string | undefined; + try { + if (!state.streamSession && state.nativeProgressStreamStartPromise) { + await state.nativeProgressStreamStartPromise; + } + if (state.streamFailed) { + await deliverNormally({ + payload: params.payload, + kind: params.kind, + forcedThreadTs: state.streamSession?.threadTs ?? state.nativeProgressStreamThreadTs, + }); + return; + } + if (!state.streamSession) { + const streamThreadTs = replyPlan.nextThreadTs(); + plannedThreadTs = streamThreadTs; + if (!streamThreadTs) { + logVerbose( + "slack-stream: no reply thread target for stream start, falling back to normal delivery", + ); + state.streamFailed = true; + await deliverNormally({ + payload: params.payload, + kind: params.kind, + }); + return; + } + if ( + deliveryTracker.hasDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: streamThreadTs, + textOverride: text, + }) + ) { + logVerbose("slack-stream: suppressed duplicate stream start payload"); + return; + } + + state.streamSession = await startSlackStream({ + client: slackClient, + channel: message.channel, + threadTs: streamThreadTs, + text, + ...(slackIdentity ? { identity: slackIdentity } : {}), + teamId: await resolveSlackStreamRecipientTeamId({ + client: slackClient, + token: ctx.botToken, + userId: message.user, + fallbackTeamId: slackStreamFallbackTeamId, + }), + userId: message.user, + }); + refreshStreamedAcknowledgements(state.streamSession); + // startSlackStream may only buffer locally. Count delivery only after + // the SDK reports a real Slack response. + if (state.streamSession.delivered) { + state.observedReplyDelivery = true; + if (params.kind === "final") { + state.observedFinalReplyDelivery = true; + } + } + // Remember the reply text delivered through the text stream so the + // `message_sent` hook can fire after stopSlackStream flushes it. + // Only the text-stream path captures this; every deliverNormally branch + // already emits via deliverReplies, so capturing there would + // double-emit for the same payload. + if (text) { + rememberStreamedDelivery(params.kind, text, state.streamSession); + } + rememberDeliveredThreadTs(params.kind, streamThreadTs); + replyPlan.markSent(); + deliveryTracker.markDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: streamThreadTs, + textOverride: text, + }); + return; + } + if ( + deliveryTracker.hasDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: state.streamSession.threadTs, + textOverride: text, + }) + ) { + logVerbose("slack-stream: suppressed duplicate append payload"); + return; + } + + if (text) { + // appendSlackStream buffers text before attempting the Slack flush. + // Record first so a later successful stop can acknowledge a thrown append. + recordStreamedDelivery(params.kind, text); + } + await appendSlackStream({ + session: state.streamSession, + text: "\n" + text, + }); + refreshStreamedAcknowledgements(state.streamSession); + // appendSlackStream also buffers locally below the SDK threshold; avoid + // optimistic "done" status until Slack acknowledges a flush. + if (state.streamSession.delivered) { + state.observedReplyDelivery = true; + if (params.kind === "final") { + state.observedFinalReplyDelivery = true; + } + } + deliveryTracker.markDelivered({ + kind: params.kind, + payload: params.payload, + threadTs: state.streamSession.threadTs, + textOverride: text, + }); + } catch (err) { + if (err instanceof SlackStreamNotDeliveredError) { + state.streamFailed = true; + if (state.streamSession) { + const delivered = await deliverBufferedStreamFallback({ + session: state.streamSession, + err, + payload: params.payload, + kind: params.kind, + textOverride: text, + }); + if (delivered) { + return; + } + throw err; + } + await deliverNormally({ + payload: params.payload, + kind: params.kind, + forcedThreadTs: plannedThreadTs, + }); + return; + } + runtime.error?.( + danger(`slack-stream: streaming API call failed: ${formatSlackError(err)}, falling back`), + ); + state.streamFailed = true; + // Non-benign streaming errors leave `pendingText` populated with every + // buffered chunk since the last flush (appendSlackStream accumulates + // into pendingText BEFORE the SDK call, so the failing chunk is + // included too). Route the full buffer through the chunked fallback so + // earlier chunks aren't lost, then skip deliverNormally - pendingText + // already contains this payload's text. + if (state.streamSession && state.streamSession.pendingText) { + const bufferedFallbackErr = new SlackStreamNotDeliveredError( + state.streamSession.pendingText, + "unknown", + ); + const delivered = await deliverBufferedStreamFallback({ + session: state.streamSession, + err: bufferedFallbackErr, + payload: params.payload, + kind: params.kind, + textOverride: text, + }); + if (delivered) { + return; + } + throw err; + } + await deliverNormally({ + payload: params.payload, + kind: params.kind, + forcedThreadTs: + state.streamSession?.threadTs ?? state.nativeProgressStreamThreadTs ?? plannedThreadTs, + }); + } + }; + + const reconcileCounts = (counts: Partial>) => { + const next = { ...counts }; + let changed = false; + for (const kind of ["tool", "block", "final"] as const) { + const failedStreamedCount = streamedDeliveries.filter( + (delivery) => delivery.kind === kind && delivery.outcome === "failure", + ).length; + const additionalFailedStreamed = Math.max( + 0, + failedStreamedCount - streamedFailuresOwnedByDispatcher[kind], + ); + if (additionalFailedStreamed > 0) { + next[kind] = Math.max(0, (next[kind] ?? 0) - additionalFailedStreamed); + changed = true; + } + } + return changed ? next : counts; + }; + + return Object.assign(state, { + acknowledgeStoppedStreamedDeliveries, + deliverNormally, + deliverPendingStreamFallback, + deliverWithStreaming, + emitAcknowledgedStreamedDeliveries, + emitFailedPendingStreamedDeliveries, + hasDelivered: (params: SlackEventDeliveryAttempt) => deliveryTracker.hasDelivered(params), + markPreviewPayloadDelivered, + rememberDeliveredThreadTs, + reconcileCounts, + resetDeliveryTracker: () => { + deliveryTracker = createSlackEventDeliveryTracker(); + }, + }); +} + +export type SlackStreamingDeliveryRuntime = ReturnType; diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index abc25774c7e8..766b61f06890 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.ts @@ -1,1184 +1,84 @@ // Slack plugin module implements dispatch behavior. import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; -import { - createStatusReactionController, - DEFAULT_TIMING, - logAckFailure, - logTypingFailure, - type StatusReactionAdapter, -} from "openclaw/plugin-sdk/channel-feedback"; import { dispatchChannelInboundTurn, type InboundReplyRecordOptions, } from "openclaw/plugin-sdk/channel-inbound"; +import { hasVisibleInboundReplyDispatch } from "openclaw/plugin-sdk/channel-inbound"; import { - type ChannelBotLoopProtectionFacts, - hasVisibleInboundReplyDispatch, -} from "openclaw/plugin-sdk/channel-inbound"; -import { - createChannelMessageReplyPipeline, - defineFinalizableLivePreviewAdapter, - deliverWithFinalizableLivePreviewAdapter, - resolveChannelMessageSourceReplyDeliveryMode, -} from "openclaw/plugin-sdk/channel-outbound"; -import { resolveAgentOutboundIdentity } from "openclaw/plugin-sdk/channel-outbound"; -import { - type AgentPlanStep, buildChannelProgressDraftLine, buildChannelProgressDraftLineForEntry, - type ChannelProgressDraftCompositorLine, - type ChannelProgressDraftCompositorSnapshot, - createChannelProgressDraftCompositor, - createChannelProgressReceiptTracker, - formatChannelProgressDraftText, - isChannelProgressDraftWorkToolName, - mergeChannelProgressDraftLine, - resolveChannelProgressDraftConfig, - resolveChannelProgressDraftMaxLines, - resolveChannelProgressDraftMaxLineChars, - resolveChannelProgressDraftRender, - resolveChannelStreamingBlockEnabled, - resolveChannelStreamingNativeTransport, - resolveChannelStreamingPreviewToolProgress, - resolveChannelStreamingSuppressDefaultToolProgressMessages, - type ChannelProgressDraftLine, } from "openclaw/plugin-sdk/channel-outbound"; -import { formatErrorMessage, toErrorObject } from "openclaw/plugin-sdk/error-runtime"; -import { mergePairLoopGuardConfig } from "openclaw/plugin-sdk/pair-loop-guard-runtime"; +import { + defineFinalizableLivePreviewAdapter, + deliverWithFinalizableLivePreviewAdapter, +} from "openclaw/plugin-sdk/channel-outbound"; +import { toErrorObject } from "openclaw/plugin-sdk/error-runtime"; import { buildTtsSupplementMediaPayload, getReplyPayloadTtsSupplement, resolveSendableOutboundReplyParts, } from "openclaw/plugin-sdk/reply-payload"; -import type { ReplyDispatchKind, ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; -import { resolveInboundLastRouteSessionKey } from "openclaw/plugin-sdk/routing"; +import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; +import type { ReplyDispatchKind } from "openclaw/plugin-sdk/reply-runtime"; import { danger, logVerbose, shouldLogVerbose } from "openclaw/plugin-sdk/runtime-env"; -import { resolvePinnedMainDmOwnerFromAllowlist } from "openclaw/plugin-sdk/security-runtime"; -import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { reactSlackMessage, removeSlackReaction } from "../../actions.js"; -import { createSlackDraftStream } from "../../draft-stream.js"; import { formatSlackError } from "../../errors.js"; import { normalizeSlackOutboundText } from "../../format.js"; -import { - compileSlackInteractiveReplies, - isSlackInteractiveRepliesEnabled, -} from "../../interactive-replies.js"; -import { SLACK_TEXT_LIMIT } from "../../limits.js"; import { emitSlackMessageSentHooks } from "../../message-sent-hook.js"; -import { - buildSlackProgressDraftBlocks, - buildSlackProgressStreamCompletionChunks, - buildSlackProgressStreamStartChunks, - buildSlackProgressStreamUpdateChunks, - reconcileSlackNativeTaskChunks, - type SlackNativeTaskSnapshot, -} from "../../progress-blocks.js"; import { resolveSlackReplyRenderPlan } from "../../reply-blocks.js"; import { recordSlackThreadParticipation } from "../../sent-thread-cache.js"; -import { applyAppendOnlyStreamUpdate, resolveSlackStreamingConfig } from "../../stream-mode.js"; -import type { SlackStreamSession } from "../../streaming.js"; import { - appendSlackStream, - markSlackStreamFallbackDelivered, SlackStreamNotDeliveredError, - startSlackStream, stopSlackStream, + type SlackStreamSession, } from "../../streaming.js"; -import { resolveSlackThreadTargets } from "../../threading.js"; -import type { SlackMessageEvent } from "../../types.js"; -import { normalizeSlackAllowOwnerEntry } from "../allow-list.js"; -import { resolveStorePath, updateLastRoute } from "../config.runtime.js"; -import { escapeSlackMrkdwn } from "../mrkdwn.js"; -import { - createSlackReplyDeliveryPlan, - deliverReplies, - readSlackReplyBlocks, - resolveDeliveredSlackReplyThreadTs, - resolveSlackThreadTs, -} from "../replies.js"; +import { resolveSlackBotLoopProtection } from "./dispatch-helpers.js"; +import { createSlackProgressRuntime } from "./dispatch-progress.js"; +import { createSlackDispatchSetup } from "./dispatch-setup.js"; +import { createSlackStreamingDeliveryRuntime } from "./dispatch-streaming.js"; import { finalizeSlackPreviewEdit } from "./preview-finalize.js"; -import { resolveSlackTimestampMs } from "./timestamp.js"; import type { PreparedSlackMessage } from "./types.js"; -function resolveSlackMessageTimestampMs(message: SlackMessageEvent): number | undefined { - const ts = message.event_ts ?? message.ts; - return resolveSlackTimestampMs(ts); -} - -function resolveSlackBotLoopProtection( - prepared: PreparedSlackMessage, -): ChannelBotLoopProtectionFacts | undefined { - const senderBotId = prepared.message.bot_id; - if (!senderBotId) { - return undefined; - } - const receiverBotId = prepared.ctx.botId || prepared.ctx.botUserId; - if ( - !receiverBotId || - senderBotId === prepared.ctx.botId || - prepared.message.user === prepared.ctx.botUserId - ) { - return undefined; - } - return { - scopeId: prepared.route.accountId, - conversationId: prepared.message.channel, - senderId: senderBotId, - receiverId: receiverBotId, - config: mergePairLoopGuardConfig( - prepared.account.config.botLoopProtection, - prepared.channelConfig?.botLoopProtection, - ), - defaultsConfig: prepared.ctx.cfg.channels?.defaults?.botLoopProtection, - defaultEnabled: true, - nowMs: resolveSlackMessageTimestampMs(prepared.message), - }; -} - -function isSlackStreamingEnabled(params: { - mode: "off" | "partial" | "block" | "progress"; - nativeStreaming: boolean; - nativeProgressTaskCards?: boolean; -}): boolean { - if (params.mode === "partial") { - return params.nativeStreaming; - } - if (params.mode === "progress") { - return params.nativeStreaming && params.nativeProgressTaskCards === true; - } - return false; -} - -function shouldEnableSlackPreviewStreaming(params: { - mode: "off" | "partial" | "block" | "progress"; -}): boolean { - return params.mode !== "off"; -} - -function shouldInitializeSlackDraftStream(params: { - previewStreamingEnabled: boolean; - useStreaming: boolean; -}): boolean { - return params.previewStreamingEnabled && !params.useStreaming; -} - -function resolveSlackDisableBlockStreaming(params: { - useStreaming: boolean; - shouldUseDraftStream: boolean; - blockStreamingEnabled: boolean | undefined; -}): boolean | undefined { - if (params.useStreaming || params.shouldUseDraftStream) { - return true; - } - return typeof params.blockStreamingEnabled === "boolean" - ? !params.blockStreamingEnabled - : undefined; -} - -function resolveExplicitSlackProgressTitle( - entry: Parameters[0], -): string | undefined { - const label = resolveChannelProgressDraftConfig(entry).label; - if (typeof label !== "string") { - return undefined; - } - const trimmed = label.trim(); - return trimmed && trimmed.toLowerCase() !== "auto" ? trimmed : undefined; -} - -function resolveSlackNativeProgressTaskCards( - entry: Parameters[0], -): boolean { - const streaming = entry?.streaming; - if (!streaming || typeof streaming !== "object" || Array.isArray(streaming)) { - return false; - } - const progressConfig = (streaming as Record).progress; - return ( - Boolean(progressConfig) && - typeof progressConfig === "object" && - !Array.isArray(progressConfig) && - (progressConfig as { nativeTaskCards?: unknown }).nativeTaskCards === true - ); -} - -function resolveSlackStreamingThreadHint(params: { - replyToMode: "off" | "first" | "all" | "batched"; - incomingThreadTs: string | undefined; - messageTs: string | undefined; - isThreadReply?: boolean; -}): string | undefined { - return resolveSlackThreadTs({ - replyToMode: params.replyToMode, - incomingThreadTs: params.incomingThreadTs, - messageTs: params.messageTs, - hasReplied: false, - isThreadReply: params.isThreadReply, - }); -} - -type SlackEventDeliveryAttempt = { - kind: ReplyDispatchKind; - payload: ReplyPayload; - threadTs?: string; - textOverride?: string; -}; - -const SLACK_STREAM_RECIPIENT_TEAM_CACHE_MAX = 2000; -const slackStreamRecipientTeamCaches = new WeakMap>(); - -function getSlackStreamRecipientTeamCache(client: object): Map { - const existing = slackStreamRecipientTeamCaches.get(client); - if (existing) { - return existing; - } - const cache = new Map(); - slackStreamRecipientTeamCaches.set(client, cache); - return cache; -} - -function buildSlackEventDeliveryKey(params: SlackEventDeliveryAttempt): string | null { - const reply = resolveSendableOutboundReplyParts(params.payload, { - text: params.textOverride, - }); - const renderPlan = resolveSlackReplyRenderPlan( - params.payload, - params.textOverride ?? params.payload.text, - ); - const plannedBlocks = - renderPlan.mode === "single" ? renderPlan.blocks : renderPlan.blockPart?.blocks; - const slackBlocks = readSlackReplyBlocks(params.payload) ?? plannedBlocks; - const renderedText = renderPlan.mode === "single" ? renderPlan.text : renderPlan.fallbackText; - if (!reply.hasContent && !slackBlocks?.length && !renderedText.trim()) { - return null; - } - return JSON.stringify({ - kind: params.kind, - threadTs: params.threadTs ?? "", - replyToId: params.payload.replyToId ?? null, - text: renderedText || reply.trimmedText, - mediaUrls: reply.mediaUrls, - blocks: slackBlocks ?? null, - }); -} - -function readSlackStreamRecipientTeamCache(params: { - client: object; - fallbackTeamId?: string; - userId?: string; -}): string | undefined { - if (!params.fallbackTeamId || !params.userId) { - return undefined; - } - const cacheKey = `${params.fallbackTeamId}:${params.userId}`; - const cache = getSlackStreamRecipientTeamCache(params.client); - const cached = cache.get(cacheKey); - if (!cached) { - return undefined; - } - cache.delete(cacheKey); - cache.set(cacheKey, cached); - return cached; -} - -function rememberSlackStreamRecipientTeam(params: { - client: object; - fallbackTeamId?: string; - userId?: string; - teamId: string; -}): void { - if (!params.fallbackTeamId || !params.userId) { - return; - } - const cacheKey = `${params.fallbackTeamId}:${params.userId}`; - const cache = getSlackStreamRecipientTeamCache(params.client); - if (cache.has(cacheKey)) { - cache.delete(cacheKey); - } - cache.set(cacheKey, params.teamId); - if (cache.size > SLACK_STREAM_RECIPIENT_TEAM_CACHE_MAX) { - const oldest = cache.keys().next().value; - if (oldest) { - cache.delete(oldest); - } - } -} - -function createSlackEventDeliveryTracker() { - const deliveredKeys = new Set(); - return { - hasDelivered(params: SlackEventDeliveryAttempt) { - const key = buildSlackEventDeliveryKey(params); - return key ? deliveredKeys.has(key) : false; - }, - markDelivered(params: SlackEventDeliveryAttempt) { - const key = buildSlackEventDeliveryKey(params); - if (key) { - deliveredKeys.add(key); - } - }, - }; -} - -function shouldUseStreaming(params: { - streamingEnabled: boolean; - threadTs: string | undefined; -}): boolean { - if (!params.streamingEnabled) { - return false; - } - if (!params.threadTs) { - logVerbose("slack-stream: streaming disabled — no reply thread target available"); - return false; - } - return true; -} - -async function resolveSlackStreamRecipientTeamId(params: { - client: Pick; - token: string; - userId?: PreparedSlackMessage["message"]["user"]; - fallbackTeamId?: string; -}): Promise { - const cachedTeamId = readSlackStreamRecipientTeamCache(params); - if (cachedTeamId) { - return cachedTeamId; - } - if (params.userId) { - try { - const info = await params.client.users.info({ - token: params.token, - user: params.userId, - }); - const teamId = info.user?.team_id ?? info.user?.profile?.team; - if (teamId) { - rememberSlackStreamRecipientTeam({ ...params, teamId }); - return teamId; - } - } catch (err) { - logVerbose(`slack-stream: users.info team lookup failed (${formatErrorMessage(err)})`); - } - } - return params.fallbackTeamId; -} - export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessage) { - const { ctx, account, message, route } = prepared; - const slackClient = prepared.eventScope?.client ?? ctx.app.client; - const slackStreamFallbackTeamId = prepared.eventScope?.teamId ?? ctx.teamId; - const cfg = ctx.cfg; - const runtime = ctx.runtime; - - // Resolve agent identity for Slack chat:write.customize overrides. - const outboundIdentity = resolveAgentOutboundIdentity(cfg, route.agentId); - const slackIdentity = outboundIdentity - ? { - username: outboundIdentity.name, - iconUrl: outboundIdentity.avatarUrl, - iconEmoji: outboundIdentity.emoji, - } - : prepared.relayIdentity; - - if (prepared.isDirectMessage) { - const sessionCfg = cfg.session; - const storePath = resolveStorePath(sessionCfg?.store, { - agentId: route.agentId, - }); - const pinnedMainDmOwner = resolvePinnedMainDmOwnerFromAllowlist({ - dmScope: cfg.session?.dmScope, - allowFrom: ctx.allowFrom, - normalizeEntry: normalizeSlackAllowOwnerEntry, - }); - const senderRecipient = normalizeOptionalLowercaseString(message.user); - const inboundLastRouteSessionKey = resolveInboundLastRouteSessionKey({ - route, - sessionKey: prepared.ctxPayload.SessionKey ?? route.sessionKey, - }); - const skipMainUpdate = - inboundLastRouteSessionKey === route.mainSessionKey && - pinnedMainDmOwner && - senderRecipient && - normalizeOptionalLowercaseString(pinnedMainDmOwner) !== senderRecipient; - if (skipMainUpdate) { - logVerbose( - `slack: skip main-session last route for ${senderRecipient} (pinned owner ${pinnedMainDmOwner})`, - ); - } else { - await updateLastRoute({ - storePath, - sessionKey: inboundLastRouteSessionKey, - deliveryContext: { - channel: "slack", - to: `user:${message.user}`, - accountId: route.accountId, - threadId: prepared.ctxPayload.MessageThreadId ?? prepared.ctxPayload.TransportThreadId, - }, - ctx: prepared.ctxPayload, - }); - } - } - - const threadTargets = resolveSlackThreadTargets({ - message, - replyToMode: prepared.replyToMode, - }); - const forcedReplyThreadTs = prepared.forcedReplyThreadTs; - const slackMessageMetadata = prepared.slackMessageMetadata; - const statusThreadTs = forcedReplyThreadTs ?? threadTargets.statusThreadTs; - const isThreadReply = threadTargets.isThreadReply; - const replyDeliveryMode = forcedReplyThreadTs ? "off" : prepared.replyToMode; - const sourceReplyDeliveryMode = resolveChannelMessageSourceReplyDeliveryMode({ + const setup = await createSlackDispatchSetup(prepared); + const { + account, cfg, - ctx: prepared.ctxPayload, - }); - const sourceRepliesAreToolOnly = sourceReplyDeliveryMode === "message_tool_only"; - const suppressRoomEventTyping = prepared.ctxPayload.InboundEventKind === "room_event"; - - // Shared context for the `message_sent` plugin hook emitted on each delivered - // reply (both the `deliverReplies` paths and the native-streaming finalizer). - const messageSentHookTarget = - prepared.ctxPayload.OriginatingTo ?? prepared.ctxPayload.To ?? prepared.replyTarget; - const messageSentHookContext = { - sessionKeyForInternalHooks: prepared.ctxPayload.SessionKey ?? route.sessionKey, - isGroup: prepared.isRoomish, - groupId: prepared.isRoomish ? message.channel : undefined, - }; - const messageSentDeliveryHookContext = { - ...messageSentHookContext, - messageSentHookTarget, - }; - - const reactionMessageTs = prepared.ackReactionMessageTs; - const messageTs = message.ts ?? message.event_ts; - const incomingThreadTs = message.thread_ts; - let didSetStatus = false; - const statusReactionsEnabled = - prepared.ctxPayload.InboundEventKind !== "room_event" && - Boolean(prepared.ackReactionPromise) && - Boolean(reactionMessageTs) && - cfg.messages?.statusReactions?.enabled === true; - const slackStatusAdapter: StatusReactionAdapter = { - setReaction: async (emoji) => { - await reactSlackMessage(message.channel, reactionMessageTs ?? "", emoji, { - token: ctx.botToken, - client: slackClient, - }).catch((err: unknown) => { - if (formatErrorMessage(err).includes("already_reacted")) { - return; - } - throw err; - }); - }, - removeReaction: async (emoji) => { - await removeSlackReaction(message.channel, reactionMessageTs ?? "", emoji, { - token: ctx.botToken, - client: slackClient, - }).catch((err: unknown) => { - if (formatErrorMessage(err).includes("no_reaction")) { - return; - } - throw err; - }); - }, - }; - const statusReactions = createStatusReactionController({ - enabled: statusReactionsEnabled, - adapter: slackStatusAdapter, - initialEmoji: prepared.ackReactionValue || "eyes", - emojis: undefined, - timing: DEFAULT_TIMING, - onError: (err) => { - logAckFailure({ - log: logVerbose, - channel: "slack", - target: `${message.channel}/${message.ts}`, - error: err, - }); - }, - }); - - if (statusReactionsEnabled) { - void statusReactions.setQueued(); - } - - // Shared mutable ref for "replyToMode=first". Both tool + auto-reply flows - // mark this to ensure only the first reply is threaded. - const hasRepliedRef = { value: false }; - const replyPlan = createSlackReplyDeliveryPlan({ - replyToMode: replyDeliveryMode, - incomingThreadTs: forcedReplyThreadTs ?? incomingThreadTs, - messageTs, + ctx, + disableBlockStreaming, hasRepliedRef, - isThreadReply: Boolean(forcedReplyThreadTs) || isThreadReply, - }); - - const typingTarget = statusThreadTs ? `${message.channel}/${statusThreadTs}` : message.channel; - const typingReaction = ctx.typingReaction; - const { onModelSelected, ...replyPipeline } = createChannelMessageReplyPipeline({ - cfg, - agentId: route.agentId, - channel: "slack", - accountId: route.accountId, - transformReplyPayload: (payload) => { - if (payload.isReasoning === true) { - return null; - } - return isSlackInteractiveRepliesEnabled({ cfg, accountId: route.accountId }) - ? compileSlackInteractiveReplies(payload) - : payload; - }, - typing: { - start: async () => { - didSetStatus = true; - await ctx.setSlackThreadStatus({ - channelId: message.channel, - threadTs: statusThreadTs, - status: "is typing...", - eventScope: prepared.eventScope, - }); - if (typingReaction && message.ts) { - await reactSlackMessage(message.channel, message.ts, typingReaction, { - token: ctx.botToken, - client: slackClient, - }).catch((err: unknown) => { - logVerbose(`slack send: typing reaction failed: ${formatSlackError(err)}`); - }); - } - }, - stop: async () => { - if (!didSetStatus) { - return; - } - didSetStatus = false; - await ctx.setSlackThreadStatus({ - channelId: message.channel, - threadTs: statusThreadTs, - status: "", - eventScope: prepared.eventScope, - }); - if (typingReaction && message.ts) { - await removeSlackReaction(message.channel, message.ts, typingReaction, { - token: ctx.botToken, - client: slackClient, - }).catch((err: unknown) => { - logVerbose(`slack send: typing reaction removal failed: ${formatSlackError(err)}`); - }); - } - }, - onStartError: (err) => { - logTypingFailure({ - log: (messageValue) => runtime.error?.(danger(messageValue)), - channel: "slack", - action: "start", - target: typingTarget, - error: err, - }); - }, - onStopError: (err) => { - logTypingFailure({ - log: (messageLocal) => runtime.error?.(danger(messageLocal)), - channel: "slack", - action: "stop", - target: typingTarget, - error: err, - }); - }, + message, + messageSentHookContext, + messageSentHookTarget, + onModelSelected, + onSlackDeliveryError, + previewStreamingEnabled, + replyPipeline, + replyPlan, + route, + runtime, + slackClient, + slackMessageMetadata, + slackStreaming, + sourceReplyDeliveryMode, + statusReactions, + statusReactionsEnabled, + statusThreadTs, + suppressRoomEventTyping, + useStreaming, + } = setup; + const delivery = createSlackStreamingDeliveryRuntime(setup); + const draftPreviewCommitted = { value: false }; + const progress = createSlackProgressRuntime({ + setup, + delivery, + resetPreviewDeliveryState: () => { + draftPreviewCommitted.value = false; + delivery.observedFinalReplyDelivery = false; }, }); + const draftStream = progress.draftStream; - const slackStreaming = resolveSlackStreamingConfig({ - streaming: account.config.streaming, - nativeStreaming: resolveChannelStreamingNativeTransport(account.config), - }); - const streamThreadHint = - forcedReplyThreadTs ?? - resolveSlackStreamingThreadHint({ - replyToMode: replyDeliveryMode, - incomingThreadTs, - messageTs, - isThreadReply, - }); - const previewStreamingEnabled = - !sourceRepliesAreToolOnly && - shouldEnableSlackPreviewStreaming({ - mode: slackStreaming.mode, - }); - const hasSlackCustomIdentity = Boolean( - slackIdentity?.username || slackIdentity?.iconUrl || slackIdentity?.iconEmoji, - ); - const streamingEnabled = - !sourceRepliesAreToolOnly && - isSlackStreamingEnabled({ - mode: slackStreaming.mode, - nativeStreaming: slackStreaming.nativeStreaming, - nativeProgressTaskCards: resolveSlackNativeProgressTaskCards(account.config), - }); - const useStreaming = shouldUseStreaming({ - streamingEnabled, - threadTs: streamThreadHint, - }); - // chat.update cannot preserve custom authorship. Use native streaming when - // possible; otherwise keep identity intact with one final postMessage. - const shouldUseDraftStream = - !hasSlackCustomIdentity && - shouldInitializeSlackDraftStream({ - previewStreamingEnabled, - useStreaming, - }); - const blockStreamingEnabled = resolveChannelStreamingBlockEnabled(account.config); - const disableBlockStreaming = sourceRepliesAreToolOnly - ? true - : resolveSlackDisableBlockStreaming({ - useStreaming, - shouldUseDraftStream, - blockStreamingEnabled, - }); - let streamSession: SlackStreamSession | null = null; - let nativeProgressStreamStartPromise: Promise | null = null; - let nativeProgressStreamThreadTs: string | undefined; - let streamFailed = false; - let usedReplyThreadTs: string | undefined; - let usedBlockReplyThreadTs: string | undefined; - let observedReplyDelivery = false; - let observedFinalReplyDelivery = false; - // Reply payloads routed through the native text stream. Track Slack - // acknowledgement separately because a later buffered suffix can fall back - // after earlier payloads are already visible. - const streamedDeliveries: Array<{ - kind: ReplyDispatchKind; - content: string; - acknowledged: boolean; - outcome?: "success" | "failure"; - }> = []; - const streamedFailuresOwnedByDispatcher: Record = { - tool: 0, - block: 0, - final: 0, - }; - const refreshStreamedAcknowledgements = (session: SlackStreamSession) => { - if (session.pendingText.length === 0) { - for (const delivery of streamedDeliveries) { - delivery.acknowledged = true; - } - } - }; - const recordStreamedDelivery = (kind: ReplyDispatchKind, content: string) => { - const delivery: (typeof streamedDeliveries)[number] = { - kind, - content, - acknowledged: false, - }; - streamedDeliveries.push(delivery); - return delivery; - }; - const rememberStreamedDelivery = ( - kind: ReplyDispatchKind, - content: string, - session: SlackStreamSession, - ) => { - recordStreamedDelivery(kind, content); - refreshStreamedAcknowledgements(session); - }; - const emitAcknowledgedStreamedDeliveries = (messageId?: string) => { - for (const delivery of streamedDeliveries) { - if (!delivery.acknowledged || delivery.outcome) { - continue; - } - emitSlackMessageSentHooks({ - ...messageSentHookContext, - to: messageSentHookTarget, - accountId: account.accountId, - content: delivery.content, - success: true, - ...(messageId ? { messageId } : {}), - }); - delivery.outcome = "success"; - } - }; - const acknowledgeStoppedStreamedDeliveries = ( - session: SlackStreamSession, - messageId?: string, - ) => { - refreshStreamedAcknowledgements(session); - for (const delivery of streamedDeliveries) { - delivery.acknowledged = true; - } - emitAcknowledgedStreamedDeliveries(messageId); - }; - const emitFailedPendingStreamedDeliveries = (error: string) => { - for (const delivery of streamedDeliveries) { - if (delivery.acknowledged || delivery.outcome) { - continue; - } - emitSlackMessageSentHooks({ - ...messageSentHookContext, - to: messageSentHookTarget, - accountId: account.accountId, - content: delivery.content, - success: false, - error, - }); - delivery.outcome = "failure"; - } - }; - const emitSuccessfulPendingStreamedDeliveries = (messageId?: string) => { - for (const delivery of streamedDeliveries) { - if (delivery.acknowledged || delivery.outcome) { - continue; - } - emitSlackMessageSentHooks({ - ...messageSentHookContext, - to: messageSentHookTarget, - accountId: account.accountId, - content: delivery.content, - success: true, - ...(messageId ? { messageId } : {}), - }); - delivery.outcome = "success"; - } - }; - let deliveryTracker = createSlackEventDeliveryTracker(); - const markPreviewPayloadDelivered = (params: { - kind: ReplyDispatchKind; - payload: ReplyPayload; - threadTs: string | undefined; - }) => { - deliveryTracker.markDelivered(params); - // Single-use reply modes move later same-turn payloads off the preview - // thread, so protect both delivery keys from duplicates. - const nextThreadTs = replyPlan.peekThreadTs(); - if (nextThreadTs !== params.threadTs) { - deliveryTracker.markDelivered({ ...params, threadTs: nextThreadTs }); - } - }; - const resolveDeliveryThreadTs = (params: { - kind: ReplyDispatchKind; - forcedThreadTs?: string; - }): string | undefined => { - const plannedThreadTs = params.forcedThreadTs ? undefined : replyPlan.nextThreadTs(); - return ( - params.forcedThreadTs ?? - plannedThreadTs ?? - (params.kind === "block" ? usedBlockReplyThreadTs : undefined) - ); - }; - const rememberDeliveredThreadTs = ( - kind: ReplyDispatchKind, - deliveredThreadTs: string | undefined, - ) => { - if (!deliveredThreadTs) { - return; - } - usedReplyThreadTs ??= deliveredThreadTs; - if (kind === "block") { - usedBlockReplyThreadTs = deliveredThreadTs; - } - }; - const deliverPendingStreamFallback = async ( - session: SlackStreamSession, - err: SlackStreamNotDeliveredError, - ): Promise => { - let fallbackError = err; - if (!session.stopped) { - try { - const stopResult = await stopSlackStream({ - session, - ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), - }); - acknowledgeStoppedStreamedDeliveries(session, stopResult.messageId); - observedReplyDelivery = true; - usedReplyThreadTs ??= session.threadTs; - return true; - } catch (stopErr) { - if (stopErr instanceof SlackStreamNotDeliveredError) { - fallbackError = stopErr; - } else { - runtime.error?.( - danger( - `slack-stream: failed to finalize buffered text before fallback: ${formatSlackError(stopErr)}`, - ), - ); - } - } - } - emitAcknowledgedStreamedDeliveries(); - // The Slack SDK still owns this text in-memory; no streaming API call has - // acknowledged it. Route through deliverReplies so pendingText that - // exceeds Slack's per-message text limit still lands (a single - // chat.postMessage would have failed with msg_too_long), and so the - // fallback respects the configured replyToMode/identity the same way - // normal replies do. - const fallbackText = fallbackError.pendingText.trim(); - if (!fallbackText) { - return false; - } - try { - await deliverReplies({ - cfg: ctx.cfg, - replies: [{ text: fallbackText } as ReplyPayload], - target: prepared.replyTarget, - token: ctx.botToken, - accountId: account.accountId, - runtime, - textLimit: ctx.textLimit, - mediaMaxBytes: ctx.mediaMaxBytes, - replyThreadTs: session.threadTs, - replyToMode: replyDeliveryMode, - ...(slackIdentity ? { identity: slackIdentity } : {}), - ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), - ...messageSentDeliveryHookContext, - deferMessageSentHooks: true, - ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), - }); - markSlackStreamFallbackDelivered(session); - if (!session.stopped) { - try { - await stopSlackStream({ - session, - ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), - }); - } catch (finalizeErr) { - runtime.error?.( - danger( - `slack-stream: failed to finalize native stream after fallback delivery: ${formatSlackError(finalizeErr)}`, - ), - ); - } - } - // The combined fallback can span multiple logical payloads and Slack - // chunks, so no single message `ts` correctly identifies every event. - emitSuccessfulPendingStreamedDeliveries(); - observedReplyDelivery = true; - usedReplyThreadTs ??= session.threadTs; - logVerbose( - `slack-stream: streamed delivery failed (${fallbackError.slackCode}); delivered ${fallbackText.length} chars via deliverReplies fallback`, - ); - return true; - } catch (postErr) { - emitFailedPendingStreamedDeliveries(formatErrorMessage(postErr)); - runtime.error?.( - danger( - `slack-stream: fallback deliverReplies failed after ${fallbackError.slackCode}: ${formatErrorMessage(postErr)}`, - ), - ); - return false; - } - }; - - const deliverNormally = async (params: { - payload: ReplyPayload; - kind: ReplyDispatchKind; - forcedThreadTs?: string; - }): Promise => { - if (params.payload.isReasoning === true) { - return undefined; - } - const replyThreadTs = resolveDeliveryThreadTs(params); - const deliveryReplyThreadTs = - replyDeliveryMode === "off" && !forcedReplyThreadTs && !isThreadReply - ? undefined - : replyThreadTs; - if ( - deliveryTracker.hasDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: deliveryReplyThreadTs, - }) - ) { - logVerbose("slack: suppressed duplicate normal delivery within the same turn"); - return deliveryReplyThreadTs; - } - await deliverReplies({ - cfg: ctx.cfg, - replies: [params.payload], - target: prepared.replyTarget, - token: ctx.botToken, - accountId: account.accountId, - runtime, - textLimit: ctx.textLimit, - mediaMaxBytes: ctx.mediaMaxBytes, - replyThreadTs: deliveryReplyThreadTs, - replyToMode: replyDeliveryMode, - ...(slackIdentity ? { identity: slackIdentity } : {}), - ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), - ...messageSentDeliveryHookContext, - ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), - }); - observedReplyDelivery = true; - if (params.kind === "final") { - observedFinalReplyDelivery = true; - } - const deliveredThreadTs = resolveDeliveredSlackReplyThreadTs({ - replyToMode: replyDeliveryMode, - payloadReplyToId: params.payload.replyToId, - replyThreadTs: deliveryReplyThreadTs, - }); - // Record the thread ts only after confirmed delivery success. - rememberDeliveredThreadTs(params.kind, deliveredThreadTs); - replyPlan.markSent(); - deliveryTracker.markDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: deliveryReplyThreadTs, - }); - return deliveryReplyThreadTs; - }; - - const deliverBufferedStreamFallback = async (params: { - session: SlackStreamSession; - err: SlackStreamNotDeliveredError; - payload: ReplyPayload; - kind: ReplyDispatchKind; - textOverride: string; - }): Promise => { - const delivered = await deliverPendingStreamFallback(params.session, params.err); - if (!delivered) { - // The reply dispatcher will charge the currently executing payload as - // failed; earlier buffered payloads need separate reconciliation below. - streamedFailuresOwnedByDispatcher[params.kind] += 1; - return false; - } - replyPlan.markSent(); - if (params.kind === "final") { - observedFinalReplyDelivery = true; - } - deliveryTracker.markDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: params.session.threadTs, - textOverride: params.textOverride, - }); - rememberDeliveredThreadTs(params.kind, params.session.threadTs); - return true; - }; - - const appendNativeProgressCompletion = async (isError: boolean) => { - const session = streamSession; - if (isError) { - nativeProgressTerminalStatus = "error"; - } - if (!session || nativeProgressCompletionSent) { - return; - } - const chunks = buildNativeProgressCompletionChunks(isError ? "error" : "complete"); - if (!chunks?.length) { - return; - } - try { - await appendSlackStream({ session, chunks }); - nativeProgressCompletionSent = true; - observedReplyDelivery ||= session.delivered; - } catch (err) { - streamFailed = true; - runtime.error?.( - danger(`slack-stream: native progress completion failed: ${formatSlackError(err)}`), - ); - } - }; - - const deliverWithStreaming = async (params: { - payload: ReplyPayload; - kind: ReplyDispatchKind; - }): Promise => { - if (params.payload.isReasoning === true) { - return; - } - const reply = resolveSendableOutboundReplyParts(params.payload); - const renderPlan = resolveSlackReplyRenderPlan(params.payload); - const plannedBlocks = - renderPlan.mode === "single" ? renderPlan.blocks : renderPlan.blockPart?.blocks; - if ( - streamFailed || - reply.hasMedia || - renderPlan.mode === "split" || - Boolean(plannedBlocks?.length) || - readSlackReplyBlocks(params.payload)?.length || - !reply.hasText - ) { - await deliverNormally({ - payload: params.payload, - kind: params.kind, - forcedThreadTs: streamSession?.threadTs ?? nativeProgressStreamThreadTs, - }); - return; - } - - const text = reply.trimmedText; - let plannedThreadTs: string | undefined; - try { - if (!streamSession && nativeProgressStreamStartPromise) { - await nativeProgressStreamStartPromise; - } - if (streamFailed) { - await deliverNormally({ - payload: params.payload, - kind: params.kind, - forcedThreadTs: streamSession?.threadTs ?? nativeProgressStreamThreadTs, - }); - return; - } - if (!streamSession) { - const streamThreadTs = replyPlan.nextThreadTs(); - plannedThreadTs = streamThreadTs; - if (!streamThreadTs) { - logVerbose( - "slack-stream: no reply thread target for stream start, falling back to normal delivery", - ); - streamFailed = true; - await deliverNormally({ - payload: params.payload, - kind: params.kind, - }); - return; - } - if ( - deliveryTracker.hasDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: streamThreadTs, - textOverride: text, - }) - ) { - logVerbose("slack-stream: suppressed duplicate stream start payload"); - return; - } - - streamSession = await startSlackStream({ - client: slackClient, - channel: message.channel, - threadTs: streamThreadTs, - text, - ...(slackIdentity ? { identity: slackIdentity } : {}), - teamId: await resolveSlackStreamRecipientTeamId({ - client: slackClient, - token: ctx.botToken, - userId: message.user, - fallbackTeamId: slackStreamFallbackTeamId, - }), - userId: message.user, - }); - refreshStreamedAcknowledgements(streamSession); - // startSlackStream may only buffer locally. Count delivery only after - // the SDK reports a real Slack response. - if (streamSession.delivered) { - observedReplyDelivery = true; - if (params.kind === "final") { - observedFinalReplyDelivery = true; - } - } - // Remember the reply text delivered through the text stream so the - // `message_sent` hook can fire after stopSlackStream flushes it. - // Only the text-stream path captures this; every deliverNormally branch - // already emits via deliverReplies, so capturing there would - // double-emit for the same payload. - if (text) { - rememberStreamedDelivery(params.kind, text, streamSession); - } - rememberDeliveredThreadTs(params.kind, streamThreadTs); - replyPlan.markSent(); - deliveryTracker.markDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: streamThreadTs, - textOverride: text, - }); - return; - } - if ( - deliveryTracker.hasDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: streamSession.threadTs, - textOverride: text, - }) - ) { - logVerbose("slack-stream: suppressed duplicate append payload"); - return; - } - - if (text) { - // appendSlackStream buffers text before attempting the Slack flush. - // Record first so a later successful stop can acknowledge a thrown append. - recordStreamedDelivery(params.kind, text); - } - await appendSlackStream({ - session: streamSession, - text: "\n" + text, - }); - refreshStreamedAcknowledgements(streamSession); - // appendSlackStream also buffers locally below the SDK threshold; avoid - // optimistic "done" status until Slack acknowledges a flush. - if (streamSession.delivered) { - observedReplyDelivery = true; - if (params.kind === "final") { - observedFinalReplyDelivery = true; - } - } - deliveryTracker.markDelivered({ - kind: params.kind, - payload: params.payload, - threadTs: streamSession.threadTs, - textOverride: text, - }); - } catch (err) { - if (err instanceof SlackStreamNotDeliveredError) { - streamFailed = true; - if (streamSession) { - const delivered = await deliverBufferedStreamFallback({ - session: streamSession, - err, - payload: params.payload, - kind: params.kind, - textOverride: text, - }); - if (delivered) { - return; - } - throw err; - } - await deliverNormally({ - payload: params.payload, - kind: params.kind, - forcedThreadTs: plannedThreadTs, - }); - return; - } - runtime.error?.( - danger(`slack-stream: streaming API call failed: ${formatSlackError(err)}, falling back`), - ); - streamFailed = true; - // Non-benign streaming errors leave `pendingText` populated with every - // buffered chunk since the last flush (appendSlackStream accumulates - // into pendingText BEFORE the SDK call, so the failing chunk is - // included too). Route the full buffer through the chunked fallback so - // earlier chunks aren't lost, then skip deliverNormally - pendingText - // already contains this payload's text. - if (streamSession && streamSession.pendingText) { - const bufferedFallbackErr = new SlackStreamNotDeliveredError( - streamSession.pendingText, - "unknown", - ); - const delivered = await deliverBufferedStreamFallback({ - session: streamSession, - err: bufferedFallbackErr, - payload: params.payload, - kind: params.kind, - textOverride: text, - }); - if (delivered) { - return; - } - throw err; - } - await deliverNormally({ - payload: params.payload, - kind: params.kind, - forcedThreadTs: streamSession?.threadTs ?? nativeProgressStreamThreadTs ?? plannedThreadTs, - }); - } - }; - - let draftPreviewCommitted = false; const deliverSlackPayload = async ( payload: ReplyPayload, info: { kind: ReplyDispatchKind }, @@ -1189,24 +89,25 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if ( info.kind === "final" && slackStreaming.mode === "progress" && - streamMode === "status_final" + progress.streamMode === "status_final" ) { - const hadProgressDraft = progressDraft.hasStarted; - progressDraft.markFinalReplyStarted(); - if (useNativeProgressStreaming) { - await waitForNativeProgressStreamStart(); - const finalThreadTs = streamSession?.threadTs ?? nativeProgressStreamThreadTs; - await deliverNormally({ + const hadProgressDraft = progress.progressDraft.hasStarted; + progress.progressDraft.markFinalReplyStarted(); + if (progress.useNativeProgressStreaming) { + await progress.waitForNativeProgressStreamStart(); + const finalThreadTs = + delivery.streamSession?.threadTs ?? delivery.nativeProgressStreamThreadTs; + await delivery.deliverNormally({ payload, kind: info.kind, forcedThreadTs: finalThreadTs, }); // Complete the cards only after the fresh final landed; a failed send // leaves completion to the outer cleanup, which can mark error state. - await appendNativeProgressCompletion(payload.isError === true); - progressDraft.markFinalReplyDelivered(); - if (!payload.isError && hadProgressDraft && streamSession) { - pendingNativeProgressReceipt = progressReceipt.buildSummaryLine(); + await progress.appendNativeProgressCompletion(payload.isError === true); + progress.progressDraft.markFinalReplyDelivered(); + if (!payload.isError && hadProgressDraft && delivery.streamSession) { + progress.pendingNativeProgressReceipt = progress.progressReceipt.buildSummaryLine(); } return; } @@ -1224,35 +125,42 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag const receiptMessageId = hadProgressDraft ? draftStream?.messageId() : undefined; // The draft already selected the reply thread; re-planning here could // route the fresh final elsewhere under stateful replyToMode values. - const draftThreadTs = hadProgressDraft ? (usedReplyThreadTs ?? statusThreadTs) : undefined; - await deliverNormally({ + const draftThreadTs = hadProgressDraft + ? (delivery.usedReplyThreadTs ?? statusThreadTs) + : undefined; + await delivery.deliverNormally({ payload, kind: info.kind, ...(draftThreadTs ? { forcedThreadTs: draftThreadTs } : {}), }); - progressDraft.markFinalReplyDelivered(); - if (!payload.isError && receiptChannelId && receiptMessageId && !progressReceiptCollapsed) { + progress.progressDraft.markFinalReplyDelivered(); + if ( + !payload.isError && + receiptChannelId && + receiptMessageId && + !progress.progressReceiptCollapsed + ) { // Collapse only after the fresh final lands; a failed send leaves the // working draft untouched as the turn record. - await collapseProgressReceipt({ + await progress.collapseProgressReceipt({ channelId: receiptChannelId, messageId: receiptMessageId, - text: progressReceipt.buildSummaryLine(), - threadTs: usedReplyThreadTs ?? statusThreadTs, + text: progress.progressReceipt.buildSummaryLine(), + threadTs: delivery.usedReplyThreadTs ?? statusThreadTs, }); } return; } - if (useNativeProgressStreaming) { - await deliverNormally({ + if (progress.useNativeProgressStreaming) { + await delivery.deliverNormally({ payload, kind: info.kind, - forcedThreadTs: streamSession?.threadTs ?? nativeProgressStreamThreadTs, + forcedThreadTs: delivery.streamSession?.threadTs ?? delivery.nativeProgressStreamThreadTs, }); return; } if (useStreaming) { - await deliverWithStreaming({ payload, kind: info.kind }); + await delivery.deliverWithStreaming({ payload, kind: info.kind }); return; } @@ -1278,8 +186,8 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag Boolean(ttsSupplement) && ttsSupplement?.visibleTextAlreadyDelivered !== true && Boolean(draftStream) && - !draftPreviewCommitted && - !observedFinalReplyDelivery && + !draftPreviewCommitted.value && + !delivery.observedFinalReplyDelivery && previewStreamingEnabled && !payload.text?.trim(); @@ -1287,8 +195,8 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag info.kind === "final" && ttsSupplement && draftStream && - !draftPreviewCommitted && - !observedFinalReplyDelivery && + !draftPreviewCommitted.value && + !delivery.observedFinalReplyDelivery && previewStreamingEnabled && !payload.isError && !requiresSeparateFallbackDelivery && @@ -1297,7 +205,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag const channelId = draftStream.channelId(); const messageId = draftStream.messageId(); if (channelId && messageId) { - const finalThreadTs = usedReplyThreadTs ?? statusThreadTs; + const finalThreadTs = delivery.usedReplyThreadTs ?? statusThreadTs; await draftStream.flush(); await draftStream.seal(); try { @@ -1318,7 +226,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag await draftStream.discardPending(); let delivered = false; try { - await deliverNormally({ + await delivery.deliverNormally({ payload: payload.text?.trim() ? payload : { @@ -1338,16 +246,16 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag } return; } - draftPreviewCommitted = true; - observedFinalReplyDelivery = true; - observedReplyDelivery = true; + draftPreviewCommitted.value = true; + delivery.observedFinalReplyDelivery = true; + delivery.observedReplyDelivery = true; replyPlan.markSent(); - await deliverNormally({ + await delivery.deliverNormally({ payload: buildTtsSupplementMediaPayload(payload), kind: info.kind, forcedThreadTs: finalThreadTs, }); - markPreviewPayloadDelivered({ kind: info.kind, payload, threadTs: finalThreadTs }); + delivery.markPreviewPayloadDelivered({ kind: info.kind, payload, threadTs: finalThreadTs }); return; } } @@ -1357,7 +265,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag payload, adapter: defineFinalizableLivePreviewAdapter({ draft: - draftStream && !draftPreviewCommitted && !observedFinalReplyDelivery + draftStream && !draftPreviewCommitted.value && !delivery.observedFinalReplyDelivery ? { flush: draftStream.flush, clear: draftStream.clear, @@ -1383,11 +291,11 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag return { text: previewFinalText, blocks: slackBlocks, - threadTs: usedReplyThreadTs ?? statusThreadTs, + threadTs: delivery.usedReplyThreadTs ?? statusThreadTs, }; }, editFinal: async (preview, edit) => { - if (deliveryTracker.hasDelivered({ kind: info.kind, payload, threadTs: edit.threadTs })) { + if (delivery.hasDelivered({ kind: info.kind, payload, threadTs: edit.threadTs })) { return; } await finalizeSlackPreviewEdit({ @@ -1410,33 +318,37 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag messageId: preview.messageId, }); } - draftPreviewCommitted = true; - observedFinalReplyDelivery = true; + draftPreviewCommitted.value = true; + delivery.observedFinalReplyDelivery = true; }, onPreviewFinalized: (_preview) => { // The preview edit promotes the draft message into the final answer. // Later same-turn payloads must not let fallback cleanup clear it. - draftPreviewCommitted = true; - observedFinalReplyDelivery = true; - const finalThreadTs = usedReplyThreadTs ?? statusThreadTs; - observedReplyDelivery = true; + draftPreviewCommitted.value = true; + delivery.observedFinalReplyDelivery = true; + const finalThreadTs = delivery.usedReplyThreadTs ?? statusThreadTs; + delivery.observedReplyDelivery = true; replyPlan.markSent(); // Supplemental TTS media is the terminal delivery for the logical // payload. Marking the preview first would suppress that media send. if (!ttsSupplement) { - markPreviewPayloadDelivered({ kind: info.kind, payload, threadTs: finalThreadTs }); + delivery.markPreviewPayloadDelivered({ + kind: info.kind, + payload, + threadTs: finalThreadTs, + }); } }, buildSupplementalPayload: () => ttsSupplement ? buildTtsSupplementMediaPayload(payload) : undefined, deliverSupplemental: async (supplementalPayload) => { - const previewThreadTs = usedReplyThreadTs ?? statusThreadTs; - const supplementalThreadTs = await deliverNormally({ + const previewThreadTs = delivery.usedReplyThreadTs ?? statusThreadTs; + const supplementalThreadTs = await delivery.deliverNormally({ payload: supplementalPayload, kind: info.kind, forcedThreadTs: previewThreadTs, }); - markPreviewPayloadDelivered({ + delivery.markPreviewPayloadDelivered({ kind: info.kind, payload, threadTs: supplementalThreadTs, @@ -1449,7 +361,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag }, }), deliverNormally: async () => { - await deliverNormally({ + await delivery.deliverNormally({ payload: shouldRestoreTtsSupplementTextForPreviewFallback ? { ...payload, @@ -1461,608 +373,6 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag }, }); }; - const onSlackDeliveryError = (err: unknown, info: { kind: string }) => { - runtime.error?.(danger(`slack ${info.kind} reply failed: ${formatSlackError(err)}`)); - replyPipeline.typingCallbacks?.onIdle?.(); - }; - - const draftStream = shouldUseDraftStream - ? createSlackDraftStream({ - target: prepared.replyTarget, - cfg, - token: ctx.botToken, - accountId: account.accountId, - ...(prepared.eventScope ? { eventScope: prepared.eventScope } : {}), - identity: slackIdentity, - ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), - maxChars: Math.min(ctx.textLimit, SLACK_TEXT_LIMIT), - resolveThreadTs: () => { - const ts = replyPlan.peekThreadTs(); - if (ts) { - usedReplyThreadTs ??= ts; - } - return ts; - }, - log: logVerbose, - warn: logVerbose, - }) - : undefined; - let hasStreamedMessage = false; - const streamMode = slackStreaming.draftMode; - const useNativeProgressStreaming = useStreaming && slackStreaming.mode === "progress"; - const progressDraftActive = Boolean(draftStream) || useNativeProgressStreaming; - const previewToolProgressEnabled = - progressDraftActive && resolveChannelStreamingPreviewToolProgress(account.config); - let shouldYieldDraftProgress: () => boolean = () => false; - const suppressDefaultToolProgressMessages = - resolveChannelStreamingSuppressDefaultToolProgressMessages(account.config, { - draftStreamActive: Boolean(draftStream) || useNativeProgressStreaming, - previewToolProgressEnabled, - previewStreamingEnabled, - }); - let previewToolProgressSuppressed = false; - let legacyPreviewToolProgressLines: ChannelProgressDraftLine[] = []; - // Last task rows emitted to the native stream; reconciliation terminalizes - // ids that drop out (plan shrinks, tool-line <-> plan source switches). - let nativeTaskState: SlackNativeTaskSnapshot = new Map(); - let appendRenderedText = ""; - let appendSourceText = ""; - let nativeProgressCompletionSent = false; - // Terminal status of the turn's final payload; completion retries and - // queued rotation must not repaint an errored turn as complete. - let nativeProgressTerminalStatus: "complete" | "error" = "complete"; - let nativeProgressChunkKey: string | undefined; - const progressReceipt = createChannelProgressReceiptTracker(); - let progressReceiptCollapsed = false; - let pendingNativeProgressReceipt: string | undefined; - const progressSeed = `${account.accountId}:${message.channel}`; - const useRichProgressDraft = - streamMode === "status_final" && resolveChannelProgressDraftRender(account.config) === "rich"; - const explicitProgressTitle = resolveExplicitSlackProgressTitle(account.config); - const progressDraftMaxLineChars = resolveChannelProgressDraftMaxLineChars(account.config); - - const waitForNativeProgressStreamStart = async (): Promise => { - if (streamSession || !nativeProgressStreamStartPromise) { - return true; - } - try { - await nativeProgressStreamStartPromise; - } catch { - streamFailed = true; - return false; - } - return !streamFailed; - }; - - const resolveStructuredProgressLines = ( - lines: readonly ChannelProgressDraftCompositorLine[], - ): ChannelProgressDraftLine[] => - lines.map((line) => { - if (typeof line !== "string") { - return line; - } - const reasoning = line.startsWith("🧠 "); - const text = line - .replace(/^(?:🧠|💬)\s+/u, "") - .replace(/^_(.*)_$/su, "$1") - .trim(); - return { - // Reasoning snapshots replace one rolling row; text-based ids would orphan it each delta. - ...(reasoning ? { id: "reasoning" } : {}), - kind: "item", - text, - label: reasoning ? "Reasoning" : "Update", - prefix: false, - }; - }); - - // Native cards derive from the compositor snapshot. Empty plans fall back - // to line tasks, and reconciliation retires rows from the prior source. - const resolveNativeProgressPlan = ( - snapshot: ChannelProgressDraftCompositorSnapshot, - ): readonly AgentPlanStep[] | undefined => (snapshot.plan?.length ? snapshot.plan : undefined); - - const resolveNativeProgressLines = ( - snapshot: ChannelProgressDraftCompositorSnapshot, - ): ChannelProgressDraftLine[] => { - const lines = resolveStructuredProgressLines(snapshot.lines); - if (snapshot.plan?.length || !snapshot.planExplanation) { - return lines; - } - const explanationLine = buildChannelProgressDraftLine({ - event: "plan", - phase: "update", - explanation: snapshot.planExplanation, - }); - return explanationLine ? [...lines, explanationLine] : lines; - }; - - const combineProgressHeadlineAndExplanation = ( - headline: string | undefined, - explanation: string | undefined, - ) => - headline && explanation && headline !== explanation - ? `${headline} — ${explanation}` - : (headline ?? explanation); - - const resolveNativeProgressTitle = (snapshot: ChannelProgressDraftCompositorSnapshot) => - combineProgressHeadlineAndExplanation( - explicitProgressTitle ?? snapshot.statusHeadline, - snapshot.planExplanation, - ); - - const buildNativeProgressChunks = (snapshot: ChannelProgressDraftCompositorSnapshot) => - streamSession - ? buildSlackProgressStreamUpdateChunks({ - title: resolveNativeProgressTitle(snapshot), - lines: resolveNativeProgressLines(snapshot), - plan: resolveNativeProgressPlan(snapshot), - maxLineChars: progressDraftMaxLineChars, - }) - : buildSlackProgressStreamStartChunks({ - title: resolveNativeProgressTitle(snapshot), - lines: resolveNativeProgressLines(snapshot), - plan: resolveNativeProgressPlan(snapshot), - maxLineChars: progressDraftMaxLineChars, - }); - - const markNativeProgressDelivered = (session: SlackStreamSession, threadTs?: string) => { - if (session.delivered) { - observedReplyDelivery = true; - } - if (threadTs) { - usedReplyThreadTs ??= threadTs; - rememberDeliveredThreadTs("block", threadTs); - } - }; - - const startNativeProgressStream = async ( - chunks: NonNullable>, - chunkKey: string, - ) => { - const streamThreadTs = replyPlan.nextThreadTs(); - if (!streamThreadTs) { - logVerbose( - "slack-stream: no reply thread target for native progress stream start, falling back", - ); - streamFailed = true; - return; - } - nativeProgressStreamThreadTs = streamThreadTs; - const startPromise = (async () => { - const session = await startSlackStream({ - client: slackClient, - channel: message.channel, - threadTs: streamThreadTs, - chunks, - taskDisplayMode: "plan", - ...(slackIdentity ? { identity: slackIdentity } : {}), - teamId: await resolveSlackStreamRecipientTeamId({ - client: slackClient, - token: ctx.botToken, - userId: message.user, - fallbackTeamId: slackStreamFallbackTeamId, - }), - userId: message.user, - }); - streamSession = session; - return session; - })(); - nativeProgressStreamStartPromise = startPromise; - let startedSession: SlackStreamSession | null; - try { - startedSession = await startPromise; - } finally { - if (nativeProgressStreamStartPromise === startPromise) { - nativeProgressStreamStartPromise = null; - } - } - if (startedSession) { - markNativeProgressDelivered(startedSession, streamThreadTs); - } - nativeProgressChunkKey = chunkKey; - replyPlan.markSent(); - }; - - const appendNativeProgressStream = async ( - chunks: NonNullable>, - chunkKey: string, - ) => { - if (!streamSession) { - return; - } - await appendSlackStream({ session: streamSession, chunks }); - markNativeProgressDelivered(streamSession); - nativeProgressChunkKey = chunkKey; - }; - - const updateNativeProgressStream = async () => { - const snapshot = progressDraft.getSnapshot(); - const progressLines = resolveNativeProgressLines(snapshot); - const hasRetirableNativeTasks = [...nativeTaskState.values()].some( - (task) => task.status !== "complete" && task.status !== "error", - ); - if ( - !useNativeProgressStreaming || - streamFailed || - (progressLines.length === 0 && - !snapshot.plan?.length && - !snapshot.statusHeadline && - !explicitProgressTitle && - !hasRetirableNativeTasks) - ) { - return; - } - const canContinue = await waitForNativeProgressStreamStart(); - if (!canContinue) { - return; - } - const reconciled = reconcileSlackNativeTaskChunks({ - previousTasks: nativeTaskState, - chunks: buildNativeProgressChunks(snapshot), - }); - const chunks = reconciled.chunks; - if (!chunks?.length) { - return; - } - const chunkKey = JSON.stringify(chunks); - if (chunkKey === nativeProgressChunkKey) { - return; - } - try { - if (!streamSession) { - await startNativeProgressStream(chunks, chunkKey); - } else { - await appendNativeProgressStream(chunks, chunkKey); - } - // Commit only after Slack accepted the chunks; a failed emit must retry - // the same reconciliation against the previous snapshot. - nativeTaskState = reconciled.tasks; - } catch (err) { - runtime.error?.( - danger( - `slack-stream: native progress stream failed: ${formatSlackError(err)}, falling back`, - ), - ); - streamFailed = true; - } - }; - - const resetProgressTurnState = () => { - progressReceipt.reset(); - progressReceiptCollapsed = false; - pendingNativeProgressReceipt = undefined; - }; - - const collapseProgressReceipt = async ( - params: Omit[0], "client" | "token" | "accountId">, - ) => { - const { botToken: token } = ctx; - try { - await finalizeSlackPreviewEdit({ - client: slackClient, - token, - accountId: account.accountId, - ...params, - }); - progressReceiptCollapsed = true; - } catch (err) { - logVerbose(`slack: progress receipt edit failed (${formatSlackError(err)})`); - } - }; - - const progressDraft = createChannelProgressDraftCompositor({ - entry: account.config, - mode: slackStreaming.mode, - active: progressDraftActive && streamMode === "status_final", - seed: progressSeed, - formatLine: escapeSlackMrkdwn, - reasoningLinePrefix: "🧠 ", - commentaryLinePrefix: "💬 ", - reasoningGate: previewToolProgressEnabled, - commentaryItalics: false, - updateOnLineChange: useNativeProgressStreaming || useRichProgressDraft, - update: async (previewText, options) => { - if (useNativeProgressStreaming) { - await updateNativeProgressStream(); - return; - } - if (!draftStream) { - return; - } - const snapshot = progressDraft.getSnapshot(); - const structuredLines = resolveStructuredProgressLines(options?.lines ?? snapshot.lines); - const richNarration = combineProgressHeadlineAndExplanation( - snapshot.statusHeadline, - snapshot.planExplanation, - ); - const richProgressBlocks = useRichProgressDraft - ? buildSlackProgressDraftBlocks({ - title: explicitProgressTitle, - lines: structuredLines, - plan: snapshot.plan, - narration: richNarration, - maxLineChars: progressDraftMaxLineChars, - }) - : undefined; - draftStream.update( - useRichProgressDraft && richProgressBlocks - ? { text: previewText, blocks: richProgressBlocks } - : previewText, - ); - hasStreamedMessage = true; - if (options?.flush) { - await draftStream.flush(); - } - }, - }); - const commentaryProgressEnabled = progressDraft.commentaryProgressEnabled; - - const buildNativeProgressCompletionChunks = (finalInProgressStatus: "complete" | "error") => { - const snapshot = progressDraft.getSnapshot(); - const lines = resolveNativeProgressLines(snapshot); - const hasRetirableNativeTasks = [...nativeTaskState.values()].some( - (task) => task.status !== "complete" && task.status !== "error", - ); - if (lines.length === 0 && !snapshot.plan?.length && !hasRetirableNativeTasks) { - return undefined; - } - return reconcileSlackNativeTaskChunks({ - previousTasks: nativeTaskState, - chunks: buildSlackProgressStreamCompletionChunks({ - title: resolveNativeProgressTitle(snapshot), - lines, - plan: resolveNativeProgressPlan(snapshot), - maxLineChars: progressDraftMaxLineChars, - finalInProgressStatus, - }), - }).chunks; - }; - - const finishNativeProgressTurn = async ( - completionChunks: ReturnType, - ) => { - if (nativeProgressStreamStartPromise) { - await nativeProgressStreamStartPromise.catch(() => null); - } - const session = streamSession; - if (session && !session.stopped) { - try { - if (completionChunks?.length) { - nativeProgressCompletionSent = true; - } - const stopResult = await stopSlackStream({ - session, - ...(completionChunks?.length ? { chunks: completionChunks } : {}), - ...(slackMessageMetadata ? { metadata: slackMessageMetadata } : {}), - }); - acknowledgeStoppedStreamedDeliveries(session, stopResult?.messageId); - if (pendingNativeProgressReceipt && stopResult?.messageId) { - await collapseProgressReceipt({ - channelId: session.channel, - messageId: stopResult.messageId, - text: pendingNativeProgressReceipt, - threadTs: session.threadTs, - }); - } - } catch (err) { - const error = formatSlackError(err); - // stopSlackStream makes the one-shot session terminal before throwing. - // Settle delivery bookkeeping before releasing that handle. - emitAcknowledgedStreamedDeliveries(); - emitFailedPendingStreamedDeliveries(error); - logVerbose(`slack-stream: failed to rotate native progress stream (${error})`); - } - } - streamSession = null; - nativeProgressStreamStartPromise = null; - nativeProgressStreamThreadTs = undefined; - streamFailed = false; - }; - - const pushPlanProgress = async (steps?: AgentPlanStep[], explanation?: string) => { - if (streamMode === "status_final") { - await progressDraft.pushPlanProgress(steps, { explanation }); - return; - } - if (previewToolProgressSuppressed || !draftStream) { - return; - } - const text = formatChannelProgressDraftText({ - entry: account.config, - lines: legacyPreviewToolProgressLines, - seed: progressSeed, - formatLine: escapeSlackMrkdwn, - narration: explanation, - plan: steps, - }); - if (text) { - draftStream.update(text); - hasStreamedMessage = true; - } - }; - - const pushPreviewProgress = async ( - line?: ChannelProgressDraftLine, - options?: { toolName?: string }, - ) => { - if (!draftStream && !useNativeProgressStreaming) { - return; - } - if (options?.toolName !== undefined && !isChannelProgressDraftWorkToolName(options.toolName)) { - return; - } - const normalized = line?.text.replace(/\s+/g, " ").trim(); - if (streamMode === "status_final") { - if (!line || !normalized) { - await progressDraft.noteActivity(); - return; - } - await progressDraft.pushToolProgress(line, options); - return; - } - if ( - !line || - !normalized || - !draftStream || - !previewToolProgressEnabled || - previewToolProgressSuppressed - ) { - return; - } - const nextLines = mergeChannelProgressDraftLine(legacyPreviewToolProgressLines, line, { - maxLines: resolveChannelProgressDraftMaxLines(account.config), - }); - if (nextLines === legacyPreviewToolProgressLines) { - return; - } - legacyPreviewToolProgressLines = nextLines; - draftStream.update( - formatChannelProgressDraftText({ - entry: account.config, - lines: legacyPreviewToolProgressLines, - seed: progressSeed, - formatLine: escapeSlackMrkdwn, - }), - ); - hasStreamedMessage = true; - }; - - const updateDraftFromPartial = (text?: string) => { - const trimmed = text?.trimEnd(); - if (!trimmed) { - return; - } - - if (streamMode === "append") { - previewToolProgressSuppressed = true; - legacyPreviewToolProgressLines = []; - const next = applyAppendOnlyStreamUpdate({ - incoming: trimmed, - rendered: appendRenderedText, - source: appendSourceText, - }); - appendRenderedText = next.rendered; - appendSourceText = next.source; - if (!next.changed) { - return; - } - draftStream?.update(next.rendered); - hasStreamedMessage = true; - return; - } - - if (streamMode === "status_final") { - return; - } - - previewToolProgressSuppressed = true; - legacyPreviewToolProgressLines = []; - draftStream?.update(trimmed); - hasStreamedMessage = true; - }; - const pushReasoningProgress = async (payload?: { - text?: string; - isReasoningSnapshot?: boolean; - }) => { - if (!payload?.text) { - return; - } - if (streamMode !== "status_final") { - const normalized = progressDraft - .mergeReasoningProgress(payload.text, { - snapshot: payload.isReasoningSnapshot === true, - }) - .replace(/^_(.*)_$/su, "$1") - .trim(); - if (!normalized) { - return; - } - await pushPreviewProgress({ - id: "reasoning", - kind: "item", - text: normalized, - label: "Reasoning", - }); - return; - } - progressReceipt.noteReasoning(); - await progressDraft.pushReasoningProgress(payload.text, { - snapshot: payload.isReasoningSnapshot === true, - }); - }; - const resetDraftDeliveryState = () => { - hasStreamedMessage = false; - appendRenderedText = ""; - appendSourceText = ""; - }; - const resetDraftProgressState = () => { - progressDraft.resetReasoningProgress(); - previewToolProgressSuppressed = false; - legacyPreviewToolProgressLines = []; - }; - const beginNewProgressTurn = async (options?: { force?: boolean }) => { - const completionChunks = - useNativeProgressStreaming && !nativeProgressCompletionSent - ? buildNativeProgressCompletionChunks(nativeProgressTerminalStatus) - : undefined; - if (!progressDraft.beginNewTurn(options)) { - return false; - } - // Native messages are one-shot streams. Stop the prior turn before the - // reset compositor can publish the queued turn's first snapshot. - if (useNativeProgressStreaming) { - await finishNativeProgressTurn(completionChunks); - } else { - draftStream?.forceNewMessage(); - } - resetProgressTurnState(); - nativeTaskState = new Map(); - nativeProgressCompletionSent = false; - nativeProgressTerminalStatus = "complete"; - nativeProgressChunkKey = undefined; - // A re-armed turn is a new visible reply: it must not dedupe against or - // inherit delivery state from the settled turn (mirrors queued admission). - draftPreviewCommitted = false; - observedFinalReplyDelivery = false; - deliveryTracker = createSlackEventDeliveryTracker(); - progressReceiptCollapsed = false; - return true; - }; - const onDraftBoundary = - !shouldUseDraftStream && !useNativeProgressStreaming - ? undefined - : async () => { - if (streamMode === "status_final") { - await beginNewProgressTurn(); - return; - } - if (hasStreamedMessage) { - draftStream?.forceNewMessage(); - } - resetDraftDeliveryState(); - resetDraftProgressState(); - }; - - const onQueuedFollowupAdmitted = - !shouldUseDraftStream && !useNativeProgressStreaming - ? undefined - : async () => { - // A queued input is a new visible reply even though it drains through - // this turn's callbacks. Do not let it edit or dedupe against this run. - await draftStream?.flush(); - draftPreviewCommitted = false; - observedFinalReplyDelivery = false; - if (streamMode === "status_final") { - await beginNewProgressTurn({ force: true }); - } else { - draftStream?.forceNewMessage(); - } - deliveryTracker = createSlackEventDeliveryTracker(); - resetDraftDeliveryState(); - resetDraftProgressState(); - }; - let dispatchError: unknown; let queuedFinal = false; let counts: Partial> = {}; @@ -2096,14 +406,16 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag hasRepliedRef, disableBlockStreaming, onModelSelected, - suppressDefaultToolProgressMessages: suppressDefaultToolProgressMessages ? true : undefined, - commentaryProgressEnabled: commentaryProgressEnabled ? true : undefined, + suppressDefaultToolProgressMessages: progress.suppressDefaultToolProgressMessages + ? true + : undefined, + commentaryProgressEnabled: progress.commentaryProgressEnabled ? true : undefined, progressPreambleEnabled: - progressDraftActive && slackStreaming.mode === "progress" ? true : undefined, - commentaryPayloadsEnabled: commentaryProgressEnabled ? true : undefined, - onVerboseProgressVisibility: commentaryProgressEnabled + progress.progressDraftActive && slackStreaming.mode === "progress" ? true : undefined, + commentaryPayloadsEnabled: progress.commentaryProgressEnabled ? true : undefined, + onVerboseProgressVisibility: progress.commentaryProgressEnabled ? (isActive) => { - shouldYieldDraftProgress = isActive; + progress.setShouldYieldDraftProgress(isActive); } : undefined, allowProgressCallbacksWhenSourceDeliverySuppressed: @@ -2116,18 +428,18 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag : !previewStreamingEnabled ? undefined : async (payload) => { - updateDraftFromPartial(payload.text); + progress.updateDraftFromPartial(payload.text); }, - onAssistantMessageStart: onDraftBoundary, + onAssistantMessageStart: progress.onDraftBoundary, onReasoningEnd: async () => { - progressReceipt.closeReasoning(); - await onDraftBoundary?.(); + progress.progressReceipt.closeReasoning(); + await progress.onDraftBoundary?.(); }, - onQueuedFollowupAdmitted, + onQueuedFollowupAdmitted: progress.onQueuedFollowupAdmitted, onReasoningStream: - statusReactionsEnabled || previewToolProgressEnabled + statusReactionsEnabled || progress.previewToolProgressEnabled ? async (payload) => { - await pushReasoningProgress(payload); + await progress.pushReasoningProgress(payload); if (!statusReactionsEnabled) { return; } @@ -2139,9 +451,9 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag await statusReactions.setTool(payload.name); } if (payload.phase === "start") { - progressReceipt.noteToolCall(payload.name); + progress.progressReceipt.noteToolCall(payload.name); } - await pushPreviewProgress( + await progress.pushPreviewProgress( buildChannelProgressDraftLineForEntry( account.config, { @@ -2158,24 +470,27 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag ); }, onItemEvent: async (payload) => { - if (streamMode === "status_final" && payload.kind === "preamble") { - if (shouldYieldDraftProgress()) { + if (progress.streamMode === "status_final" && payload.kind === "preamble") { + if (progress.shouldYieldDraftProgress()) { return; } - await progressDraft.pushPreambleHeadline(payload.progressText, { + await progress.progressDraft.pushPreambleHeadline(payload.progressText, { itemId: payload.itemId, }); - if (commentaryProgressEnabled) { - const accepted = await progressDraft.pushCommentaryProgress(payload.progressText, { - itemId: payload.itemId, - }); + if (progress.commentaryProgressEnabled) { + const accepted = await progress.progressDraft.pushCommentaryProgress( + payload.progressText, + { + itemId: payload.itemId, + }, + ); if (accepted) { - progressReceipt.noteCommentary(payload.itemId, payload.progressText); + progress.progressReceipt.noteCommentary(payload.itemId, payload.progressText); } } return; } - await pushPreviewProgress( + await progress.pushPreviewProgress( buildChannelProgressDraftLineForEntry(account.config, { event: "item", itemId: payload.itemId, @@ -2195,13 +510,13 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (payload.phase !== "update") { return; } - await pushPlanProgress(payload.steps, payload.explanation); + await progress.pushPlanProgress(payload.steps, payload.explanation); }, onApprovalEvent: async (payload) => { if (payload.phase !== "requested") { return; } - await pushPreviewProgress( + await progress.pushPreviewProgress( buildChannelProgressDraftLine({ event: "approval", phase: payload.phase, @@ -2216,7 +531,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (payload.phase !== "end") { return; } - await pushPreviewProgress( + await progress.pushPreviewProgress( buildChannelProgressDraftLine({ event: "command-output", itemId: payload.itemId, @@ -2233,7 +548,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (payload.phase !== "end") { return; } - await pushPreviewProgress( + await progress.pushPreviewProgress( buildChannelProgressDraftLine({ event: "patch", itemId: payload.itemId, @@ -2258,7 +573,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag } catch (err) { dispatchError = err; } finally { - progressDraft.cancel(); + progress.progressDraft.cancel(); await draftStream?.discardPending(); } @@ -2266,17 +581,17 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag // Finalize the stream if one was started // ----------------------------------------------------------------------- let streamFallbackDelivered = false; - const finalStream = streamSession as SlackStreamSession | null; + const finalStream = delivery.streamSession as SlackStreamSession | null; if (finalStream && !finalStream.stopped) { try { const completionChunks = - useNativeProgressStreaming && !nativeProgressCompletionSent - ? buildNativeProgressCompletionChunks( - dispatchError ? "error" : nativeProgressTerminalStatus, + progress.useNativeProgressStreaming && !progress.nativeProgressCompletionSent + ? progress.buildNativeProgressCompletionChunks( + dispatchError ? "error" : progress.nativeProgressTerminalStatus, ) : undefined; if (completionChunks?.length) { - nativeProgressCompletionSent = true; + progress.nativeProgressCompletionSent = true; } const stopResult = await stopSlackStream({ session: finalStream, @@ -2288,25 +603,29 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag // here (the streaming happy-path never goes through deliverReplies, which // emits for the non-streaming/fallback paths). emitSlackMessageSentHooks // self-gates on registered listeners, so this is a no-op when unused. - acknowledgeStoppedStreamedDeliveries(finalStream, stopResult?.messageId); - if (pendingNativeProgressReceipt && stopResult?.messageId && !progressReceiptCollapsed) { - await collapseProgressReceipt({ + delivery.acknowledgeStoppedStreamedDeliveries(finalStream, stopResult?.messageId); + if ( + progress.pendingNativeProgressReceipt && + stopResult?.messageId && + !progress.progressReceiptCollapsed + ) { + await progress.collapseProgressReceipt({ channelId: finalStream.channel, messageId: stopResult.messageId, - text: pendingNativeProgressReceipt, + text: progress.pendingNativeProgressReceipt, threadTs: finalStream.threadTs, }); } } catch (err) { if (err instanceof SlackStreamNotDeliveredError) { - streamFallbackDelivered = await deliverPendingStreamFallback(finalStream, err); + streamFallbackDelivered = await delivery.deliverPendingStreamFallback(finalStream, err); if (!streamFallbackDelivered) { dispatchError ??= err; } } else { const error = formatSlackError(err); - emitAcknowledgedStreamedDeliveries(); - emitFailedPendingStreamedDeliveries(error); + delivery.emitAcknowledgedStreamedDeliveries(); + delivery.emitFailedPendingStreamedDeliveries(error); runtime.error?.(danger(`slack-stream: failed to stop stream: ${error}`)); if (!finalStream.delivered) { dispatchError ??= err; @@ -2315,27 +634,13 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag } } - for (const kind of ["tool", "block", "final"] as const) { - const failedStreamedCount = streamedDeliveries.filter( - (delivery) => delivery.kind === kind && delivery.outcome === "failure", - ).length; - const additionalFailedStreamed = Math.max( - 0, - failedStreamedCount - streamedFailuresOwnedByDispatcher[kind], - ); - if (additionalFailedStreamed > 0) { - counts = { - ...counts, - [kind]: Math.max(0, (counts[kind] ?? 0) - additionalFailedStreamed), - }; - } - } + counts = delivery.reconcileCounts(counts); queuedFinal = queuedFinal && (counts.final ?? 0) > 0; const anyReplyDelivered = hasVisibleInboundReplyDispatch( { queuedFinal, counts }, { - observedReplyDelivery, + observedReplyDelivery: delivery.observedReplyDelivery, fallbackDelivered: streamFallbackDelivered, }, ); @@ -2356,7 +661,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag // Record thread participation only when we actually delivered a reply and // know the thread ts that was used (set by deliverNormally, streaming start, // or draft stream). Falls back to statusThreadTs for edge cases. - const participationThreadTs = usedReplyThreadTs ?? statusThreadTs; + const participationThreadTs = delivery.usedReplyThreadTs ?? statusThreadTs; if (anyReplyDelivered && participationThreadTs) { recordSlackThreadParticipation(account.accountId, message.channel, participationThreadTs, { agentId: route.agentId, @@ -2366,7 +671,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (dispatchError) { throw toErrorObject(dispatchError, "Slack dispatch failed"); } - if (!anyReplyDelivered && !draftPreviewCommitted) { + if (!anyReplyDelivered && !draftPreviewCommitted.value) { await draftStream?.clear(); return; } @@ -2378,4 +683,3 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag ); } } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */