Merge remote-tracking branch 'prhead/feat/slack-text-streaming'

# Conflicts:
#	docs/channels/slack.md
#	src/config/types.slack.ts
#	src/slack/monitor/message-handler/dispatch.ts

src/slack/monitor/message-handler/dispatch.ts

@@ -1,3 +1,6 @@
+import type { ReplyPayload } from "../../../auto-reply/types.js";
+import type { SlackStreamSession } from "../../streaming.js";
+import type { PreparedSlackMessage } from "./types.js";
 import { resolveHumanDelayConfig } from "../../../agents/identity.js";
 import { dispatchInboundMessage } from "../../../auto-reply/dispatch.js";
 import { clearHistoryEntriesIfEnabled } from "../../../auto-reply/reply/history.js";
@@ -15,9 +18,44 @@ import {
   buildStatusFinalPreviewText,
   resolveSlackStreamMode,
 } from "../../stream-mode.js";
+import { appendSlackStream, startSlackStream, stopSlackStream } from "../../streaming.js";
 import { resolveSlackThreadTargets } from "../../threading.js";
-import { createSlackReplyDeliveryPlan, deliverReplies } from "../replies.js";
-import type { PreparedSlackMessage } from "./types.js";
+import { createSlackReplyDeliveryPlan, deliverReplies, resolveSlackThreadTs } from "../replies.js";
+
+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 resolveSlackStreamingThreadHint(params: {
+  replyToMode: "off" | "first" | "all";
+  incomingThreadTs: string | undefined;
+  messageTs: string | undefined;
+}): string | undefined {
+  return resolveSlackThreadTs({
+    replyToMode: params.replyToMode,
+    incomingThreadTs: params.incomingThreadTs,
+    messageTs: params.messageTs,
+    hasReplied: false,
+  });
+}
+
+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 dispatchPreparedSlackMessage(prepared: PreparedSlackMessage) {
   const { ctx, account, message, route } = prepared;
@@ -108,10 +146,84 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
     accountId: route.accountId,
   });
 
+  const streamingEnabled = isSlackStreamingEnabled(account.config.streaming);
+  const streamThreadHint = resolveSlackStreamingThreadHint({
+    replyToMode: ctx.replyToMode,
+    incomingThreadTs,
+    messageTs,
+  });
+  const useStreaming = shouldUseStreaming({
+    streamingEnabled,
+    threadTs: streamThreadHint,
+  });
+  let streamSession: SlackStreamSession | null = null;
+  let streamFailed = false;
+
+  const deliverNormally = async (payload: ReplyPayload, forcedThreadTs?: string): Promise<void> => {
+    const replyThreadTs = forcedThreadTs ?? replyPlan.nextThreadTs();
+    await deliverReplies({
+      replies: [payload],
+      target: prepared.replyTarget,
+      token: ctx.botToken,
+      accountId: account.accountId,
+      runtime,
+      textLimit: ctx.textLimit,
+      replyThreadTs,
+    });
+    replyPlan.markSent();
+  };
+
+  const deliverWithStreaming = async (payload: ReplyPayload): Promise<void> => {
+    if (streamFailed || hasMedia(payload) || !payload.text?.trim()) {
+      await deliverNormally(payload, streamSession?.threadTs);
+      return;
+    }
+
+    const text = payload.text.trim();
+    try {
+      if (!streamSession) {
+        const streamThreadTs = replyPlan.nextThreadTs();
+        if (!streamThreadTs) {
+          logVerbose(
+            "slack-stream: no reply thread target for stream start, falling back to normal delivery",
+          );
+          streamFailed = true;
+          await deliverNormally(payload);
+          return;
+        }
+
+        streamSession = await startSlackStream({
+          client: ctx.app.client,
+          channel: message.channel,
+          threadTs: streamThreadTs,
+          text,
+        });
+        replyPlan.markSent();
+        return;
+      }
+
+      await appendSlackStream({
+        session: streamSession,
+        text: "\n" + text,
+      });
+    } catch (err) {
+      runtime.error?.(
+        danger(`slack-stream: streaming API call failed: ${String(err)}, falling back`),
+      );
+      streamFailed = true;
+      await deliverNormally(payload, streamSession?.threadTs);
+    }
+  };
+
   const { dispatcher, replyOptions, markDispatchIdle } = createReplyDispatcherWithTyping({
     ...prefixOptions,
     humanDelay: resolveHumanDelayConfig(cfg, route.agentId),
     deliver: async (payload) => {
+      if (useStreaming) {
+        await deliverWithStreaming(payload);
+        return;
+      }
+
       const mediaCount = payload.mediaUrls?.length ?? (payload.mediaUrl ? 1 : 0);
       const draftMessageId = draftStream?.messageId();
       const draftChannelId = draftStream?.channelId();
@@ -239,38 +351,57 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
       ...replyOptions,
       skillFilter: prepared.channelConfig?.skills,
       hasRepliedRef,
-      disableBlockStreaming:
-        typeof account.config.blockStreaming === "boolean"
+      disableBlockStreaming: useStreaming
+        ? false
+        : typeof account.config.blockStreaming === "boolean"
           ? !account.config.blockStreaming
           : undefined,
       onModelSelected,
-      onPartialReply: async (payload) => {
-        updateDraftFromPartial(payload.text);
-      },
-      onAssistantMessageStart: async () => {
-        if (hasStreamedMessage) {
-          draftStream.forceNewMessage();
-          hasStreamedMessage = false;
-          appendRenderedText = "";
-          appendSourceText = "";
-          statusUpdateCount = 0;
-        }
-      },
-      onReasoningEnd: async () => {
-        if (hasStreamedMessage) {
-          draftStream.forceNewMessage();
-          hasStreamedMessage = false;
-          appendRenderedText = "";
-          appendSourceText = "";
-          statusUpdateCount = 0;
-        }
-      },
+      onPartialReply: useStreaming
+        ? undefined
+        : async (payload) => {
+            updateDraftFromPartial(payload.text);
+          },
+      onAssistantMessageStart: useStreaming
+        ? 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;
+            }
+          },
     },
   });
   await draftStream.flush();
   draftStream.stop();
   markDispatchIdle();
 
+  // -----------------------------------------------------------------------
+  // Finalize the stream if one was started
+  // -----------------------------------------------------------------------
+  const finalStream = streamSession as SlackStreamSession | null;
+  if (finalStream && !finalStream.stopped) {
+    try {
+      await stopSlackStream({ session: finalStream });
+    } catch (err) {
+      runtime.error?.(danger(`slack-stream: failed to stop stream: ${String(err)}`));
+    }
+  }
+
   const anyReplyDelivered = queuedFinal || (counts.block ?? 0) > 0 || (counts.final ?? 0) > 0;
 
   if (!anyReplyDelivered) {