Skip to content

Commit 42d9d16

Browse files
vincentkocvincentkoc
authored
Merge 0d8d759 into 4c3b15b 
2 parents 4c3b15b + 0d8d759 commit 42d9d16

4 files changed

Lines changed: 134 additions & 21 deletions

File tree

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

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ const finalizeSlackPreviewEditMock = vi.fn(async () => {});
1313
const postMessageMock = vi.fn(async () => ({ ok: true, ts: "171234.999" }));
1414
const chatUpdateMock = vi.fn(async () => ({ ok: true, ts: "171234.999" }));
1515
const recordInboundSessionMock = vi.fn(async () => undefined);
16+
const recordSlackThreadParticipationMock = vi.fn();
1617
const updateLastRouteMock = vi.fn(async () => {});
1718
const appendSlackStreamMock = vi.fn(async () => {});
1819
const startSlackStreamMock = vi.fn(async () => ({
@@ -830,7 +831,7 @@ vi.mock("../../limits.js", () => ({
830831
}));
831832

832833
vi.mock("../../sent-thread-cache.js", () => ({
833-
recordSlackThreadParticipation: () => {},
834+
recordSlackThreadParticipation: recordSlackThreadParticipationMock,
834835
}));
835836

836837
vi.mock("../../stream-mode.js", () => ({
@@ -1228,6 +1229,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {
12281229
postMessageMock.mockClear();
12291230
chatUpdateMock.mockClear();
12301231
recordInboundSessionMock.mockReset();
1232+
recordSlackThreadParticipationMock.mockReset();
12311233
updateLastRouteMock.mockReset();
12321234
appendSlackStreamMock.mockReset();
12331235
startSlackStreamMock.mockReset();
@@ -3506,6 +3508,67 @@ describe("dispatchPreparedSlackMessage preview fallback", () => {
35063508
expect(session.stopped).toBe(true);
35073509
});
35083510

3511+
it("routes pending native stream text through chunked sender for unexpected finalize failures", async () => {
3512+
mockedNativeStreaming = true;
3513+
const session = {
3514+
channel: "C123",
3515+
threadTs: THREAD_TS,
3516+
stopped: false,
3517+
delivered: false,
3518+
pendingText: FINAL_REPLY_TEXT,
3519+
};
3520+
startSlackStreamMock.mockResolvedValueOnce(session);
3521+
stopSlackStreamMock.mockRejectedValueOnce(
3522+
new TestSlackStreamNotDeliveredError(
3523+
FINAL_REPLY_TEXT,
3524+
"method_not_supported_for_channel_type",
3525+
),
3526+
);
3527+
3528+
await dispatchPreparedSlackMessage(createPreparedSlackMessage());
3529+
3530+
expect(postMessageMock).not.toHaveBeenCalled();
3531+
expect(deliverRepliesMock).toHaveBeenCalledTimes(1);
3532+
expect(deliverRepliesMock).toHaveBeenCalledWith(
3533+
expect.objectContaining({
3534+
replyThreadTs: THREAD_TS,
3535+
replies: [expect.objectContaining({ text: FINAL_REPLY_TEXT })],
3536+
}),
3537+
);
3538+
expect(session.stopped).toBe(true);
3539+
});
3540+
3541+
it("fails dispatch when an unexpected finalize fallback cannot deliver a buffered tail", async () => {
3542+
mockedNativeStreaming = true;
3543+
const session = {
3544+
channel: "C123",
3545+
threadTs: THREAD_TS,
3546+
stopped: false,
3547+
delivered: true,
3548+
pendingText: "buffered tail",
3549+
};
3550+
startSlackStreamMock.mockResolvedValueOnce(session);
3551+
stopSlackStreamMock.mockRejectedValueOnce(
3552+
new TestSlackStreamNotDeliveredError(
3553+
"buffered tail",
3554+
"method_not_supported_for_channel_type",
3555+
),
3556+
);
3557+
deliverRepliesMock.mockRejectedValueOnce(new Error("fallback send failed"));
3558+
3559+
await expect(dispatchPreparedSlackMessage(createPreparedSlackMessage())).rejects.toThrowError(
3560+
"slack-stream not delivered: method_not_supported_for_channel_type",
3561+
);
3562+
3563+
expectDeliverReplyCall(0, "buffered tail");
3564+
expect(recordSlackThreadParticipationMock).toHaveBeenCalledWith(
3565+
expect.any(String),
3566+
"C123",
3567+
THREAD_TS,
3568+
expect.any(Object),
3569+
);
3570+
});
3571+
35093572
it("routes all pending native stream text through chunked sender when an append flush fails", async () => {
35103573
mockedNativeStreaming = true;
35113574
mockedDispatchSequence = [

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

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2043,11 +2043,17 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
20432043
} catch (err) {
20442044
if (err instanceof SlackStreamNotDeliveredError) {
20452045
streamFallbackDelivered = await deliverPendingStreamFallback(finalStream, err);
2046+
if (!streamFallbackDelivered) {
2047+
dispatchError ??= err;
2048+
}
20462049
} else {
20472050
const error = formatSlackError(err);
20482051
emitAcknowledgedStreamedDeliveries();
20492052
emitFailedPendingStreamedDeliveries(error);
20502053
runtime.error?.(danger(`slack-stream: failed to stop stream: ${error}`));
2054+
if (!finalStream.delivered) {
2055+
dispatchError ??= err;
2056+
}
20512057
}
20522058
}
20532059
}
@@ -2105,10 +2111,6 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
21052111
}
21062112
}
21072113

2108-
if (dispatchError) {
2109-
throw toLintErrorObject(dispatchError, "Slack dispatch failed");
2110-
}
2111-
21122114
// Record thread participation only when we actually delivered a reply and
21132115
// know the thread ts that was used (set by deliverNormally, streaming start,
21142116
// or draft stream). Falls back to statusThreadTs for edge cases.
@@ -2118,6 +2120,9 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
21182120
agentId: route.agentId,
21192121
});
21202122
}
2123+
if (dispatchError) {
2124+
throw toLintErrorObject(dispatchError, "Slack dispatch failed");
2125+
}
21212126
if (!anyReplyDelivered && !draftPreviewCommitted) {
21222127
await draftStream?.clear();
21232128
return;

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

Lines changed: 44 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,37 @@ describe("stopSlackStream finalize error handling", () => {
132132
expect((thrown as SlackStreamNotDeliveredError).pendingText).toBe("hello world");
133133
});
134134

135+
it("throws SlackStreamNotDeliveredError for unexpected finalize codes while text is buffered", async () => {
136+
const session = makeSession({
137+
appendImpl: async () => null,
138+
stopImpl: async () => {
139+
throw slackApiError("method_not_supported_for_channel_type");
140+
},
141+
});
142+
await appendSlackStream({ session, text: "short thread reply" });
143+
144+
const thrown = await stopSlackStream({ session }).catch((err: unknown) => err);
145+
146+
expect(thrown).toBeInstanceOf(SlackStreamNotDeliveredError);
147+
expect((thrown as SlackStreamNotDeliveredError).slackCode).toBe(
148+
"method_not_supported_for_channel_type",
149+
);
150+
expect((thrown as SlackStreamNotDeliveredError).pendingText).toBe("short thread reply");
151+
});
152+
153+
it("does not retry ambiguous transport failures while text is buffered", async () => {
154+
const session = makeSession({
155+
appendImpl: async () => null,
156+
stopImpl: async () => {
157+
throw new Error("socket reset");
158+
},
159+
});
160+
await appendSlackStream({ session, text: "locally buffered reply" });
161+
162+
await expect(stopSlackStream({ session })).rejects.toThrow("socket reset");
163+
expect(session.pendingText).toBe("locally buffered reply");
164+
});
165+
135166
it("clears pendingText after an append flush is acknowledged by Slack", async () => {
136167
const session = makeSession({
137168
appendImpl: async () => ({ ts: "1700000000.100203" }),
@@ -372,13 +403,24 @@ describe("stopSlackStream finalize error handling", () => {
372403

373404
describe("error classification", () => {
374405
it("isBenignSlackFinalizeError matches each allowlisted code", () => {
375-
for (const code of ["user_not_found", "team_not_found", "missing_recipient_user_id"]) {
406+
for (const code of [
407+
"user_not_found",
408+
"team_not_found",
409+
"missing_recipient_user_id",
410+
"method_not_supported_for_channel_type",
411+
]) {
376412
expect(isBenignSlackFinalizeError(slackApiError(code))).toBe(true);
377413
}
378414
});
379415

380416
it("isBenignSlackFinalizeError rejects non-listed codes", () => {
381-
for (const code of ["not_authed", "ratelimited", "channel_not_found"]) {
417+
for (const code of [
418+
"not_authed",
419+
"ratelimited",
420+
"channel_not_found",
421+
"internal_error",
422+
"fatal_error",
423+
]) {
382424
expect(isBenignSlackFinalizeError(slackApiError(code))).toBe(false);
383425
}
384426
});

‎extensions/slack/src/streaming.ts‎

Lines changed: 17 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -80,17 +80,16 @@ type StopSlackStreamParams = {
8080
};
8181

8282
/**
83-
* Thrown when Slack rejects a stream flush/finalize with a recipient-resolution
84-
* error (see {@link BENIGN_SLACK_FINALIZE_ERROR_CODES}) while text is still
85-
* only buffered locally by the Slack SDK. Carries the pending text so the
83+
* Thrown when Slack definitively rejects a stream flush/finalize while text
84+
* remains buffered locally by the Slack SDK. Carries the pending text so the
8685
* caller can deliver it via the normal Slack reply path.
8786
*/
8887
export class SlackStreamNotDeliveredError extends Error {
8988
readonly pendingText: string;
9089
readonly slackCode: string;
9190
constructor(pendingText: string, slackCode: string) {
9291
super(
93-
`slack-stream: finalize failed with ${slackCode} before any text reached Slack ` +
92+
`slack-stream: finalize failed with ${slackCode} before buffered text reached Slack ` +
9493
`(${pendingText.length} chars pending)`,
9594
);
9695
this.name = "SlackStreamNotDeliveredError";
@@ -235,17 +234,18 @@ export type StopSlackStreamResult = {
235234
* After calling this the stream message becomes a normal Slack message.
236235
* Optionally include final text to append before stopping.
237236
*
238-
* If Slack's `chat.stopStream` responds with a known benign finalize error
239-
* (see {@link BENIGN_SLACK_FINALIZE_ERROR_CODES}) AND any prior `append`
240-
* has already landed on Slack, the error is swallowed and the session is
241-
* marked stopped - the already-delivered text stays visible.
237+
* If Slack's `chat.stopStream` responds with a definitive recipient/channel
238+
* rejection while text is still buffered locally, this function throws a
239+
* {@link SlackStreamNotDeliveredError} carrying that pending text so the caller
240+
* can deliver it through the normal Slack reply path. Ambiguous failures
241+
* propagate unchanged because Slack may have committed the request.
242242
*
243-
* If the same benign error fires while text is still only buffered locally
244-
* (e.g. short replies that never exceeded the SDK's buffer_size), this
245-
* function throws a {@link SlackStreamNotDeliveredError} carrying that pending
246-
* text so the caller can deliver it through the normal Slack reply path.
243+
* If Slack responds with a known benign finalize error (see
244+
* {@link BENIGN_SLACK_FINALIZE_ERROR_CODES}) after prior `append` calls already
245+
* landed, the error is swallowed and the session is marked stopped - the
246+
* already-delivered text stays visible.
247247
*
248-
* All other errors propagate unchanged.
248+
* Errors without buffered text propagate unchanged.
249249
*
250250
* On success, returns the finalized message's Slack `ts` (when reported) so the
251251
* caller can emit the `message_sent` hook with a populated `messageId`.
@@ -293,7 +293,8 @@ export async function stopSlackStream(
293293
const code = extractSlackErrorCode(err) ?? "unknown";
294294
if (session.pendingText) {
295295
// stop() can be the first network call for short replies. If Slack
296-
// Connect rejects it, the user has not seen the SDK-buffered text yet.
296+
// definitively rejects that finalize, the user has not seen the
297+
// SDK-buffered text. Let the caller fall back to chat.postMessage.
297298
throw new SlackStreamNotDeliveredError(session.pendingText, code);
298299
}
299300
if (session.delivered) {
@@ -327,6 +328,8 @@ const BENIGN_SLACK_FINALIZE_ERROR_CODES = new Set<string>([
327328
"team_not_found",
328329
// DMs that closed between stream start and stop.
329330
"missing_recipient_user_id",
331+
// Channels where Slack accepts ordinary messages but not native streaming.
332+
"method_not_supported_for_channel_type",
330333
]);
331334

332335
export function isBenignSlackFinalizeError(err: unknown): boolean {

0 commit comments

Comments
 (0)