Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -1286,20 +1286,24 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {

await dispatchPreparedSlackMessage(createPreparedSlackMessage({ relayIdentity }));

expect(createSlackDraftStreamMock).not.toHaveBeenCalled();
expect(finalizeSlackPreviewEditMock).not.toHaveBeenCalled();
expect(deliverRepliesMock).toHaveBeenCalledTimes(1);
expectDeliverReplyCall(0, FINAL_REPLY_TEXT, { identity: relayIdentity });
});

it("does not use native Slack streaming when a custom identity is active", async () => {
it("uses supported native Slack streaming authorship when a custom identity is active", async () => {
mockedNativeStreaming = true;
const relayIdentity = { username: "Nik Team Claw" };

await dispatchPreparedSlackMessage(createPreparedSlackMessage({ relayIdentity }));

expect(startSlackStreamMock).not.toHaveBeenCalled();
expect(createSlackDraftStreamMock).toHaveBeenCalledTimes(1);
expect(deliverRepliesMock).toHaveBeenCalledTimes(1);
expectDeliverReplyCall(0, FINAL_REPLY_TEXT, { identity: relayIdentity });
expectMockCallArgFields(startSlackStreamMock, 0, "Slack stream start params", {
text: FINAL_REPLY_TEXT,
identity: relayIdentity,
});
expect(createSlackDraftStreamMock).not.toHaveBeenCalled();
expect(deliverRepliesMock).not.toHaveBeenCalled();
});

it("does not create a Slack thread for top-level messages when replyToMode is off", async () => {
Expand Down
21 changes: 13 additions & 8 deletions extensions/slack/src/monitor/message-handler/dispatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -696,11 +696,10 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
shouldEnableSlackPreviewStreaming({
mode: slackStreaming.mode,
});
// Slack's native streaming APIs do not accept chat:write.customize identity
// fields. Keep custom-identity replies on the draft/standard postMessage
// path so the configured username and icon are not silently discarded.
const hasSlackCustomIdentity = Boolean(
slackIdentity?.username || slackIdentity?.iconUrl || slackIdentity?.iconEmoji,
);
const streamingEnabled =
!slackIdentity &&
!sourceRepliesAreToolOnly &&
isSlackStreamingEnabled({
mode: slackStreaming.mode,
Expand All @@ -711,10 +710,14 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
streamingEnabled,
threadTs: streamThreadHint,
});
const shouldUseDraftStream = shouldInitializeSlackDraftStream({
previewStreamingEnabled,
useStreaming,
});
// 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
Expand Down Expand Up @@ -1105,6 +1108,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
channel: message.channel,
threadTs: streamThreadTs,
text,
...(slackIdentity ? { identity: slackIdentity } : {}),
teamId: await resolveSlackStreamRecipientTeamId({
client: ctx.app.client,
token: ctx.botToken,
Expand Down Expand Up @@ -1610,6 +1614,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
threadTs: streamThreadTs,
chunks,
taskDisplayMode: "plan",
...(slackIdentity ? { identity: slackIdentity } : {}),
teamId: await resolveSlackStreamRecipientTeamId({
client: ctx.app.client,
token: ctx.botToken,
Expand Down
17 changes: 17 additions & 0 deletions extensions/slack/src/streaming.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,15 @@ describe("stopSlackStream finalize error handling", () => {
threadTs: "1700000000.000100",
chunks,
taskDisplayMode: "plan",
identity: { username: "Research Agent", iconEmoji: ":mag:" },
});

expect(client.chatStream).toHaveBeenCalledWith({
channel: "C123",
thread_ts: "1700000000.000100",
task_display_mode: "plan",
username: "Research Agent",
icon_emoji: ":mag:",
});
expect(append).toHaveBeenCalledWith({ chunks });
expect(session.delivered).toBe(true);
Expand Down Expand Up @@ -98,6 +101,20 @@ describe("stopSlackStream finalize error handling", () => {
expect(session.stopped).toBe(true);
});

it("falls back when deferred stream start rejects custom identity scope", async () => {
const session = makeSession({
stopImpl: async () => {
throw slackApiError("missing_scope");
},
});
session.pendingText = "short reply";

const thrown = await stopSlackStream({ session }).catch((error: unknown) => error);

expect(thrown).toBeInstanceOf(SlackStreamNotDeliveredError);
expect(thrown).toMatchObject({ pendingText: "short reply", slackCode: "missing_scope" });
});

it("throws SlackStreamNotDeliveredError when user_not_found fires before any flush", async () => {
const session = makeSession({
appendImpl: async () => null, // null => buffered, never hit Slack
Expand Down
33 changes: 24 additions & 9 deletions extensions/slack/src/streaming.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ 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
Expand Down Expand Up @@ -50,6 +51,8 @@ type StartSlackStreamParams = {
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`.
Expand Down Expand Up @@ -113,7 +116,18 @@ export class SlackStreamNotDeliveredError extends Error {
export async function startSlackStream(
params: StartSlackStreamParams,
): Promise<SlackStreamSession> {
const { client, channel, threadTs, text, chunks, taskDisplayMode, teamId, userId } = params;
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}` : ""}`,
Expand All @@ -125,6 +139,7 @@ export async function startSlackStream(
...(taskDisplayMode ? { task_display_mode: taskDisplayMode } : {}),
...(teamId ? { recipient_team_id: teamId } : {}),
...(userId ? { recipient_user_id: userId } : {}),
...identityPayload,
});

const session: SlackStreamSession = {
Expand Down Expand Up @@ -289,14 +304,14 @@ export async function stopSlackStream(
const messageId = stopResponse?.ts ?? stopResponse?.message?.ts;
return messageId ? { messageId } : {};
} catch (err) {
if (isBenignSlackFinalizeError(err)) {
const code = extractSlackErrorCode(err) ?? "unknown";
if (session.pendingText) {
// stop() can be the first network call for short replies. If Slack
// definitively rejects that finalize, the user has not seen the
// SDK-buffered text. Let the caller fall back to chat.postMessage.
throw new SlackStreamNotDeliveredError(session.pendingText, code);
}
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`,
Expand Down
Loading