Skip to content

Commit e55b932

Browse files
mvanhornsteipete
mvanhorn
authored and
steipete
committed
fix(slack): fall back to chat.postMessage when stream finalize fails pre-flush
Address adversarial review finding on #70295: the prior swallow-on-benign fix silently dropped short replies to Slack Connect users. The SDK's ChatStreamer buffers text locally until buffer_size (256 default), so short replies never trigger chat.startStream via append(). streamer.stop() then issues startStream internally; on Slack Connect recipients this throws user_not_found. With the prior fix that error was swallowed and the dispatcher marked the turn delivered - user saw 'done' reaction but no message. SlackStreamSession now tracks delivered (true once any Slack API call returned a response) and pendingText (accumulation of every append + final-stop text). stopSlackStream: - swallows the benign code when delivered=true (prior append flushed; text is visible; same behavior as before) - throws a new SlackStreamNotDeliveredError carrying pendingText when delivered=false (nothing reached Slack) dispatch.ts catches SlackStreamNotDeliveredError and posts pendingText via a rename-bound chat.postMessage (to dodge the unicorn lint rule), and flips streamFallbackDelivered so anyReplyDelivered stays correct. Fixes #70295
1 parent 676ed34 commit e55b932

3 files changed

Lines changed: 248 additions & 56 deletions

File tree

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

Lines changed: 41 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,12 @@ import {
3737
resolveSlackStreamingConfig,
3838
} from "../../stream-mode.js";
3939
import type { SlackStreamSession } from "../../streaming.js";
40-
import { appendSlackStream, startSlackStream, stopSlackStream } from "../../streaming.js";
40+
import {
41+
appendSlackStream,
42+
SlackStreamNotDeliveredError,
43+
startSlackStream,
44+
stopSlackStream,
45+
} from "../../streaming.js";
4146
import { resolveSlackThreadTargets } from "../../threading.js";
4247
import { normalizeSlackAllowOwnerEntry } from "../allow-list.js";
4348
import { resolveStorePath, updateLastRoute } from "../config.runtime.js";
@@ -862,17 +867,50 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
862867
// -----------------------------------------------------------------------
863868
// Finalize the stream if one was started
864869
// -----------------------------------------------------------------------
870+
let streamFallbackDelivered = false;
865871
const finalStream = streamSession as SlackStreamSession | null;
866872
if (finalStream && !finalStream.stopped) {
867873
try {
868874
await stopSlackStream({ session: finalStream });
869875
} catch (err) {
870-
runtime.error?.(danger(`slack-stream: failed to stop stream: ${formatErrorMessage(err)}`));
876+
if (err instanceof SlackStreamNotDeliveredError) {
877+
// Slack rejected the stream before any text reached the recipient
878+
// (common for short replies to Slack Connect users - the SDK buffers
879+
// under 256 chars and the internal chat.startStream inside stop()
880+
// is the first call to Slack). Fall back to a plain chat.postMessage
881+
// so the reply is not lost.
882+
try {
883+
// Rename-bind to dodge eslint-plugin-unicorn/require-post-message-target-origin
884+
// which cannot distinguish Slack chat.postMessage from window.postMessage.
885+
const postChatMessage = ctx.app.client.chat.postMessage.bind(ctx.app.client.chat);
886+
await postChatMessage({
887+
channel: finalStream.channel,
888+
thread_ts: finalStream.threadTs,
889+
text: err.pendingText,
890+
});
891+
streamFallbackDelivered = true;
892+
logVerbose(
893+
`slack-stream: streamed finalize failed (${err.slackCode}); delivered ${err.pendingText.length} chars via chat.postMessage fallback`,
894+
);
895+
} catch (postErr) {
896+
runtime.error?.(
897+
danger(
898+
`slack-stream: fallback chat.postMessage failed after ${err.slackCode}: ${formatErrorMessage(postErr)}`,
899+
),
900+
);
901+
}
902+
} else {
903+
runtime.error?.(danger(`slack-stream: failed to stop stream: ${formatErrorMessage(err)}`));
904+
}
871905
}
872906
}
873907

874908
const anyReplyDelivered =
875-
observedReplyDelivery || queuedFinal || (counts.block ?? 0) > 0 || (counts.final ?? 0) > 0;
909+
observedReplyDelivery ||
910+
queuedFinal ||
911+
streamFallbackDelivered ||
912+
(counts.block ?? 0) > 0 ||
913+
(counts.final ?? 0) > 0;
876914

