Fix Slack native streaming thread replies missing recipient ids

AsserAl1012 · AsserAl1012 · commit 7b28b33e08cc · 2026-02-18T20:33:44.000Z
diff --git a/src/slack/monitor/message-handler/dispatch.streaming.test.ts b/src/slack/monitor/message-handler/dispatch.streaming.test.ts
@@ -1,5 +1,9 @@
 import { describe, expect, it } from "vitest";
-import { isSlackStreamingEnabled, resolveSlackStreamingThreadHint } from "./dispatch.js";
+import {
+  isSlackStreamingEnabled,
+  resolveSlackStreamingThreadHint,
+  shouldUseSlackNativeStreaming,
+} from "./dispatch.js";
 
 describe("slack native streaming defaults", () => {
   it("is enabled when config is undefined", () => {
@@ -43,3 +47,27 @@ describe("slack native streaming thread hint", () => {
     ).toBe("2000.1");
   });
 });
+
+describe("slack native streaming recipient gating", () => {
+  it("is disabled when team id is missing", () => {
+    expect(
+      shouldUseSlackNativeStreaming({
+        streamingEnabled: true,
+        threadTs: "1234.56",
+        recipientTeamId: undefined,
+        recipientUserId: "U123",
+      }),
+    ).toBe(false);
+  });
+
+  it("is disabled when user id is missing", () => {
+    expect(
+      shouldUseSlackNativeStreaming({
+        streamingEnabled: true,
+        threadTs: "1234.56",
+        recipientTeamId: "T123",
+        recipientUserId: undefined,
+      }),
+    ).toBe(false);
+  });
+});
diff --git a/src/slack/monitor/message-handler/dispatch.ts b/src/slack/monitor/message-handler/dispatch.ts
@@ -43,15 +43,25 @@ export function resolveSlackStreamingThreadHint(params: {
   });
 }
 
-function shouldUseStreaming(params: {
+export function shouldUseSlackNativeStreaming(params: {
   streamingEnabled: boolean;
   threadTs: string | undefined;
+  recipientTeamId: string | undefined;
+  recipientUserId: string | undefined;
 }): boolean {
   if (!params.streamingEnabled) {
     return false;
   }
   if (!params.threadTs) {
-    logVerbose("slack-stream: streaming disabled — no reply thread target available");
+    logVerbose("slack-stream: streaming disabled - no reply thread target available");
+    return false;
+  }
+  if (!params.recipientTeamId) {
+    logVerbose("slack-stream: streaming disabled - missing recipient team id");
+    return false;
+  }
+  if (!params.recipientUserId) {
+    logVerbose("slack-stream: streaming disabled - missing recipient user id");
     return false;
   }
   return true;
@@ -152,9 +162,13 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
     incomingThreadTs,
     messageTs,
   });
-  const useStreaming = shouldUseStreaming({
+  const recipientTeamId = ctx.teamId.trim() || undefined;
+  const recipientUserId = message.user?.trim() || undefined;
+  const useStreaming = shouldUseSlackNativeStreaming({
     streamingEnabled,
     threadTs: streamThreadHint,
+    recipientTeamId,
+    recipientUserId,
   });
   let streamSession: SlackStreamSession | null = null;
   let streamFailed = false;
@@ -194,10 +208,21 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
           return;
         }
 
+        if (!recipientTeamId || !recipientUserId) {
+          logVerbose(
+            "slack-stream: missing recipient ids for stream start, falling back to normal delivery",
+          );
+          streamFailed = true;
+          await deliverNormally(payload, streamThreadTs);
+          return;
+        }
+
         streamSession = await startSlackStream({
           client: ctx.app.client,
-          channel: message.channel,
+          channelId: message.channel,
           threadTs: streamThreadTs,
+          recipientTeamId,
+          recipientUserId,
           text,
         });
         replyPlan.markSent();
diff --git a/src/slack/streaming.test.ts b/src/slack/streaming.test.ts
@@ -0,0 +1,57 @@
+import type { WebClient } from "@slack/web-api";
+import { describe, expect, it, vi } from "vitest";
+import { startSlackStream, stopSlackStream } from "./streaming.js";
+
+describe("startSlackStream", () => {
+  it("passes recipient ids to chatStream", async () => {
+    const streamer = {
+      append: vi.fn(),
+      stop: vi.fn(),
+    };
+    const chatStream = vi.fn().mockReturnValue(streamer);
+    const client = {
+      chatStream,
+    } as unknown as WebClient;
+
+    await startSlackStream({
+      client,
+      channelId: "C123",
+      threadTs: "1730.1",
+      recipientTeamId: "T123",
+      recipientUserId: "U123",
+    });
+
+    expect(chatStream).toHaveBeenCalledWith(
+      expect.objectContaining({
+        channel: "C123",
+        thread_ts: "1730.1",
+        recipient_team_id: "T123",
+        recipient_user_id: "U123",
+      }),
+    );
+  });
+});
+
+describe("stopSlackStream", () => {
+  it("does not throw when Slack rejects stop", async () => {
+    const streamer = {
+      append: vi.fn(),
+      stop: vi.fn().mockRejectedValue(new Error("stop failed")),
+    };
+    const chatStream = vi.fn().mockReturnValue(streamer);
+    const client = {
+      chatStream,
+    } as unknown as WebClient;
+
+    const session = await startSlackStream({
+      client,
+      channelId: "C123",
+      threadTs: "1730.2",
+      recipientTeamId: "T123",
+      recipientUserId: "U123",
+    });
+
+    await expect(stopSlackStream({ session })).resolves.toBeUndefined();
+    expect(streamer.stop).toHaveBeenCalledTimes(1);
+  });
+});
diff --git a/src/slack/streaming.ts b/src/slack/streaming.ts
@@ -32,8 +32,10 @@ export type SlackStreamSession = {
 
 export type StartSlackStreamParams = {
   client: WebClient;
-  channel: string;
+  channelId: string;
   threadTs: string;
+  recipientTeamId: string;
+  recipientUserId: string;
   /** Optional initial markdown text to include in the stream start. */
   text?: string;
 };
@@ -64,18 +66,20 @@ export type StopSlackStreamParams = {
 export async function startSlackStream(
   params: StartSlackStreamParams,
 ): Promise<SlackStreamSession> {
-  const { client, channel, threadTs, text } = params;
+  const { client, channelId, threadTs, recipientTeamId, recipientUserId, text } = params;
 
-  logVerbose(`slack-stream: starting stream in ${channel} thread=${threadTs}`);
+  logVerbose(`slack-stream: starting stream in ${channelId} thread=${threadTs}`);
 
   const streamer = client.chatStream({
-    channel,
+    channel: channelId,
     thread_ts: threadTs,
+    recipient_team_id: recipientTeamId,
+    recipient_user_id: recipientUserId,
   });
 
   const session: SlackStreamSession = {
     streamer,
-    channel,
+    channel: channelId,
     threadTs,
     stopped: false,
   };
@@ -131,7 +135,10 @@ export async function stopSlackStream(params: StopSlackStreamParams): Promise<vo
     }`,
   );
 
-  await session.streamer.stop(text ? { markdown_text: text } : undefined);
-
-  logVerbose("slack-stream: stream stopped");
+  try {
+    await session.streamer.stop(text ? { markdown_text: text } : undefined);
+    logVerbose("slack-stream: stream stopped");
+  } catch (err) {
+    logVerbose(`slack-stream: failed to stop stream (${String(err)})`);
+  }
 }
