diff --git a/.changeset/slack-stream-chunk-mode-text.md b/.changeset/slack-stream-chunk-mode-text.md new file mode 100644 index 00000000..83b6d610 --- /dev/null +++ b/.changeset/slack-stream-chunk-mode-text.md @@ -0,0 +1,5 @@ +--- +"@chat-adapter/slack": patch +--- + +Fix Slack native streaming losing reply text after structured chunks. Once a stream segment had appended `plan_update` or `task_update` chunks, later text deltas were still sent through the `markdown_text` parameter, which Slack rejects with `streaming_mode_mismatch` in chunk mode, so the text never appeared in the final message. Text deltas on a segment that has appended structured chunks — including a segment opened by rotation, which replays the plan and open tasks — are now sent as `markdown_text` chunks. Text sent before any structured chunk is unchanged. diff --git a/packages/adapter-slack/src/index.test.ts b/packages/adapter-slack/src/index.test.ts index dce828af..87c08595 100644 --- a/packages/adapter-slack/src/index.test.ts +++ b/packages/adapter-slack/src/index.test.ts @@ -13896,7 +13896,10 @@ describe("native stream rotation", () => { expect(segments[0].stop).toHaveBeenCalledTimes(2); expect(segments[0].stop).toHaveBeenNthCalledWith(1, { chunks: [{ ...task, status: "complete" }], - markdown_text: "second\n\n", + // "first\n" arrived after the task chunk, so it sat in the + // adapter's chunk-mode buffer; it rides the rotation stop ahead + // of the delta's head instead of going out as its own append. + markdown_text: "first\nsecond\n\n", token: TOKEN, }); // The streamer kept the text from the rejected call, so the retry @@ -14146,6 +14149,248 @@ describe("native stream rotation", () => { }); }); +describe("native streaming after structured chunks", () => { + const THREAD = "slack:D123:1234567890.000000"; + const TOKEN = "xoxb-test-token"; + + function setup(config: Record = {}, segmentCount = 1) { + const adapter = createSlackAdapter({ + botToken: TOKEN, + signingSecret: "test-signing-secret", + logger: mockLogger, + ...config, + }); + const segments = Array.from({ length: segmentCount }, (_, index) => ({ + append: vi.fn().mockResolvedValue({ ok: true }), + stop: vi.fn().mockResolvedValue({ ok: true, ts: `1234567890.${index}` }), + ts: `1234567890.${index}`, + })); + const chatStream = vi + .fn() + .mockImplementation(() => segments[chatStream.mock.calls.length - 1]); + mockClientMethod(adapter, "chatStream", chatStream); + return { adapter, segments, chatStream }; + } + + /** Append calls that carry text through the `markdown_text` parameter. */ + function markdownParamCalls(append: ReturnType) { + return append.mock.calls.filter(([args]) => "markdown_text" in args); + } + + /** Text carried by append calls made in chunk form. */ + function chunkFormText(append: ReturnType): string { + return append.mock.calls + .flatMap(([args]) => args.chunks ?? []) + .filter((chunk) => chunk.type === "markdown_text") + .map((chunk) => chunk.text) + .join(""); + } + + it("sends text deltas as markdown_text chunks once a segment has plan/task chunks", async () => { + const { adapter, segments } = setup(); + async function* stream() { + yield { title: "Plan", type: "plan_update" as const }; + yield { + id: "t1", + status: "complete" as const, + title: "Task one", + type: "task_update" as const, + }; + yield "The answer is 42.\n\n"; + yield "More detail follows.\n\n"; + } + + await adapter.stream(THREAD, stream()); + + const append = segments[0].append; + // The plan and task chunks go out as chunks. + expect(append).toHaveBeenNthCalledWith(1, { + chunks: [{ title: "Plan", type: "plan_update" }], + token: TOKEN, + }); + expect(append).toHaveBeenNthCalledWith(2, { + chunks: [ + { + id: "t1", + status: "complete", + title: "Task one", + type: "task_update", + }, + ], + token: TOKEN, + }); + // Once the segment has chunks, Slack rejects appends that use the + // markdown_text parameter with streaming_mode_mismatch, so every text + // delta after the chunks must travel as a markdown_text chunk instead. + expect(markdownParamCalls(append)).toHaveLength(0); + expect(chunkFormText(append)).toContain("The answer is 42."); + expect(chunkFormText(append)).toContain("More detail follows."); + }); + + it("keeps the markdown_text parameter for text sent before any structured chunk", async () => { + const { adapter, segments } = setup(); + async function* stream() { + yield "Intro first.\n\n"; + yield { title: "Plan", type: "plan_update" as const }; + yield "Body after plan.\n\n"; + } + + await adapter.stream(THREAD, stream()); + + const append = segments[0].append; + expect(append).toHaveBeenNthCalledWith(1, { + markdown_text: "Intro first.\n\n", + token: TOKEN, + }); + // Only the pre-chunk delta may use the parameter; the post-chunk one + // must switch to chunk form. + expect(markdownParamCalls(append)).toHaveLength(1); + expect(chunkFormText(append)).toContain("Body after plan."); + }); + + it("keeps chunk form for text after a rotation replays plan/task chunks", async () => { + let now = 0; + const dateNow = vi.spyOn(Date, "now").mockImplementation(() => now); + try { + const { adapter, segments, chatStream } = setup( + { streamSegmentMaxAgeMs: 100 }, + 2 + ); + async function* stream() { + yield { title: "Plan", type: "plan_update" as const }; + yield { + id: "t1", + status: "in_progress" as const, + title: "Task one", + type: "task_update" as const, + }; + yield "First part.\n\n"; + now = 100 + 30_001; + yield "Second part.\n\nThird part.\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(2); + // The new segment opens by replaying the plan and the open task. + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + chunks: [ + { title: "Plan", type: "plan_update" }, + { + id: "t1", + status: "in_progress", + title: "Task one", + type: "task_update", + }, + ], + token: TOKEN, + }); + // The replay puts the new segment in chunk mode from the start: the + // text rotated into it ("Second part." travels with the old segment's + // stop call) must use chunk form too, never the markdown_text + // parameter. + expect(markdownParamCalls(segments[1].append)).toHaveLength(0); + expect(chunkFormText(segments[1].append)).toContain("Third part."); + } finally { + dateNow.mockRestore(); + } + }); + + it("buffers token-sized deltas in chunk mode instead of one append per delta", async () => { + const { adapter, segments } = setup(); + async function* stream() { + yield { title: "Plan", type: "plan_update" as const }; + yield "```ts\n"; + yield "const answer = "; + yield "40 + "; + yield "2;\n"; + yield "```\n\n"; + } + + await adapter.stream(THREAD, stream()); + + const append = segments[0].append; + expect(markdownParamCalls(append)).toHaveLength(0); + // One append for the plan chunk and a single one for the buffered + // text, flushed at stream end — not one awaited append per delta, + // which is what sending every delta as its own chunk would cost + // (the streamer flushes on any append that carries chunks). + expect(append).toHaveBeenCalledTimes(2); + expect(chunkFormText(append)).toContain("const answer = 40 + 2;"); + }); + + it("flushes buffered chunk-mode text once it reaches the flush size", async () => { + const { adapter, segments } = setup(); + const line = "x".repeat(100); + async function* stream() { + yield { title: "Plan", type: "plan_update" as const }; + yield `${line}\n`; + yield `${line}\n`; + yield `${line}\n`; + yield "tail\n"; + } + + await adapter.stream(THREAD, stream()); + + const append = segments[0].append; + // Plan chunk, one mid-stream flush once the buffer passes the flush + // size (303 chars over three deltas), one stream-end flush for the + // tail. + expect(append).toHaveBeenCalledTimes(3); + const textChunks = append.mock.calls + .flatMap(([args]) => args.chunks ?? []) + .filter((chunk) => chunk.type === "markdown_text"); + expect(textChunks.map((chunk) => chunk.text)).toEqual([ + `${line}\n${line}\n${line}\n`, + "tail\n", + ]); + }); + + it("flushes buffered text before a structured chunk so order is preserved", async () => { + const { adapter, segments } = setup(); + async function* stream() { + yield { title: "Plan", type: "plan_update" as const }; + yield "before task\n"; + yield { + id: "t1", + status: "complete" as const, + title: "Task one", + type: "task_update" as const, + }; + yield "after task\n"; + } + + await adapter.stream(THREAD, stream()); + + const append = segments[0].append; + expect(markdownParamCalls(append)).toHaveLength(0); + expect(append).toHaveBeenCalledTimes(4); + expect(append).toHaveBeenNthCalledWith(1, { + chunks: [{ title: "Plan", type: "plan_update" }], + token: TOKEN, + }); + expect(append).toHaveBeenNthCalledWith(2, { + chunks: [{ text: "before task\n", type: "markdown_text" }], + token: TOKEN, + }); + expect(append).toHaveBeenNthCalledWith(3, { + chunks: [ + { + id: "t1", + status: "complete", + title: "Task one", + type: "task_update", + }, + ], + token: TOKEN, + }); + expect(append).toHaveBeenNthCalledWith(4, { + chunks: [{ text: "after task\n", type: "markdown_text" }], + token: TOKEN, + }); + }); +}); + describe("native streaming fallback", () => { function createAdapter(config?: Record) { const adapter = createSlackAdapter({ diff --git a/packages/adapter-slack/src/index.ts b/packages/adapter-slack/src/index.ts index 2015188b..a988fca3 100644 --- a/packages/adapter-slack/src/index.ts +++ b/packages/adapter-slack/src/index.ts @@ -297,6 +297,14 @@ const DEFAULT_STREAM_SEGMENT_MAX_AGE_MS = 240_000; * five-minute stream expiry. */ const STREAM_SEGMENT_ROTATION_GRACE_MS = 30_000; +/** + * In chunk mode (a segment that has appended structured chunks), text is + * buffered adapter-side up to this many characters before going out as a + * single markdown_text chunk. Mirrors the ChatStreamer default + * buffer_size that paces markdown_text appends, so both modes deliver + * text at the same cadence instead of one awaited API call per delta. + */ +const CHUNK_MODE_TEXT_FLUSH_SIZE = 256; /** Slack's error for appending to or stopping a stream it already expired. */ const STREAM_EXPIRED_ERROR = "message_not_in_streaming_state"; /** @@ -5761,12 +5769,33 @@ export class SlackAdapter implements Adapter { * text if that text continues the table cut by the previous segment. */ tableHeader: string; + /** + * Whether this segment has appended structured chunks. The first + * chunks append puts the Slack stream in chunk mode, and Slack then + * rejects text sent through the `markdown_text` parameter with + * `streaming_mode_mismatch`, so once this is set, text deltas must + * travel as `markdown_text` chunks instead. + */ + appendedChunks: boolean; + /** + * Text held back in chunk mode. The streamer flushes on ANY append + * that carries chunks, so sending each delta as its own chunk would + * cost one awaited API call per token-sized delta; chunk-mode text + * accumulates here instead and goes out as a single markdown_text + * chunk once it reaches CHUNK_MODE_TEXT_FLUSH_SIZE, or at the next + * ordering boundary (a structured chunk, a rotation, or the final + * stop). Holds the wire form (segment prefix already applied), like + * the streamer's own markdown_text buffer. + */ + pendingChunkText: string; } = { streamer: createStreamer(), startedAt: undefined, flushNextAppend: false, reopenFence: "", tableHeader: "", + appendedChunks: false, + pendingChunkText: "", }; let rotations = 0; /** Message IDs of the segments finalized so far, in order. */ @@ -5989,6 +6018,8 @@ export class SlackAdapter implements Adapter { segment.streamer = createStreamer(); segment.startedAt = undefined; segment.flushNextAppend = true; + segment.appendedChunks = false; + segment.pendingChunkText = ""; const fence = openFenceIn(sent); segment.reopenFence = fence ? `${fence.opening}\n` : ""; segment.tableHeader = fence ? "" : tableContinuation(sent); @@ -6004,6 +6035,7 @@ export class SlackAdapter implements Adapter { token, }); markSegmentStarted(); + segment.appendedChunks = true; } catch (error) { disableStructuredChunks("replay", error); } @@ -6054,9 +6086,16 @@ export class SlackAdapter implements Adapter { ? `${sent.endsWith("\n") ? "" : "\n"}${fence.marker}` : ""; // `head` may be this segment's first text, so it takes any prefix a - // previous rotation queued for it. + // previous rotation queued for it. Chunk-mode text still buffered in + // the adapter rides the same stop call ahead of it — the streamer's + // stop converts its markdown_text argument to a chunk, and the + // expired-stop recovery below recomputes the tail from lastFlushed, + // which still covers the buffered text in resolved coordinates. const finalText = - (head.length > 0 ? withSegmentPrefix(head) : "") + closer; + segment.pendingChunkText + + (head.length > 0 ? withSegmentPrefix(head) : "") + + closer; + segment.pendingChunkText = ""; const ageMs = segmentAgeMs(); const closingTaskChunks = structuredChunksSupported && openTasks.size > 0 @@ -6121,6 +6160,27 @@ export class SlackAdapter implements Adapter { return tail; }; + /** + * Deliver chunk-mode text buffered in the adapter as a single + * markdown_text chunk. Returns the append response, or null when + * nothing was buffered. On failure the text stays buffered, so a + * later flush can retry it. + */ + const flushPendingChunkText = async () => { + if (segment.pendingChunkText.length === 0) { + return null; + } + const text = segment.pendingChunkText; + const response = await segment.streamer.append({ + chunks: [ + { text, type: "markdown_text" }, + ] as ChatAppendStreamArguments["chunks"], + token, + }); + segment.pendingChunkText = ""; + return response; + }; + /** * Send a resolved-text delta to the native stream, rotating the segment * first when it is due. Rotation carries the leading part of the delta @@ -6137,12 +6197,33 @@ export class SlackAdapter implements Adapter { // append() buffers small deltas in memory and returns null until it // actually calls the API — only a non-null response proves content // is rendering natively. An empty chunks array makes it flush right - // away, so a segment opened by rotation is visible immediately. - const response = await segment.streamer.append({ - markdown_text: withSegmentPrefix(text), - token, - ...(segment.flushNextAppend ? { chunks: [] } : {}), - }); + // away, so a segment opened by rotation is visible immediately. A + // segment that has appended structured chunks is in chunk mode: + // Slack rejects the markdown_text parameter there + // (streaming_mode_mismatch), so the delta travels as a + // markdown_text chunk instead. But the streamer flushes on ANY + // append carrying chunks, so chunk-mode deltas are buffered in the + // adapter first (see pendingChunkText) and sent as one chunk per + // flush — the cadence the streamer's own buffer gives the + // markdown_text path — not one awaited API call per delta. + const prefixed = withSegmentPrefix(text); + let response: Awaited> = + null; + if (segment.appendedChunks) { + segment.pendingChunkText += prefixed; + if ( + segment.flushNextAppend || + segment.pendingChunkText.length >= CHUNK_MODE_TEXT_FLUSH_SIZE + ) { + response = await flushPendingChunkText(); + } + } else { + response = await segment.streamer.append({ + markdown_text: prefixed, + token, + ...(segment.flushNextAppend ? { chunks: [] } : {}), + }); + } if (response) { markSegmentStarted(); lastFlushed = resolvedCommitted; @@ -6204,11 +6285,21 @@ export class SlackAdapter implements Adapter { await sendDelta("", true); try { + // A structured chunk must not overtake chunk-mode text still + // buffered in the adapter, so flush that first, in order. A + // failure lands in the same catch as a failed chunk append, + // mirroring how this append flushes the streamer's own text + // buffer in markdown mode. + if (await flushPendingChunkText()) { + markSegmentStarted(); + lastFlushed = lastAppended; + } await segment.streamer.append({ chunks: [chunk] as ChatAppendStreamArguments["chunks"], token, }); markSegmentStarted(); + segment.appendedChunks = true; // Sending chunks flushes the streamer's text buffer as well. lastFlushed = lastAppended; rememberStructuredChunk(chunk); @@ -6272,6 +6363,13 @@ export class SlackAdapter implements Adapter { }; let result: Awaited>; try { + // Chunk mode may still hold text buffered in the adapter that the + // streamer never saw; deliver it before stop finalizes the message. + // Inside the try so an expired segment lands in the recovery below, + // whose unconfirmed tail still covers the buffered text. + if (await flushPendingChunkText()) { + lastFlushed = lastAppended; + } result = await segment.streamer.stop(stopArgs); } catch (error) { if (!fallback.nativeRendered) {