From d177cf68893d81e916122818709cbf2d54026399 Mon Sep 17 00:00:00 2001 From: dancer Date: Tue, 9 Jun 2026 15:43:42 +0100 Subject: [PATCH 1/4] fix(slack): continue oversized streams --- packages/adapter-slack/src/index.test.ts | 328 ++++++++++++++++++++++ packages/adapter-slack/src/index.ts | 239 +++++++++++++--- packages/adapter-slack/src/stream.test.ts | 113 ++++++++ packages/adapter-slack/src/stream.ts | 157 +++++++++++ 4 files changed, 802 insertions(+), 35 deletions(-) create mode 100644 packages/adapter-slack/src/stream.test.ts create mode 100644 packages/adapter-slack/src/stream.ts diff --git a/packages/adapter-slack/src/index.test.ts b/packages/adapter-slack/src/index.test.ts index 556036da..a600b712 100644 --- a/packages/adapter-slack/src/index.test.ts +++ b/packages/adapter-slack/src/index.test.ts @@ -10,6 +10,7 @@ import type { ChatInstance, Logger, StateAdapter, + StreamChunk, } from "chat"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { SlackInstallation } from "./index"; @@ -7768,6 +7769,333 @@ describe("stream with empty threadTs", () => { expect.objectContaining({ token: "xoxb-test-token" }) ); }); + + it("rotates oversized streams without corrupting unicode", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const segments: Array<{ + markdown: string; + stop: ReturnType; + }> = []; + const chatStream = vi.fn().mockImplementation(() => { + const index = segments.length; + const segment = { + markdown: "", + stop: vi.fn().mockResolvedValue({ + ok: true, + ts: `1234567890.${String(index).padStart(6, "0")}`, + }), + }; + segments.push(segment); + return { + append: vi.fn().mockImplementation(async (payload) => { + segment.markdown += payload.markdown_text ?? ""; + return { ok: true }; + }), + stop: segment.stop, + }; + }); + mockClientMethod(adapter, "chatStream", chatStream); + + const text = `${"a".repeat(11_999)}๐Ÿ˜€${"b".repeat(12_001)}`; + async function* longStream() { + yield "x".repeat(255); + yield { text, type: "markdown_text" as const }; + } + + const result = await adapter.stream( + "slack:C123:1234567890.000000", + longStream(), + { + recipientUserId: "U123", + recipientTeamId: "T123", + stopBlocks: [{ type: "divider" }], + } + ); + + expect(chatStream).toHaveBeenCalledWith( + expect.objectContaining({ buffer_size: 256 }) + ); + expect(segments.length).toBeGreaterThan(1); + expect(segments.map((segment) => segment.markdown).join("")).toBe( + `${"x".repeat(255)}${text}` + ); + for (const segment of segments) { + expect(segment.markdown.length).toBeLessThanOrEqual(11_500); + const firstCode = segment.markdown.charCodeAt(0); + const lastCode = segment.markdown.charCodeAt(segment.markdown.length - 1); + expect(firstCode < 0xdc00 || firstCode > 0xdfff).toBe(true); + expect(lastCode < 0xd800 || lastCode > 0xdbff).toBe(true); + } + for (const segment of segments.slice(0, -1)) { + expect(segment.stop).toHaveBeenCalledWith({ + token: "xoxb-test-token", + }); + } + expect(segments.at(-1)?.stop).toHaveBeenCalledWith({ + blocks: [{ type: "divider" }], + token: "xoxb-test-token", + }); + expect(result.id).toBe( + `1234567890.${String(segments.length - 1).padStart(6, "0")}` + ); + }); + + it("balances fenced code across continuation streams", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const segments: Array<{ + markdown: string; + stop: ReturnType; + }> = []; + mockClientMethod( + adapter, + "chatStream", + vi.fn().mockImplementation(() => { + const segment = { + markdown: "", + stop: vi.fn().mockResolvedValue({ + ok: true, + ts: `1234567890.${String(segments.length).padStart(6, "0")}`, + }), + }; + segments.push(segment); + return { + append: vi.fn().mockImplementation(async (payload) => { + segment.markdown += payload.markdown_text ?? ""; + return { ok: true }; + }), + stop: segment.stop, + }; + }) + ); + + async function* codeStream() { + yield "```typescript\n"; + for (let index = 0; index < 900; index++) { + yield `const value${index} = true;\n`; + } + yield "```\n"; + } + + await adapter.stream("slack:C123:1234567890.000000", codeStream(), { + recipientUserId: "U123", + recipientTeamId: "T123", + }); + + expect(segments.length).toBeGreaterThan(1); + for (const segment of segments) { + const fences = segment.markdown + .split("\n") + .filter((line) => line.trimStart().startsWith("```")); + expect(fences.length % 2).toBe(0); + expect(segment.markdown.length).toBeLessThanOrEqual(11_500); + } + expect(segments[0]?.markdown.endsWith("\n```")).toBe(true); + expect(segments[1]?.markdown.startsWith("```typescript\n")).toBe(true); + for (const segment of segments) { + expect(segment.markdown).not.toContain("\nc\n```"); + expect(segment.markdown).not.toContain("```typescript\nonst "); + } + }); + + it("splits oversized task content into continuation cards", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const chunks: StreamChunk[] = []; + const append = vi.fn().mockImplementation(async (payload) => { + chunks.push(...(payload.chunks ?? [])); + return { ok: true }; + }); + mockClientMethod( + adapter, + "chatStream", + vi.fn().mockReturnValue({ + append, + stop: vi.fn().mockResolvedValue({ + ok: true, + ts: "1234567890.111111", + }), + }) + ); + + const details = "searched the workspace for matching records ".repeat(20); + const output = "found a relevant result with supporting context ".repeat( + 18 + ); + async function* taskStream() { + yield { + id: "research", + status: "in_progress" as const, + title: "Researching workspace context", + type: "task_update" as const, + }; + yield { + details, + id: "research", + output, + status: "complete" as const, + title: "Research complete", + type: "task_update" as const, + }; + yield { + title: "Plan ".repeat(80), + type: "plan_update" as const, + }; + } + + await adapter.stream("slack:C123:1234567890.000000", taskStream(), { + recipientUserId: "U123", + recipientTeamId: "T123", + }); + + const completed = chunks.filter( + (chunk): chunk is Extract => + chunk.type === "task_update" && chunk.status === "complete" + ); + expect(completed.length).toBeGreaterThan(1); + expect(completed.map((chunk) => chunk.details ?? "").join("")).toBe( + details + ); + expect(completed.map((chunk) => chunk.output ?? "").join("")).toBe(output); + expect(new Set(completed.map((chunk) => chunk.id)).size).toBe( + completed.length + ); + for (const chunk of completed) { + expect(chunk.title.length).toBeLessThanOrEqual(256); + expect(chunk.details?.length ?? 0).toBeLessThanOrEqual(256); + expect(chunk.output?.length ?? 0).toBeLessThanOrEqual(256); + } + + const plan = chunks.find( + (chunk): chunk is Extract => + chunk.type === "plan_update" + ); + expect(plan?.title.length).toBe(256); + expect(plan?.title.endsWith("...")).toBe(true); + }); + + it("updates tasks on their original stream after text rotates", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const segments: Array<{ + chunks: StreamChunk[]; + markdown: string; + stop: ReturnType; + }> = []; + mockClientMethod( + adapter, + "chatStream", + vi.fn().mockImplementation(() => { + const index = segments.length; + const segment = { + chunks: [] as StreamChunk[], + markdown: "", + stop: vi.fn().mockResolvedValue({ + ok: true, + ts: `1234567890.${String(index).padStart(6, "0")}`, + }), + }; + segments.push(segment); + return { + append: vi.fn().mockImplementation(async (payload) => { + segment.chunks.push(...(payload.chunks ?? [])); + segment.markdown += payload.markdown_text ?? ""; + return { ok: true }; + }), + stop: segment.stop, + }; + }) + ); + + async function* mixedStream() { + yield { + id: "research", + status: "in_progress" as const, + title: "Researching", + type: "task_update" as const, + }; + yield "text ".repeat(5000); + yield { + id: "research", + status: "complete" as const, + title: "Research complete", + type: "task_update" as const, + }; + } + + await adapter.stream("slack:C123:1234567890.000000", mixedStream(), { + recipientUserId: "U123", + recipientTeamId: "T123", + }); + + expect(segments.length).toBeGreaterThan(1); + const taskSegments = segments.filter((segment) => + segment.chunks.some((chunk) => chunk.type === "task_update") + ); + expect(taskSegments).toHaveLength(1); + expect( + taskSegments[0]?.chunks + .filter((chunk) => chunk.type === "task_update") + .map((chunk) => chunk.status) + ).toEqual(["in_progress", "complete"]); + for (const segment of segments) { + expect(segment.stop).toHaveBeenCalledOnce(); + } + }); + + it("stops a partial stream when its source fails", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const stop = vi.fn().mockResolvedValue({ + ok: true, + ts: "1234567890.111111", + }); + mockClientMethod( + adapter, + "chatStream", + vi.fn().mockReturnValue({ + append: vi.fn().mockResolvedValue({ ok: true }), + stop, + }) + ); + + const sourceError = new Error("stream failed"); + async function* failingStream() { + yield { + id: "research", + status: "in_progress" as const, + title: "Researching", + type: "task_update" as const, + }; + throw sourceError; + } + + await expect( + adapter.stream("slack:C123:1234567890.000000", failingStream(), { + recipientUserId: "U123", + recipientTeamId: "T123", + }) + ).rejects.toBe(sourceError); + expect(stop).toHaveBeenCalledWith({ + token: "xoxb-test-token", + }); + }); }); describe("scheduleMessage with empty threadTs", () => { diff --git a/packages/adapter-slack/src/index.ts b/packages/adapter-slack/src/index.ts index bbbee206..ed5f4f99 100644 --- a/packages/adapter-slack/src/index.ts +++ b/packages/adapter-slack/src/index.ts @@ -10,7 +10,11 @@ import { ValidationError, } from "@chat-adapter/shared"; import { SocketModeClient } from "@slack/socket-mode"; -import { type ChatStopStreamArguments, WebClient } from "@slack/web-api"; +import { + type ChatStopStreamArguments, + type ChatStreamer, + WebClient, +} from "@slack/web-api"; import type { ActionEvent, Adapter, @@ -73,6 +77,16 @@ import { type SlackModalResponse, selectOptionToSlackOption, } from "./modals"; +import { + Fence, + STREAM_BUFFER_SIZE, + STREAM_CHUNK_LIMIT, + STREAM_FENCE_RESERVE, + STREAM_SEGMENT_LIMIT, + splitTask, + splitText, + truncateText, +} from "./stream"; import { verifySlackRequest } from "./webhook/index"; const SLACK_USER_ID_PATTERN = /^[A-Z0-9_]+$/; @@ -3981,26 +3995,120 @@ export class SlackAdapter implements Adapter { this.logger.debug("Slack: starting stream", { channel, threadTs }); const token = await this.getToken(); - const streamer = this._client.chatStream({ - channel, - thread_ts: threadTs, - recipient_user_id: options.recipientUserId, - recipient_team_id: options.recipientTeamId, - ...(options.taskDisplayMode && { - task_display_mode: options.taskDisplayMode, + interface Segment { + hasContent: boolean; + length: number; + stopped: boolean; + streamer: ChatStreamer; + } + type StopResult = Awaited>; + + const createSegment = (): Segment => ({ + hasContent: false, + length: 0, + stopped: false, + streamer: this._client.chatStream({ + channel, + thread_ts: threadTs, + recipient_user_id: options.recipientUserId, + recipient_team_id: options.recipientTeamId, + ...(options.taskDisplayMode && { + task_display_mode: options.taskDisplayMode, + }), + buffer_size: STREAM_BUFFER_SIZE, }), }); + let current = createSegment(); + let structured: Segment | undefined; + let lastResult: StopResult | undefined; let lastAppended = ""; + const taskParts = new Map(); + const fence = new Fence(); const renderer = new StreamingMarkdownRenderer({ wrapTablesForAppend: false, }); + const stopSegment = async ( + segment: Segment, + blocks?: ChatStopStreamArguments["blocks"], + force = false + ): Promise => { + if (segment.stopped || !(segment.hasContent || force)) { + return undefined; + } + const result = await segment.streamer.stop({ + token, + ...(blocks ? { blocks } : {}), + }); + segment.stopped = true; + return result; + }; + + const rotateSegment = async ( + closing?: string, + opening?: string + ): Promise => { + if (closing) { + await current.streamer.append({ markdown_text: closing, token }); + current.hasContent = true; + current.length += closing.length; + } + if (current !== structured) { + await stopSegment(current); + } + current = createSegment(); + if (opening) { + await current.streamer.append({ markdown_text: opening, token }); + current.hasContent = true; + current.length += opening.length; + } + }; + const flushMarkdownDelta = async (delta: string): Promise => { if (delta.length === 0) { return; } - await streamer.append({ markdown_text: delta, token }); + + let remaining = delta; + while (remaining.length > 0) { + if (current.length === STREAM_SEGMENT_LIMIT) { + await rotateSegment(); + } + + const available = STREAM_SEGMENT_LIMIT - current.length; + const first = remaining.codePointAt(0); + const width = first !== undefined && first > 0xffff ? 2 : 1; + if (available < width) { + await rotateSegment(fence.closing, fence.opening); + continue; + } + if ( + remaining.length > available && + available <= STREAM_FENCE_RESERVE + width + ) { + await rotateSegment(fence.closing, fence.opening); + continue; + } + const limit = + remaining.length > available + ? Math.max(width, available - STREAM_FENCE_RESERVE) + : available; + const [text] = splitText(remaining, limit); + if (!text) { + break; + } + + await current.streamer.append({ markdown_text: text, token }); + current.hasContent = true; + current.length += text.length; + fence.push(text); + remaining = remaining.slice(text.length); + + if (remaining.length > 0) { + await rotateSegment(fence.closing, fence.opening); + } + } }; /** @@ -4014,7 +4122,9 @@ export class SlackAdapter implements Adapter { * Text streaming continues unaffected. */ let structuredChunksSupported = true; - const sendStructuredChunk = async (chunk: StreamChunk): Promise => { + const sendStructuredChunk = async ( + chunk: Exclude + ): Promise => { if (!structuredChunksSupported) { return; } @@ -4026,8 +4136,29 @@ export class SlackAdapter implements Adapter { lastAppended = committable; try { - // biome-ignore lint/suspicious/noExplicitAny: chunks not in ChatAppendStreamArguments for older @slack/web-api - await streamer.append({ chunks: [chunk], token } as any); + structured ??= current; + const chunks = + chunk.type === "task_update" + ? splitTask(chunk, taskParts.get(chunk.id) ?? 0) + : [ + { + ...chunk, + title: truncateText(chunk.title, STREAM_CHUNK_LIMIT), + }, + ]; + + if (chunk.type === "task_update") { + taskParts.set(chunk.id, chunks.length); + } + + for (const normalized of chunks) { + await structured.streamer.append({ + chunks: [normalized], + token, + // biome-ignore lint/suspicious/noExplicitAny: chunks not in ChatAppendStreamArguments for older @slack/web-api + } as any); + structured.hasContent = true; + } } catch (error) { // Structured chunks may fail if the app doesn't have the required // Assistant scopes/features. Disable for the rest of this stream @@ -4050,39 +4181,77 @@ export class SlackAdapter implements Adapter { lastAppended = committable; }; - for await (const chunk of textStream) { - if (typeof chunk === "string") { - await pushTextAndFlush(chunk); - } else if (chunk.type === "markdown_text") { - await pushTextAndFlush(chunk.text); - } else { - // Structured chunk (task_update, plan_update) โ€” send directly to Slack - await sendStructuredChunk(chunk); + try { + for await (const chunk of textStream) { + if (typeof chunk === "string") { + await pushTextAndFlush(chunk); + } else if (chunk.type === "markdown_text") { + await pushTextAndFlush(chunk.text); + } else { + // Structured chunk (task_update, plan_update) โ€” send directly to Slack + await sendStructuredChunk(chunk); + } } - } - // Flush any remaining buffered content (e.g. held table rows at end of stream). - renderer.finish(); - const finalCommittable = renderer.getCommittableText(); - const finalDelta = finalCommittable.slice(lastAppended.length); - await flushMarkdownDelta(finalDelta); + // Flush any remaining buffered content (e.g. held table rows at end of stream). + renderer.finish(); + const finalCommittable = renderer.getCommittableText(); + const finalDelta = finalCommittable.slice(lastAppended.length); + await flushMarkdownDelta(finalDelta); + fence.finish(); + if (fence.closing) { + await current.streamer.append({ + markdown_text: fence.closing, + token, + }); + current.hasContent = true; + current.length += fence.closing.length; + } - const result = await streamer.stop({ - token, - ...(options?.stopBlocks - ? { - blocks: options.stopBlocks as ChatStopStreamArguments["blocks"], + if (structured && structured !== current) { + await stopSegment(structured); + } + lastResult = await stopSegment( + current, + options?.stopBlocks as ChatStopStreamArguments["blocks"], + true + ); + } catch (error) { + const segments = new Set([structured, current]); + fence.finish(); + for (const segment of segments) { + if (!segment?.hasContent) { + continue; + } + try { + if (segment === current && fence.closing) { + await segment.streamer.append({ + markdown_text: fence.closing, + token, + }); + segment.length += fence.closing.length; } - : {}), - }); - const messageTs = (result.message?.ts ?? result.ts) as string; + await stopSegment(segment); + } catch (stopError) { + this.logger.warn("Slack: failed to stop partial stream", { + error: stopError, + }); + } + } + throw error; + } + + if (!lastResult) { + throw new NetworkError("slack", "Slack stream returned no result"); + } + const messageTs = (lastResult.message?.ts ?? lastResult.ts) as string; this.logger.debug("Slack: stream complete", { messageId: messageTs }); return { id: messageTs, threadId, - raw: result, + raw: lastResult, }; } diff --git a/packages/adapter-slack/src/stream.test.ts b/packages/adapter-slack/src/stream.test.ts new file mode 100644 index 00000000..cc8a4c37 --- /dev/null +++ b/packages/adapter-slack/src/stream.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it } from "vitest"; +import { + Fence, + STREAM_CHUNK_LIMIT, + splitTask, + splitText, + truncateText, +} from "./stream"; + +describe("Slack stream limits", () => { + it("splits text without changing content or breaking unicode", () => { + const text = `${"word ".repeat(80)}๐Ÿ˜€${"tail ".repeat(80)}`; + const chunks = splitText(text, 256); + + expect(chunks.join("")).toBe(text); + expect(chunks.length).toBeGreaterThan(1); + for (const chunk of chunks) { + expect(chunk.length).toBeLessThanOrEqual(256); + const first = chunk.charCodeAt(0); + const last = chunk.charCodeAt(chunk.length - 1); + expect(first < 0xdc00 || first > 0xdfff).toBe(true); + expect(last < 0xd800 || last > 0xdbff).toBe(true); + } + }); + + it("keeps every prior task part in later status updates", () => { + const details = "searched matching workspace context ".repeat(30); + const initial = splitTask( + { + details, + id: "research", + status: "in_progress", + title: "Researching", + type: "task_update", + }, + 0 + ); + const complete = splitTask( + { + id: "research", + status: "complete", + title: "Research complete", + type: "task_update", + }, + initial.length + ); + + expect(initial.length).toBeGreaterThan(1); + expect(complete).toHaveLength(initial.length); + expect(complete.every((chunk) => chunk.status === "complete")).toBe(true); + expect(complete.map((chunk) => chunk.id)).toEqual( + initial.map((chunk) => chunk.id) + ); + }); + + it("keeps task sources on the first continuation", () => { + const sources = [ + { + text: "Slack", + type: "url" as const, + url: "https://docs.slack.dev", + }, + ]; + const chunks = splitTask( + { + details: "workspace context ".repeat(50), + id: "research", + sources, + status: "complete", + title: "Research complete", + type: "task_update", + } as Parameters[0], + 0 + ); + + expect(chunks.length).toBeGreaterThan(1); + expect(chunks[0]).toMatchObject({ sources }); + expect(chunks.slice(1).every((chunk) => !("sources" in chunk))).toBe(true); + }); + + it("tracks fenced code across streamed chunks", () => { + const fence = new Fence(); + + fence.push("before\n```typescript\nconst value = 1;"); + expect(fence.closing).toBe("\n```"); + expect(fence.opening).toBe("```typescript\n"); + + fence.push("\n```\nafter"); + expect(fence.closing).toBeUndefined(); + expect(fence.opening).toBeUndefined(); + }); + + it("tracks a final fence without a trailing newline", () => { + const fence = new Fence(); + + fence.push("```typescript\nconst value = 1;\n```"); + expect(fence.closing).toBe("\n```"); + + fence.finish(); + expect(fence.closing).toBeUndefined(); + }); + + it("truncates titles without splitting unicode", () => { + const text = `${"a".repeat(STREAM_CHUNK_LIMIT - 4)}๐Ÿ˜€overflow`; + const truncated = truncateText(text, STREAM_CHUNK_LIMIT); + + expect(truncated.length).toBeLessThanOrEqual(STREAM_CHUNK_LIMIT); + expect(truncated.endsWith("...")).toBe(true); + expect( + truncated.charCodeAt(truncated.length - 4) + ).not.toBeGreaterThanOrEqual(0xd800); + }); +}); diff --git a/packages/adapter-slack/src/stream.ts b/packages/adapter-slack/src/stream.ts new file mode 100644 index 00000000..82e02d48 --- /dev/null +++ b/packages/adapter-slack/src/stream.ts @@ -0,0 +1,157 @@ +import type { StreamChunk } from "chat"; + +export const STREAM_BUFFER_SIZE = 256; +export const STREAM_SEGMENT_LIMIT = 11_500; +export const STREAM_CHUNK_LIMIT = 256; +export const STREAM_FENCE_RESERVE = 64; + +type TaskChunk = Extract; +const FENCE_PATTERN = /^(`{3,}|~{3,})/; + +export class Fence { + private buffer = ""; + private value?: { + marker: string; + opening: string; + }; + + get closing(): string | undefined { + return this.value ? `\n${this.value.marker}` : undefined; + } + + get opening(): string | undefined { + return this.value ? `${this.value.opening}\n` : undefined; + } + + finish(): void { + if (this.buffer) { + this.track(this.buffer); + this.buffer = ""; + } + } + + push(text: string): void { + const lines = `${this.buffer}${text}`.split("\n"); + this.buffer = lines.pop() ?? ""; + + for (const line of lines) { + this.track(line); + } + } + + private track(line: string): void { + const trimmed = line.trimStart(); + const match = FENCE_PATTERN.exec(trimmed); + if (!match) { + return; + } + + const marker = match[1]; + if (!this.value) { + this.value = { + marker, + opening: trimmed, + }; + } else if ( + marker[0] === this.value.marker[0] && + marker.length >= this.value.marker.length + ) { + this.value = undefined; + } + } +} + +export function splitText(text: string, limit: number): string[] { + if (text.length <= limit) { + return [text]; + } + + const chunks: string[] = []; + let offset = 0; + let remaining = Math.ceil(text.length / limit); + + while (offset < text.length) { + const left = text.length - offset; + if (left <= limit) { + chunks.push(text.slice(offset)); + break; + } + + const target = Math.min(limit, Math.ceil(left / Math.max(remaining, 1))); + const minimum = Math.max(1, Math.floor(target * 0.75)); + let end = offset + target; + const window = text.slice(offset, end); + const boundary = Math.max( + window.lastIndexOf("\n"), + window.lastIndexOf(" ") + ); + + if (boundary >= minimum) { + end = offset + boundary + 1; + } + + const lastCode = text.charCodeAt(end - 1); + if (lastCode >= 0xd800 && lastCode <= 0xdbff) { + end += end - offset === 1 ? 1 : -1; + } + + chunks.push(text.slice(offset, end)); + offset = end; + remaining = Math.max(1, remaining - 1); + } + + return chunks; +} + +export function truncateText(text: string, limit: number): string { + if (text.length <= limit) { + return text; + } + let end = limit - 3; + const lastCode = text.charCodeAt(end - 1); + if (lastCode >= 0xd800 && lastCode <= 0xdbff) { + end -= 1; + } + return `${text.slice(0, end)}...`; +} + +function taskId(id: string, index: number): string { + if (index === 0) { + return truncateText(id, STREAM_CHUNK_LIMIT); + } + const suffix = `:part:${index + 1}`; + return `${truncateText(id, STREAM_CHUNK_LIMIT - suffix.length)}${suffix}`; +} + +function taskTitle(title: string, index: number, total: number): string { + if (total === 1) { + return truncateText(title, STREAM_CHUNK_LIMIT); + } + const suffix = ` (${index + 1}/${total})`; + return `${truncateText(title, STREAM_CHUNK_LIMIT - suffix.length)}${suffix}`; +} + +export function splitTask(chunk: TaskChunk, previous: number): TaskChunk[] { + const sources = ( + chunk as TaskChunk & { + sources?: unknown; + } + ).sources; + const details = chunk.details + ? splitText(chunk.details, STREAM_CHUNK_LIMIT) + : []; + const output = chunk.output + ? splitText(chunk.output, STREAM_CHUNK_LIMIT) + : []; + const total = Math.max(previous, details.length, output.length, 1); + + return Array.from({ length: total }, (_, index) => ({ + ...(details[index] ? { details: details[index] } : {}), + id: taskId(chunk.id, index), + ...(output[index] ? { output: output[index] } : {}), + ...(index === 0 && sources !== undefined ? { sources } : {}), + status: chunk.status, + title: taskTitle(chunk.title, index, total), + type: "task_update", + })); +} From 08ecb959c36da8664d7ae33ed252ea44b2726250 Mon Sep 17 00:00:00 2001 From: dancer Date: Tue, 9 Jun 2026 15:44:24 +0100 Subject: [PATCH 2/4] docs(slack): document stream continuation --- .changeset/petite-wombats-hang.md | 5 +++++ apps/docs/content/docs/streaming.mdx | 4 ++++ 2 files changed, 9 insertions(+) create mode 100644 .changeset/petite-wombats-hang.md diff --git a/.changeset/petite-wombats-hang.md b/.changeset/petite-wombats-hang.md new file mode 100644 index 00000000..ffb50d97 --- /dev/null +++ b/.changeset/petite-wombats-hang.md @@ -0,0 +1,5 @@ +--- +"@chat-adapter/slack": patch +--- + +continue Slack streams across replies when text or task content exceeds platform limits diff --git a/apps/docs/content/docs/streaming.mdx b/apps/docs/content/docs/streaming.mdx index 1740fb11..6db16797 100644 --- a/apps/docs/content/docs/streaming.mdx +++ b/apps/docs/content/docs/streaming.mdx @@ -152,6 +152,10 @@ await thread.post(stream); | `task_update` | `id`, `title`, `status`, `details?`, `output?` | Tool/step progress updates (`pending`, `in_progress`, `complete`, `error`) with optional extra task context | | `plan_update` | `title` | Plan title updates on supported platforms | +Slack limits native streamed text fields to 12,000 characters and task or plan chunks to 256 characters. Chat SDK handles these limits automatically. Long text continues in ordered replies, long task details and output continue in additional task cards, and task or plan titles are shortened to fit. + +When Slack creates continuation replies, `thread.post()` returns the final reply. Stop blocks are attached to that final reply. + ### Streaming with options Wrap a stream in a `StreamingPlan` to pass platform-specific options through `thread.post()` without dropping down to `adapter.stream()` directly: From 38980d5f5d12db4cdb063f5576cb52eb54b9a860 Mon Sep 17 00:00:00 2001 From: dancer Date: Tue, 9 Jun 2026 15:55:01 +0100 Subject: [PATCH 3/4] fix(slack): preserve exact-boundary fences --- packages/adapter-slack/src/index.test.ts | 52 ++++++++++++++++++++++++ packages/adapter-slack/src/index.ts | 2 +- 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/packages/adapter-slack/src/index.test.ts b/packages/adapter-slack/src/index.test.ts index a600b712..2d68c74e 100644 --- a/packages/adapter-slack/src/index.test.ts +++ b/packages/adapter-slack/src/index.test.ts @@ -7905,6 +7905,58 @@ describe("stream with empty threadTs", () => { } }); + it("balances fenced code when a segment fills exactly", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const segments: Array<{ + markdown: string; + stop: ReturnType; + }> = []; + mockClientMethod( + adapter, + "chatStream", + vi.fn().mockImplementation(() => { + const segment = { + markdown: "", + stop: vi.fn().mockResolvedValue({ + ok: true, + ts: `1234567890.${String(segments.length).padStart(6, "0")}`, + }), + }; + segments.push(segment); + return { + append: vi.fn().mockImplementation(async (payload) => { + segment.markdown += payload.markdown_text ?? ""; + return { ok: true }; + }), + stop: segment.stop, + }; + }) + ); + + const opening = "```typescript\n"; + async function* codeStream() { + yield `${opening}${"a".repeat(11_500 - opening.length)}`; + yield "\nconst next = true;\n```\n"; + } + + await adapter.stream("slack:C123:1234567890.000000", codeStream(), { + recipientUserId: "U123", + recipientTeamId: "T123", + }); + + expect(segments).toHaveLength(2); + expect(segments[0]?.markdown.endsWith("\n```")).toBe(true); + expect(segments[1]?.markdown.startsWith(opening)).toBe(true); + expect(segments[1]?.markdown.endsWith("```\n")).toBe(true); + for (const segment of segments) { + expect(segment.markdown.length).toBeLessThanOrEqual(12_000); + } + }); + it("splits oversized task content into continuation cards", async () => { const adapter = createSlackAdapter({ botToken: "xoxb-test-token", diff --git a/packages/adapter-slack/src/index.ts b/packages/adapter-slack/src/index.ts index ed5f4f99..41a208e9 100644 --- a/packages/adapter-slack/src/index.ts +++ b/packages/adapter-slack/src/index.ts @@ -4073,7 +4073,7 @@ export class SlackAdapter implements Adapter { let remaining = delta; while (remaining.length > 0) { if (current.length === STREAM_SEGMENT_LIMIT) { - await rotateSegment(); + await rotateSegment(fence.closing, fence.opening); } const available = STREAM_SEGMENT_LIMIT - current.length; From 246e5ed9f98401f6ce76780dbbb4c440c9d6d4c0 Mon Sep 17 00:00:00 2001 From: dancer Date: Tue, 9 Jun 2026 23:31:16 +0100 Subject: [PATCH 4/4] fix(slack): continue task cards across replies --- apps/docs/content/docs/streaming.mdx | 2 +- packages/adapter-slack/src/index.test.ts | 94 ++++++++++++++++++++++++ packages/adapter-slack/src/index.ts | 93 ++++++++++++++++------- packages/adapter-slack/src/stream.ts | 1 + 4 files changed, 164 insertions(+), 26 deletions(-) diff --git a/apps/docs/content/docs/streaming.mdx b/apps/docs/content/docs/streaming.mdx index 6db16797..4e1062d0 100644 --- a/apps/docs/content/docs/streaming.mdx +++ b/apps/docs/content/docs/streaming.mdx @@ -152,7 +152,7 @@ await thread.post(stream); | `task_update` | `id`, `title`, `status`, `details?`, `output?` | Tool/step progress updates (`pending`, `in_progress`, `complete`, `error`) with optional extra task context | | `plan_update` | `title` | Plan title updates on supported platforms | -Slack limits native streamed text fields to 12,000 characters and task or plan chunks to 256 characters. Chat SDK handles these limits automatically. Long text continues in ordered replies, long task details and output continue in additional task cards, and task or plan titles are shortened to fit. +Slack limits native streamed text fields to 12,000 characters, task or plan chunks to 256 characters, and messages to 50 blocks. Chat SDK handles these limits automatically. Long text continues in ordered replies, long task details and output continue in additional task cards, and task or plan titles are shortened to fit. When a stream reaches 50 task cards, later tasks continue in another reply and updates remain attached to the reply containing the original task card. When Slack creates continuation replies, `thread.post()` returns the final reply. Stop blocks are attached to that final reply. diff --git a/packages/adapter-slack/src/index.test.ts b/packages/adapter-slack/src/index.test.ts index 2d68c74e..444cfcc6 100644 --- a/packages/adapter-slack/src/index.test.ts +++ b/packages/adapter-slack/src/index.test.ts @@ -8108,6 +8108,100 @@ describe("stream with empty threadTs", () => { } }); + it("continues task cards across stream messages after fifty cards", async () => { + const adapter = createSlackAdapter({ + botToken: "xoxb-test-token", + signingSecret: "test-signing-secret", + logger: mockLogger, + }); + const segments: Array<{ + chunks: StreamChunk[]; + stop: ReturnType; + }> = []; + mockClientMethod( + adapter, + "chatStream", + vi.fn().mockImplementation(() => { + const index = segments.length; + const segment = { + chunks: [] as StreamChunk[], + stop: vi.fn().mockResolvedValue({ + ok: true, + ts: `1234567890.${String(index).padStart(6, "0")}`, + }), + }; + segments.push(segment); + return { + append: vi.fn().mockImplementation(async (payload) => { + segment.chunks.push(...(payload.chunks ?? [])); + return { ok: true }; + }), + stop: segment.stop, + }; + }) + ); + + async function* taskStream() { + yield { + title: "Migration plan", + type: "plan_update" as const, + }; + for (let index = 0; index < 60; index++) { + yield { + id: `task-${index}`, + status: "in_progress" as const, + title: `Task ${index}`, + type: "task_update" as const, + }; + } + for (let index = 0; index < 60; index++) { + yield { + id: `task-${index}`, + status: "complete" as const, + title: `Task ${index}`, + type: "task_update" as const, + }; + } + yield { + title: "Migration complete", + type: "plan_update" as const, + }; + } + + const result = await adapter.stream( + "slack:C123:1234567890.000000", + taskStream(), + { + recipientUserId: "U123", + recipientTeamId: "T123", + } + ); + + expect(segments).toHaveLength(2); + const owners = new Map(); + for (const [index, segment] of segments.entries()) { + const tasks = segment.chunks.filter( + (chunk): chunk is Extract => + chunk.type === "task_update" + ); + expect(new Set(tasks.map((chunk) => chunk.id)).size).toBeLessThanOrEqual( + 50 + ); + for (const task of tasks) { + expect(owners.get(task.id) ?? index).toBe(index); + owners.set(task.id, index); + } + expect( + segment.chunks + .filter((chunk) => chunk.type === "plan_update") + .map((chunk) => chunk.title) + ).toEqual(["Migration plan", "Migration complete"]); + expect(segment.stop).toHaveBeenCalledOnce(); + } + expect(owners.size).toBe(60); + expect(result.id).toBe("1234567890.000001"); + }); + it("stops a partial stream when its source fails", async () => { const adapter = createSlackAdapter({ botToken: "xoxb-test-token", diff --git a/packages/adapter-slack/src/index.ts b/packages/adapter-slack/src/index.ts index 41a208e9..e07b3d10 100644 --- a/packages/adapter-slack/src/index.ts +++ b/packages/adapter-slack/src/index.ts @@ -83,6 +83,7 @@ import { STREAM_CHUNK_LIMIT, STREAM_FENCE_RESERVE, STREAM_SEGMENT_LIMIT, + STREAM_TASK_LIMIT, splitTask, splitText, truncateText, @@ -3996,6 +3997,7 @@ export class SlackAdapter implements Adapter { const token = await this.getToken(); interface Segment { + cards: number; hasContent: boolean; length: number; stopped: boolean; @@ -4004,6 +4006,7 @@ export class SlackAdapter implements Adapter { type StopResult = Awaited>; const createSegment = (): Segment => ({ + cards: 0, hasContent: false, length: 0, stopped: false, @@ -4020,10 +4023,12 @@ export class SlackAdapter implements Adapter { }); let current = createSegment(); - let structured: Segment | undefined; + const structured = new Set(); let lastResult: StopResult | undefined; let lastAppended = ""; + let plan: Extract | undefined; const taskParts = new Map(); + const taskSegments = new Map(); const fence = new Fence(); const renderer = new StreamingMarkdownRenderer({ wrapTablesForAppend: false, @@ -4054,7 +4059,7 @@ export class SlackAdapter implements Adapter { current.hasContent = true; current.length += closing.length; } - if (current !== structured) { + if (!structured.has(current)) { await stopSegment(current); } current = createSegment(); @@ -4122,6 +4127,44 @@ export class SlackAdapter implements Adapter { * Text streaming continues unaffected. */ let structuredChunksSupported = true; + const appendStructuredChunk = async ( + segment: Segment, + chunk: Exclude + ): Promise => { + await segment.streamer.append({ + chunks: [chunk], + token, + // biome-ignore lint/suspicious/noExplicitAny: chunks not in ChatAppendStreamArguments for older @slack/web-api + } as any); + segment.hasContent = true; + }; + + const createStructuredSegment = async (): Promise => { + current = createSegment(); + structured.add(current); + if (plan) { + await appendStructuredChunk(current, plan); + } + return current; + }; + + const getTaskSegment = async (id: string): Promise => { + const existing = taskSegments.get(id); + if (existing) { + return existing; + } + + if (current.cards >= STREAM_TASK_LIMIT) { + await createStructuredSegment(); + } else { + structured.add(current); + } + + current.cards += 1; + taskSegments.set(id, current); + return current; + }; + const sendStructuredChunk = async ( chunk: Exclude ): Promise => { @@ -4136,28 +4179,26 @@ export class SlackAdapter implements Adapter { lastAppended = committable; try { - structured ??= current; - const chunks = - chunk.type === "task_update" - ? splitTask(chunk, taskParts.get(chunk.id) ?? 0) - : [ - { - ...chunk, - title: truncateText(chunk.title, STREAM_CHUNK_LIMIT), - }, - ]; - - if (chunk.type === "task_update") { - taskParts.set(chunk.id, chunks.length); + if (chunk.type === "plan_update") { + plan = { + ...chunk, + title: truncateText(chunk.title, STREAM_CHUNK_LIMIT), + }; + if (structured.size === 0) { + structured.add(current); + } + for (const segment of structured) { + await appendStructuredChunk(segment, plan); + } + return; } + const chunks = splitTask(chunk, taskParts.get(chunk.id) ?? 0); + taskParts.set(chunk.id, chunks.length); + for (const normalized of chunks) { - await structured.streamer.append({ - chunks: [normalized], - token, - // biome-ignore lint/suspicious/noExplicitAny: chunks not in ChatAppendStreamArguments for older @slack/web-api - } as any); - structured.hasContent = true; + const segment = await getTaskSegment(normalized.id); + await appendStructuredChunk(segment, normalized); } } catch (error) { // Structured chunks may fail if the app doesn't have the required @@ -4208,8 +4249,10 @@ export class SlackAdapter implements Adapter { current.length += fence.closing.length; } - if (structured && structured !== current) { - await stopSegment(structured); + for (const segment of structured) { + if (segment !== current) { + await stopSegment(segment); + } } lastResult = await stopSegment( current, @@ -4217,10 +4260,10 @@ export class SlackAdapter implements Adapter { true ); } catch (error) { - const segments = new Set([structured, current]); + const segments = new Set([...structured, current]); fence.finish(); for (const segment of segments) { - if (!segment?.hasContent) { + if (!segment.hasContent) { continue; } try { diff --git a/packages/adapter-slack/src/stream.ts b/packages/adapter-slack/src/stream.ts index 82e02d48..50a522e0 100644 --- a/packages/adapter-slack/src/stream.ts +++ b/packages/adapter-slack/src/stream.ts @@ -3,6 +3,7 @@ import type { StreamChunk } from "chat"; export const STREAM_BUFFER_SIZE = 256; export const STREAM_SEGMENT_LIMIT = 11_500; export const STREAM_CHUNK_LIMIT = 256; +export const STREAM_TASK_LIMIT = 50; export const STREAM_FENCE_RESERVE = 64; type TaskChunk = Extract;