mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-25 00:51:17 +00:00
383 lines
13 KiB
TypeScript
383 lines
13 KiB
TypeScript
/**
|
|
* Slack native text streaming helpers.
|
|
*
|
|
* Uses the Slack SDK's `ChatStreamer` (via `client.chatStream()`) to stream
|
|
* text responses word-by-word in a single updating message, matching Slack's
|
|
* "Agents & AI Apps" streaming UX.
|
|
*
|
|
* @see https://docs.slack.dev/ai/developing-ai-apps#streaming
|
|
* @see https://docs.slack.dev/reference/methods/chat.startStream
|
|
* @see https://docs.slack.dev/reference/methods/chat.appendStream
|
|
* @see https://docs.slack.dev/reference/methods/chat.stopStream
|
|
*/
|
|
|
|
import type { AnyChunk, MessageMetadata } from "@slack/types";
|
|
import type { WebClient } from "@slack/web-api";
|
|
import type { ChatStreamer } from "@slack/web-api/dist/chat-stream.js";
|
|
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
|
import type { SlackSendIdentity } from "./send.js";
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Types
|
|
// ---------------------------------------------------------------------------
|
|
|
|
export type SlackStreamSession = {
|
|
/** The SDK ChatStreamer instance managing this stream. */
|
|
streamer: ChatStreamer;
|
|
/** Channel this stream lives in. */
|
|
channel: string;
|
|
/** Thread timestamp (required for streaming). */
|
|
threadTs: string;
|
|
/** True once stop() has been called. */
|
|
stopped: boolean;
|
|
/**
|
|
* True once any Slack API call (startStream / appendStream) has succeeded.
|
|
* The SDK buffers appended text locally until the buffer exceeds
|
|
* `buffer_size` (default 256 chars); only then does it issue a network
|
|
* call. Until `delivered` flips, nothing has actually reached Slack.
|
|
*/
|
|
delivered: boolean;
|
|
/** Text accepted by the SDK but not yet acknowledged by Slack. */
|
|
pendingText: string;
|
|
};
|
|
|
|
type StartSlackStreamParams = {
|
|
client: WebClient;
|
|
channel: string;
|
|
threadTs: string;
|
|
/** Optional initial markdown text to include in the stream start. */
|
|
text?: string;
|
|
/** Optional structured Slack stream chunks to include in the stream start. */
|
|
chunks?: AnyChunk[];
|
|
/** Native Slack task display mode for task_update chunks. */
|
|
taskDisplayMode?: "plan" | "timeline";
|
|
/** Optional custom authorship supported by chat.startStream. */
|
|
identity?: SlackSendIdentity;
|
|
/**
|
|
* The team ID of the workspace this stream belongs to.
|
|
* Required by the Slack API for `chat.startStream` / `chat.stopStream`.
|
|
* Obtain from `auth.test` response (`team_id`).
|
|
*/
|
|
teamId?: string;
|
|
/**
|
|
* The user ID of the message recipient (required for DM streaming).
|
|
* Without this, `chat.stopStream` fails with `missing_recipient_user_id`
|
|
* in direct message conversations.
|
|
*/
|
|
userId?: string;
|
|
};
|
|
|
|
type AppendSlackStreamParams = {
|
|
session: SlackStreamSession;
|
|
text?: string;
|
|
chunks?: AnyChunk[];
|
|
};
|
|
|
|
type StopSlackStreamParams = {
|
|
session: SlackStreamSession;
|
|
/** Optional final markdown text to append before stopping. */
|
|
text?: string;
|
|
/** Optional final stream chunks to append before stopping. */
|
|
chunks?: AnyChunk[];
|
|
metadata?: MessageMetadata;
|
|
};
|
|
|
|
/**
|
|
* Thrown when Slack definitively rejects a stream flush/finalize while text
|
|
* remains buffered locally by the Slack SDK. Carries the pending text so the
|
|
* caller can deliver it via the normal Slack reply path.
|
|
*/
|
|
export class SlackStreamNotDeliveredError extends Error {
|
|
readonly pendingText: string;
|
|
readonly slackCode: string;
|
|
constructor(pendingText: string, slackCode: string) {
|
|
super(
|
|
`slack-stream: finalize failed with ${slackCode} before buffered text reached Slack ` +
|
|
`(${pendingText.length} chars pending)`,
|
|
);
|
|
this.name = "SlackStreamNotDeliveredError";
|
|
this.pendingText = pendingText;
|
|
this.slackCode = slackCode;
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Stream lifecycle
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Start a new Slack text stream.
|
|
*
|
|
* Returns a {@link SlackStreamSession} that should be passed to
|
|
* {@link appendSlackStream} and {@link stopSlackStream}.
|
|
*
|
|
* The first chunk of text can optionally be included via `text`.
|
|
*/
|
|
export async function startSlackStream(
|
|
params: StartSlackStreamParams,
|
|
): Promise<SlackStreamSession> {
|
|
const { client, channel, threadTs, text, chunks, taskDisplayMode, teamId, userId, identity } =
|
|
params;
|
|
const identityPayload = identity?.iconUrl
|
|
? { ...(identity.username ? { username: identity.username } : {}), icon_url: identity.iconUrl }
|
|
: identity?.iconEmoji
|
|
? {
|
|
...(identity.username ? { username: identity.username } : {}),
|
|
icon_emoji: identity.iconEmoji,
|
|
}
|
|
: identity?.username
|
|
? { username: identity.username }
|
|
: {};
|
|
|
|
logVerbose(
|
|
`slack-stream: starting stream in ${channel} thread=${threadTs}${teamId ? ` team=${teamId}` : ""}${userId ? ` user=${userId}` : ""}`,
|
|
);
|
|
|
|
const streamer = client.chatStream({
|
|
channel,
|
|
thread_ts: threadTs,
|
|
...(taskDisplayMode ? { task_display_mode: taskDisplayMode } : {}),
|
|
...(teamId ? { recipient_team_id: teamId } : {}),
|
|
...(userId ? { recipient_user_id: userId } : {}),
|
|
...identityPayload,
|
|
});
|
|
|
|
const session: SlackStreamSession = {
|
|
streamer,
|
|
channel,
|
|
threadTs,
|
|
stopped: false,
|
|
delivered: false,
|
|
pendingText: "",
|
|
};
|
|
|
|
if (text || chunks?.length) {
|
|
if (text) {
|
|
session.pendingText += text;
|
|
}
|
|
// Slack SDK ChatStreamer keeps short markdown_text chunks in a local buffer
|
|
// and returns null until buffer_size is reached. Structured chunks force a
|
|
// flush. Only a non-null response means Slack acknowledged
|
|
// startStream/appendStream.
|
|
try {
|
|
const result = await streamer.append({
|
|
...(text ? { markdown_text: text } : {}),
|
|
...(chunks?.length ? { chunks } : {}),
|
|
});
|
|
if (result) {
|
|
session.delivered = true;
|
|
session.pendingText = "";
|
|
}
|
|
logVerbose(
|
|
`slack-stream: appended initial payload (${text?.length ?? 0} chars, ${
|
|
chunks?.length ?? 0
|
|
} chunks, ${result ? "flushed" : "buffered"})`,
|
|
);
|
|
} catch (err) {
|
|
if (isBenignSlackFinalizeError(err) && session.pendingText) {
|
|
throw new SlackStreamNotDeliveredError(
|
|
session.pendingText,
|
|
extractSlackErrorCode(err) ?? "unknown",
|
|
);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
return session;
|
|
}
|
|
|
|
/**
|
|
* Append markdown text to an active Slack stream.
|
|
*/
|
|
export async function appendSlackStream(params: AppendSlackStreamParams): Promise<void> {
|
|
const { session, text, chunks } = params;
|
|
|
|
if (session.stopped) {
|
|
logVerbose("slack-stream: attempted to append to a stopped stream, ignoring");
|
|
return;
|
|
}
|
|
|
|
if (!text && !chunks?.length) {
|
|
return;
|
|
}
|
|
|
|
if (text) {
|
|
session.pendingText += text;
|
|
}
|
|
try {
|
|
// Same SDK contract as startSlackStream: null means local-only buffer,
|
|
// non-null means Slack accepted the pending buffer/chunks and it is visible.
|
|
const result = await session.streamer.append({
|
|
...(text ? { markdown_text: text } : {}),
|
|
...(chunks?.length ? { chunks } : {}),
|
|
});
|
|
if (result) {
|
|
session.delivered = true;
|
|
session.pendingText = "";
|
|
}
|
|
logVerbose(
|
|
`slack-stream: appended ${text?.length ?? 0} chars, ${chunks?.length ?? 0} chunks (${
|
|
result ? "flushed" : "buffered"
|
|
})`,
|
|
);
|
|
} catch (err) {
|
|
if (isBenignSlackFinalizeError(err) && session.pendingText) {
|
|
throw new SlackStreamNotDeliveredError(
|
|
session.pendingText,
|
|
extractSlackErrorCode(err) ?? "unknown",
|
|
);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
/** Result of {@link stopSlackStream}. */
|
|
type StopSlackStreamResult = {
|
|
/**
|
|
* The Slack `ts` of the finalized streamed message, when `chat.stopStream`
|
|
* reports it. Used to populate `MessageSentEvent.messageId` for the
|
|
* streaming reply path. Undefined when the stream was already stopped or
|
|
* Slack omitted the timestamp.
|
|
*/
|
|
messageId?: string;
|
|
};
|
|
|
|
/**
|
|
* Stop (finalize) a Slack stream.
|
|
*
|
|
* After calling this the stream message becomes a normal Slack message.
|
|
* Optionally include final text to append before stopping.
|
|
*
|
|
* If Slack's `chat.stopStream` responds with a definitive recipient/channel
|
|
* rejection while text is still buffered locally, this function throws a
|
|
* {@link SlackStreamNotDeliveredError} carrying that pending text so the caller
|
|
* can deliver it through the normal Slack reply path. Ambiguous failures
|
|
* propagate unchanged because Slack may have committed the request.
|
|
*
|
|
* If Slack responds with a known benign finalize error (see
|
|
* {@link BENIGN_SLACK_FINALIZE_ERROR_CODES}) after prior `append` calls already
|
|
* landed, the error is swallowed and the session is marked stopped - the
|
|
* already-delivered text stays visible.
|
|
*
|
|
* Errors without buffered text propagate unchanged.
|
|
*
|
|
* On success, returns the finalized message's Slack `ts` (when reported) so the
|
|
* caller can emit the `message_sent` hook with a populated `messageId`.
|
|
*/
|
|
export async function stopSlackStream(
|
|
params: StopSlackStreamParams,
|
|
): Promise<StopSlackStreamResult> {
|
|
const { session, text, chunks, metadata } = params;
|
|
|
|
if (session.stopped) {
|
|
logVerbose("slack-stream: stream already stopped, ignoring duplicate stop");
|
|
return {};
|
|
}
|
|
|
|
session.stopped = true;
|
|
if (text) {
|
|
session.pendingText += text;
|
|
}
|
|
|
|
logVerbose(
|
|
`slack-stream: stopping stream in ${session.channel} thread=${session.threadTs}${
|
|
text ? ` (final text: ${text.length} chars)` : ""
|
|
}`,
|
|
);
|
|
|
|
try {
|
|
const stopResponse = await session.streamer.stop(
|
|
text || chunks?.length || metadata
|
|
? {
|
|
...(text ? { markdown_text: text } : {}),
|
|
...(chunks?.length ? { chunks } : {}),
|
|
...(metadata ? { metadata } : {}),
|
|
}
|
|
: undefined,
|
|
);
|
|
session.delivered = true;
|
|
session.pendingText = "";
|
|
logVerbose("slack-stream: stream stopped");
|
|
// `chat.stopStream` reports the finalized message `ts` at the top level
|
|
// (and on `message.ts`); prefer the former and fall back to the latter.
|
|
const messageId = stopResponse?.ts ?? stopResponse?.message?.ts;
|
|
return messageId ? { messageId } : {};
|
|
} catch (err) {
|
|
const code = extractSlackErrorCode(err) ?? "unknown";
|
|
const benignFinalizeError = isBenignSlackFinalizeError(err);
|
|
if (session.pendingText && (benignFinalizeError || code === "missing_scope")) {
|
|
// stop() can be the first network call for short replies. Recipient or
|
|
// custom-authorship rejection means nothing landed; preserve the fallback.
|
|
throw new SlackStreamNotDeliveredError(session.pendingText, code);
|
|
}
|
|
if (benignFinalizeError) {
|
|
if (session.delivered) {
|
|
logVerbose(
|
|
`slack-stream: finalize rejected by Slack (${code}); prior appends delivered, treating stream as stopped`,
|
|
);
|
|
return {};
|
|
}
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Finalize error classification
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Slack API error codes that indicate `chat.stopStream` (or the
|
|
* `chat.startStream` call the SDK issues inside `stop()` when the buffer
|
|
* never flushed) cannot finalize the stream for the current recipient or
|
|
* team. Either the caller falls back to a normal message (see
|
|
* {@link SlackStreamNotDeliveredError}) or, if prior appends already
|
|
* delivered text, the error is logged verbosely and swallowed.
|
|
*/
|
|
const BENIGN_SLACK_FINALIZE_ERROR_CODES = new Set<string>([
|
|
// Slack Connect recipients: finalize fails because the external user id
|
|
// is not resolvable in the host workspace (#70295).
|
|
"user_not_found",
|
|
// Slack Connect team mismatch in shared channels.
|
|
"team_not_found",
|
|
// DMs that closed between stream start and stop.
|
|
"missing_recipient_user_id",
|
|
// Channels where Slack accepts ordinary messages but not native streaming.
|
|
"method_not_supported_for_channel_type",
|
|
]);
|
|
|
|
function isBenignSlackFinalizeError(err: unknown): boolean {
|
|
const code = extractSlackErrorCode(err);
|
|
return code !== undefined && BENIGN_SLACK_FINALIZE_ERROR_CODES.has(code);
|
|
}
|
|
|
|
function extractSlackErrorCode(err: unknown): string | undefined {
|
|
if (!err || typeof err !== "object") {
|
|
return undefined;
|
|
}
|
|
const record = err as Record<string, unknown>;
|
|
// @slack/web-api errors expose `data.error` with the Slack error code.
|
|
if (record.data && typeof record.data === "object") {
|
|
const inner = (record.data as Record<string, unknown>).error;
|
|
if (typeof inner === "string") {
|
|
return inner;
|
|
}
|
|
}
|
|
// Fallback: parse from message string ("An API error occurred: user_not_found").
|
|
const message = typeof record.message === "string" ? record.message : "";
|
|
const match = message.match(/An API error occurred:\s*([a-z_][a-z0-9_]*)/i);
|
|
return match?.[1];
|
|
}
|
|
|
|
export function markSlackStreamFallbackDelivered(session: SlackStreamSession): void {
|
|
const nativeStreamWasStarted = session.delivered;
|
|
session.pendingText = "";
|
|
// @slack/web-api 7.16.0 retains its private buffer after a failed flush.
|
|
// Clear fallback-owned text before retrying stop(), or the SDK resends it.
|
|
(session.streamer as unknown as { buffer: string }).buffer = "";
|
|
// A visible native stream still needs stop() to leave streaming state. If no
|
|
// native call succeeded, there is no Slack stream to finalize.
|
|
session.stopped = !nativeStreamWasStarted;
|
|
}
|