diff --git a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts index 2e5a26349955..721ac89a1830 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts @@ -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 () => { diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index 0f3be9a31410..fe5e63524953 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.ts @@ -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, @@ -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 @@ -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, @@ -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, diff --git a/extensions/slack/src/streaming.test.ts b/extensions/slack/src/streaming.test.ts index de27335b755e..3a8a95c1364d 100644 --- a/extensions/slack/src/streaming.test.ts +++ b/extensions/slack/src/streaming.test.ts @@ -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); @@ -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 diff --git a/extensions/slack/src/streaming.ts b/extensions/slack/src/streaming.ts index 44eff4f597b5..d560928fd632 100644 --- a/extensions/slack/src/streaming.ts +++ b/extensions/slack/src/streaming.ts @@ -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 @@ -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`. @@ -113,7 +116,18 @@ export class SlackStreamNotDeliveredError extends Error { export async function startSlackStream( params: StartSlackStreamParams, ): Promise { - 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}` : ""}`, @@ -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 = { @@ -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`,