mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-02 07:41:33 +00:00
refactor(slack): split message dispatch (#113602)
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
1defbc5473
commit
64daf8dd23
@@ -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
|
||||
|
||||
274
extensions/slack/src/monitor/message-handler/dispatch-helpers.ts
Normal file
274
extensions/slack/src/monitor/message-handler/dispatch-helpers.ts
Normal file
@@ -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<typeof resolveChannelProgressDraftConfig>[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<typeof resolveChannelProgressDraftConfig>[0],
|
||||
): boolean {
|
||||
const streaming = entry?.streaming;
|
||||
if (!streaming || typeof streaming !== "object" || Array.isArray(streaming)) {
|
||||
return false;
|
||||
}
|
||||
const progressConfig = (streaming as Record<string, unknown>).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<object, Map<string, string>>();
|
||||
|
||||
function getSlackStreamRecipientTeamCache(client: object): Map<string, string> {
|
||||
const existing = slackStreamRecipientTeamCaches.get(client);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const cache = new Map<string, string>();
|
||||
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<string>();
|
||||
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<PreparedSlackMessage["ctx"]["app"]["client"], "users">;
|
||||
token: string;
|
||||
userId?: PreparedSlackMessage["message"]["user"];
|
||||
fallbackTeamId?: string;
|
||||
}): Promise<string | undefined> {
|
||||
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;
|
||||
}
|
||||
@@ -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<SlackDispatchSetup, "account" | "ctx" | "slackClient">;
|
||||
edit: Omit<Parameters<typeof finalizeSlackPreviewEdit>[0], "client" | "token" | "accountId">;
|
||||
}): Promise<boolean> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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<boolean> => {
|
||||
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<ReturnType<typeof buildSlackProgressStreamStartChunks>>,
|
||||
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<ReturnType<typeof buildSlackProgressStreamUpdateChunks>>,
|
||||
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<typeof collapseSlackProgressReceipt>[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<typeof buildNativeProgressCompletionChunks>,
|
||||
) => {
|
||||
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(),
|
||||
};
|
||||
}
|
||||
359
extensions/slack/src/monitor/message-handler/dispatch-setup.ts
Normal file
359
extensions/slack/src/monitor/message-handler/dispatch-setup.ts
Normal file
@@ -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<ReturnType<typeof createSlackDispatchSetup>>;
|
||||
@@ -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<SlackStreamSession | null> | 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<ReplyDispatchKind, number> = {
|
||||
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<boolean> => {
|
||||
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<string | undefined> => {
|
||||
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<boolean> => {
|
||||
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<void> => {
|
||||
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<Record<ReplyDispatchKind, number>>) => {
|
||||
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<typeof createSlackStreamingDeliveryRuntime>;
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user