github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-11 00:14:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(config): unify streaming config across channels

Browse Source
This commit is contained in:
Peter Steinberger
2026-02-21 19:53:23 +01:00
parent 747bb581b3
commit 2c14b0cf4c
26 changed files with 885 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
src/slack/monitor/message-handler/dispatch.ts
View File

@@ -14,7 +14,7 @@ import { createSlackDraftStream } from "../../draft-stream.js";
import {
applyAppendOnlyStreamUpdate,
buildStatusFinalPreviewText,
resolveSlackStreamMode,
resolveSlackStreamingConfig,
} from "../../stream-mode.js";
import type { SlackStreamSession } from "../../streaming.js";
import { appendSlackStream, startSlackStream, stopSlackStream } from "../../streaming.js";
@@ -26,8 +26,14 @@ function hasMedia(payload: ReplyPayload): boolean {
return Boolean(payload.mediaUrl) || (payload.mediaUrls?.length ?? 0) > 0;
}
export function isSlackStreamingEnabled(streaming: boolean | undefined): boolean {
return streaming !== false;
export function isSlackStreamingEnabled(params: {
mode: "off" | "partial" | "block" | "progress";
nativeStreaming: boolean;
}): boolean {
if (params.mode !== "partial") {
return false;
}
return params.nativeStreaming;
}
export function resolveSlackStreamingThreadHint(params: {
@@ -146,7 +152,16 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
accountId: route.accountId,
});
const streamingEnabled = isSlackStreamingEnabled(account.config.streaming);
const slackStreaming = resolveSlackStreamingConfig({
streaming: account.config.streaming,
streamMode: account.config.streamMode,
nativeStreaming: account.config.nativeStreaming,
});
const previewStreamingEnabled = slackStreaming.mode !== "off";
const streamingEnabled = isSlackStreamingEnabled({
mode: slackStreaming.mode,
nativeStreaming: slackStreaming.nativeStreaming,
});
const streamThreadHint = resolveSlackStreamingThreadHint({
replyToMode: ctx.replyToMode,
incomingThreadTs,
@@ -233,6 +248,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
const draftChannelId = draftStream?.channelId();
const finalText = payload.text;
const canFinalizeViaPreviewEdit =
previewStreamingEnabled &&
streamMode !== "status_final" &&
mediaCount === 0 &&
!payload.isError &&
@@ -256,7 +272,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
`slack: preview final edit failed; falling back to standard send (${String(err)})`,
);
}
} else if (streamMode === "status_final" && hasStreamedMessage) {
} else if (previewStreamingEnabled && streamMode === "status_final" && hasStreamedMessage) {
try {
const statusChannelId = draftStream?.channelId();
const statusMessageId = draftStream?.messageId();
@@ -307,7 +323,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
warn: logVerbose,
});
let hasStreamedMessage = false;
const streamMode = resolveSlackStreamMode(account.config.streamMode);
const streamMode = slackStreaming.draftMode;
let appendRenderedText = "";
let appendSourceText = "";
let statusUpdateCount = 0;
@@ -363,31 +379,37 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
onModelSelected,
onPartialReply: useStreaming
? undefined
: async (payload) => {
updateDraftFromPartial(payload.text);
},
: !previewStreamingEnabled
? undefined
: async (payload) => {
updateDraftFromPartial(payload.text);
},
onAssistantMessageStart: useStreaming
? undefined
: async () => {
if (hasStreamedMessage) {
draftStream.forceNewMessage();
hasStreamedMessage = false;
appendRenderedText = "";
appendSourceText = "";
statusUpdateCount = 0;
}
},
: !previewStreamingEnabled
? undefined
: async () => {
if (hasStreamedMessage) {
draftStream.forceNewMessage();
hasStreamedMessage = false;
appendRenderedText = "";
appendSourceText = "";
statusUpdateCount = 0;
}
},
onReasoningEnd: useStreaming
? undefined
: async () => {
if (hasStreamedMessage) {
draftStream.forceNewMessage();
hasStreamedMessage = false;
appendRenderedText = "";
appendSourceText = "";
statusUpdateCount = 0;
}
},
: !previewStreamingEnabled
? undefined
: async () => {
if (hasStreamedMessage) {
draftStream.forceNewMessage();
hasStreamedMessage = false;
appendRenderedText = "";
appendSourceText = "";
statusUpdateCount = 0;
}
},
},
});
await draftStream.flush();