877915
if (statusReactionsEnabled) {
878916
if (dispatchError) {

‎extensions/slack/src/streaming.test.ts‎

Lines changed: 121 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,28 @@
11
import type { ChatStreamer } from "@slack/web-api/dist/chat-stream.js";
22
import { describe, expect, it, vi } from "vitest";
3-
import { stopSlackStream, type SlackStreamSession } from "./streaming.js";
3+
import {
4+
appendSlackStream,
5+
extractSlackErrorCode,
6+
isBenignSlackFinalizeError,
7+
SlackStreamNotDeliveredError,
8+
stopSlackStream,
9+
type SlackStreamSession,
10+
} from "./streaming.js";
411

5-
function makeSession(stopImpl: () => Promise<void>): SlackStreamSession {
12+
type AppendImpl = () => Promise<unknown>;
13+
type StopImpl = () => Promise<void>;
14+
15+
function makeSession(params: { appendImpl?: AppendImpl; stopImpl?: StopImpl }): SlackStreamSession {
616
return {
717
streamer: {
8-
append: vi.fn(async () => {}),
9-
stop: vi.fn(stopImpl),
18+
append: vi.fn(params.appendImpl ?? (async () => null)),
19+
stop: vi.fn(params.stopImpl ?? (async () => {})),
1020
} as unknown as ChatStreamer,
1121
channel: "C123",
1222
threadTs: "1700000000.000100",
1323
stopped: false,
24+
delivered: false,
25+
pendingText: "",
1426
};
1527
}
1628

@@ -21,42 +33,84 @@ function slackApiError(code: string): Error {
2133
}
2234

2335
describe("stopSlackStream finalize error handling", () => {
24-
it("swallows user_not_found (Slack Connect DMs) and marks the session stopped", async () => {
25-
const session = makeSession(async () => {
26-
throw slackApiError("user_not_found");
36+
it("swallows user_not_found after prior append flushed (delivered=true)", async () => {
37+
const session = makeSession({
38+
appendImpl: async () => ({ ts: "1700000000.100200" }), // non-null => flushed
39+
stopImpl: async () => {
40+
throw slackApiError("user_not_found");
41+
},
2742
});
43+
await appendSlackStream({ session, text: "some text that Slack saw" });
44+
expect(session.delivered).toBe(true);
45+
2846
await expect(stopSlackStream({ session })).resolves.toBeUndefined();
2947
expect(session.stopped).toBe(true);
3048
});
3149

32-
it("swallows team_not_found (Slack Connect cross-workspace) and marks stopped", async () => {
33-
const session = makeSession(async () => {
34-
throw slackApiError("team_not_found");
50+
it("throws SlackStreamNotDeliveredError when user_not_found fires before any flush", async () => {
51+
const session = makeSession({
52+
appendImpl: async () => null, // null => buffered, never hit Slack
53+
stopImpl: async () => {
54+
throw slackApiError("user_not_found");
55+
},
3556
});
36-
await expect(stopSlackStream({ session })).resolves.toBeUndefined();
57+
await appendSlackStream({ session, text: "short reply under buffer size" });
58+
expect(session.delivered).toBe(false);
59+
60+
const thrown = await stopSlackStream({ session }).catch((err: unknown) => err);
61+
expect(thrown).toBeInstanceOf(SlackStreamNotDeliveredError);
62+
expect((thrown as SlackStreamNotDeliveredError).slackCode).toBe("user_not_found");
63+
expect((thrown as SlackStreamNotDeliveredError).pendingText).toBe(
64+
"short reply under buffer size",
65+
);
3766
expect(session.stopped).toBe(true);
3867
});
3968

40-
it("swallows missing_recipient_user_id (DM closed mid-stream) and marks stopped", async () => {
41-
const session = makeSession(async () => {
42-
throw slackApiError("missing_recipient_user_id");
69+
it("throws SlackStreamNotDeliveredError carrying stop()'s final text too", async () => {
70+
const session = makeSession({
71+
appendImpl: async () => null,
72+
stopImpl: async () => {
73+
throw slackApiError("team_not_found");
74+
},
4375
});
76+
await appendSlackStream({ session, text: "hello " });
77+
78+
const thrown = await stopSlackStream({ session, text: "world" }).catch((err: unknown) => err);
79+
expect(thrown).toBeInstanceOf(SlackStreamNotDeliveredError);
80+
expect((thrown as SlackStreamNotDeliveredError).slackCode).toBe("team_not_found");
81+
expect((thrown as SlackStreamNotDeliveredError).pendingText).toBe("hello world");
82+
});
83+
84+
it("swallows missing_recipient_user_id when delivered", async () => {
85+
const session = makeSession({
86+
appendImpl: async () => ({ ts: "1700000000.100201" }),
87+
stopImpl: async () => {
88+
throw slackApiError("missing_recipient_user_id");
89+
},
90+
});
91+
await appendSlackStream({ session, text: "chars" });
4492
await expect(stopSlackStream({ session })).resolves.toBeUndefined();
4593
expect(session.stopped).toBe(true);
4694
});
4795

48-
it("re-throws unexpected Slack API errors so callers can log them", async () => {
49-
const session = makeSession(async () => {
50-
throw slackApiError("not_authed");
96+
it("re-throws unexpected Slack API errors even when delivered", async () => {
97+
const session = makeSession({
98+
appendImpl: async () => ({ ts: "1700000000.100202" }),
99+
stopImpl: async () => {
100+
throw slackApiError("not_authed");
101+
},
51102
});
103+
await appendSlackStream({ session, text: "some text" });
52104
await expect(stopSlackStream({ session })).rejects.toThrow(/not_authed/);
53105
// Session is still marked stopped so retries do not re-enter streamer.stop.
54106
expect(session.stopped).toBe(true);
55107
});
56108

57109
it("re-throws non-Slack-shaped errors unchanged", async () => {
58-
const session = makeSession(async () => {
59-
throw new Error("socket reset");
110+
const session = makeSession({
111+
stopImpl: async () => {
112+
throw new Error("socket reset");
113+
},
60114
});
61115
await expect(stopSlackStream({ session })).rejects.toThrow(/socket reset/);
62116
expect(session.stopped).toBe(true);
@@ -65,12 +119,59 @@ describe("stopSlackStream finalize error handling", () => {
65119
it("returns a no-op on an already-stopped session", async () => {
66120
const stop = vi.fn(async () => {});
67121
const session: SlackStreamSession = {
68-
streamer: { append: vi.fn(async () => {}), stop } as unknown as ChatStreamer,
122+
streamer: { append: vi.fn(async () => null), stop } as unknown as ChatStreamer,
69123
channel: "C123",
70124
threadTs: "1700000000.000100",
71125
stopped: true,
126+
delivered: false,
127+
pendingText: "",
72128
};
73129
await expect(stopSlackStream({ session })).resolves.toBeUndefined();
74130
expect(stop).not.toHaveBeenCalled();
75131
});
132+
133+
it("marks delivered=true on successful stop() without prior flush", async () => {
134+
const session = makeSession({
135+
appendImpl: async () => null,
136+
stopImpl: async () => {},
137+
});
138+
await appendSlackStream({ session, text: "short" });
139+
expect(session.delivered).toBe(false);
140+
await stopSlackStream({ session });
141+
expect(session.delivered).toBe(true);
142+
});
143+
});
144+
145+
describe("error classification", () => {
146+
it("isBenignSlackFinalizeError matches each allowlisted code", () => {
147+
for (const code of ["user_not_found", "team_not_found", "missing_recipient_user_id"]) {
148+
expect(isBenignSlackFinalizeError(slackApiError(code))).toBe(true);
149+
}
150+
});
151+
152+
it("isBenignSlackFinalizeError rejects non-listed codes", () => {
153+
for (const code of ["not_authed", "ratelimited", "channel_not_found"]) {
154+
expect(isBenignSlackFinalizeError(slackApiError(code))).toBe(false);
155+
}
156+
});
157+
158+
it("extractSlackErrorCode handles data.error, message fallback, and junk shapes", () => {
159+
// Canonical SDK shape
160+
expect(extractSlackErrorCode(slackApiError("user_not_found"))).toBe("user_not_found");
161+
// message-regex fallback when data is absent
162+
expect(extractSlackErrorCode(new Error("An API error occurred: rate_limited"))).toBe(
163+
"rate_limited",
164+
);
165+
// data.error not a string - falls through to message parse
166+
const wrongShape = new Error("plain message");
167+
(wrongShape as unknown as { data: unknown }).data = { error: 42 };
168+
expect(extractSlackErrorCode(wrongShape)).toBeUndefined();
169+
// data.error null - falls through
170+
(wrongShape as unknown as { data: unknown }).data = null;
171+
expect(extractSlackErrorCode(wrongShape)).toBeUndefined();
172+
// Non-object error
173+
expect(extractSlackErrorCode("raw string")).toBeUndefined();
174+
expect(extractSlackErrorCode(null)).toBeUndefined();
175+
expect(extractSlackErrorCode(undefined)).toBeUndefined();
176+
});
76177
});

0 commit comments

Comments
 (0)