diff --git a/.changeset/fresh-ducks-stream.md b/.changeset/fresh-ducks-stream.md new file mode 100644 index 00000000..95ffdce8 --- /dev/null +++ b/.changeset/fresh-ducks-stream.md @@ -0,0 +1,5 @@ +--- +"@chat-adapter/slack": minor +--- + +Rotate long-running native Slack streams before Slack expires them. Once a stream segment passes `streamSegmentMaxAgeMs` (default four minutes) the adapter finalizes it at the next paragraph break and continues the reply in a new message, closing and reopening code fences, repeating table headers, replaying open task cards and the plan title, and keeping the agent session in `processing`. A segment Slack already expired during an idle gap is recovered the same way instead of failing the reply. diff --git a/apps/docs/content/adapters/official/slack.mdx b/apps/docs/content/adapters/official/slack.mdx index e3b986cf..b5f081e2 100644 --- a/apps/docs/content/adapters/official/slack.mdx +++ b/apps/docs/content/adapters/official/slack.mdx @@ -164,6 +164,12 @@ bot.onNewMention(async (thread, message) => { description: "Use Slack's native streaming API for streamed posts. Set false to always stream via post-and-edit.", }, + streamSegmentMaxAgeMs: { + type: "number", + default: "240000", + description: + "Finalize and continue native streams in a new message after this many milliseconds to stay below Slack's roughly five-minute stream expiry. Rotation waits up to 30 seconds for a paragraph break. Set Infinity to disable.", + }, feedbackButtons: { type: "boolean | SlackFeedbackButtonsOptions", description: @@ -828,6 +834,8 @@ Threads without streaming context fall back to post-and-edit (`chat.update` delt const slack = createSlackAdapter({ nativeStreaming: false }); ``` +Slack expires a native stream after roughly five minutes. A reply that streams longer than that is finalized and continued in a new message: once a segment is four minutes old (`streamSegmentMaxAgeMs`, default `240000`; `Infinity` disables rotation), the adapter rotates at the next paragraph break, or after at most 30 more seconds if none arrives. Across the boundary an open code fence is closed and reopened, a table that continues gets its header repeated, the plan title and any task cards still in progress are replayed so later updates land on them, and with `agentView` the session stays in `processing`. The finalized message keeps its task cards in their last state, and a list split across the boundary restarts its numbering. The `SentMessage` returned by `thread.post()` refers to the last message of the reply. If Slack expires a segment during a long idle gap anyway, the adapter continues in a new message with any text Slack had not confirmed rather than failing the reply. + #### Feedback buttons Slack's agent UX guidance recommends native thumbs up/down feedback on agent replies (a `context_actions` block with a `feedback_buttons` element). Configure `feedbackButtons` and the adapter appends them to every streamed reply when the stream finishes: diff --git a/packages/adapter-slack/AGENTS.md b/packages/adapter-slack/AGENTS.md index 1069e38e..65f8a375 100644 --- a/packages/adapter-slack/AGENTS.md +++ b/packages/adapter-slack/AGENTS.md @@ -273,7 +273,19 @@ Fallback to post-and-edit (`chat.update` deltas, throttled by skipped and text streaming continues. Failures after native content has rendered still propagate — mixing the -two surfaces would duplicate output. +two surfaces would duplicate output. The one exception is Slack expiring +a stream (`message_not_in_streaming_state`): the adapter continues in a +new message with any text Slack had not confirmed. + +Slack expires a native stream after roughly five minutes, so `stream()` +works in segments: once a segment is `streamSegmentMaxAgeMs` old (the +clock starts at the first call Slack accepts, not at construction), the +next paragraph break (or, after a 30 s grace, the next line break) +finalizes it and a new `chatStream` continues the reply. Rotation closes +and reopens an open code fence, repeats a table header, replays the plan +and open task cards, and keeps `session_status: "processing"` with +`agentView`. Tests drive this with a mocked `Date.now()`; see the +"native stream rotation" describe block. ## Socket Mode diff --git a/packages/adapter-slack/src/index.test.ts b/packages/adapter-slack/src/index.test.ts index e7b17c8e..cafa49ea 100644 --- a/packages/adapter-slack/src/index.test.ts +++ b/packages/adapter-slack/src/index.test.ts @@ -10469,6 +10469,478 @@ describe("stream with empty threadTs", () => { }); }); +describe("native stream rotation", () => { + const THREAD = "slack:D123:1234567890.000000"; + const TOKEN = "xoxb-test-token"; + // Max age 100ms; the fixed 30s grace window applies on top of it. + const MAX_AGE = 100; + const PAST_GRACE = MAX_AGE + 30_001; + + function expiredStreamError() { + return Object.assign(new Error("message_not_in_streaming_state"), { + code: "slack_webapi_platform_error", + data: { error: "message_not_in_streaming_state" }, + }); + } + + function setup(config: Record = {}, segmentCount = 3) { + const adapter = createSlackAdapter({ + botToken: TOKEN, + signingSecret: "test-signing-secret", + logger: mockLogger, + streamSegmentMaxAgeMs: MAX_AGE, + ...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 }; + } + + /** Runs `run` with a controllable Date.now(); `tick(ms)` sets the clock. */ + async function withClock( + run: (tick: (ms: number) => void) => Promise + ): Promise { + let now = 0; + const dateNow = vi.spyOn(Date, "now").mockImplementation(() => now); + try { + await run((ms) => { + now = ms; + }); + } finally { + dateNow.mockRestore(); + } + } + + it("rotates at a paragraph break past the max age and carries an open fence over", async () => { + await withClock(async (tick) => { + const { adapter, segments, chatStream } = setup(); + async function* stream() { + yield "```ts\nconst first = true;\n"; + tick(MAX_AGE + 1); + yield "const second = true;\n\nconst third = true;\n"; + yield "```\n"; + } + + const result = await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(2); + // The text before the paragraph break and the fence closer travel + // with the stop call; no extra newline since the text ends with one. + expect(segments[0].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "const second = true;\n\n```", + }); + // The new segment reopens the fence and flushes immediately. + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "```ts\nconst third = true;\n", + token: TOKEN, + chunks: [], + }); + expect(segments[1].append).toHaveBeenNthCalledWith(2, { + markdown_text: "```\n", + token: TOKEN, + }); + expect(segments[1].stop).toHaveBeenCalledWith({ token: TOKEN }); + expect(result).toMatchObject({ id: "1234567890.1" }); + }); + }); + + it("waits for a paragraph break within the grace window, then cuts at a line break", async () => { + await withClock(async (tick) => { + const { adapter, segments, chatStream } = setup(); + async function* stream() { + yield "First line.\n"; + tick(MAX_AGE + 1); + yield "Second line.\n"; + tick(PAST_GRACE); + yield "Third line.\n"; + yield "Fourth line.\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(2); + expect(segments[0].append).toHaveBeenCalledTimes(2); + expect(segments[0].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "Third line.\n", + }); + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "Fourth line.\n", + token: TOKEN, + chunks: [], + }); + }); + }); + + it("does not rotate when the final flush has nothing new to send", async () => { + await withClock(async (tick) => { + const { adapter, segments, chatStream } = setup(); + async function* stream() { + yield "hello\n"; + tick(PAST_GRACE); + } + + const result = await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(1); + expect(segments[0].stop).toHaveBeenCalledTimes(1); + expect(segments[0].stop).toHaveBeenCalledWith({ token: TOKEN }); + expect(result).toMatchObject({ id: "1234567890.0" }); + }); + }); + + it("starts the segment clock at the first call Slack accepts, not at construction", async () => { + await withClock(async (tick) => { + const { adapter, chatStream, segments } = setup(); + // The first delta is only buffered locally: no Slack stream yet. + segments[0].append.mockResolvedValueOnce(null); + async function* stream() { + yield "a\n"; + tick(1000); + yield "b\n"; + tick(1000 + MAX_AGE - 1); + yield "c\n\nd\n"; + expect(chatStream).toHaveBeenCalledTimes(1); + tick(1000 + MAX_AGE + 1); + yield "e\n\nf\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(2); + }); + }); + + it("keeps the agent session processing while the reply continues", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup({ agentView: true }); + async function* stream() { + yield "a\n"; + tick(MAX_AGE + 1); + yield "b\n\nc\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[0].stop).toHaveBeenCalledWith( + expect.objectContaining({ session_status: "processing" }) + ); + expect(segments[1].stop).toHaveBeenCalledWith( + expect.objectContaining({ session_status: "active" }) + ); + }); + }); + + it("continues in a new message when Slack expired the segment before rotation", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + segments[0].append + .mockResolvedValueOnce({ ok: true }) + .mockResolvedValueOnce(null); + segments[0].stop.mockRejectedValue(expiredStreamError()); + async function* stream() { + yield "first\n"; + // Buffered only: never confirmed by Slack. + yield "second\n"; + tick(MAX_AGE + 1); + yield "third\n\nfourth\n"; + } + + const result = await adapter.stream(THREAD, stream()); + + // Everything after the last confirmed flush is resent. + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "second\nthird\n\nfourth\n", + token: TOKEN, + chunks: [], + }); + expect(result).toMatchObject({ id: "1234567890.1" }); + expect(mockLogger.warn).toHaveBeenCalledWith( + expect.stringContaining("expired before rotation"), + expect.anything() + ); + }); + }); + + it("delivers unconfirmed text in a new message when the last segment expired before stop", async () => { + const { adapter, segments } = setup(); + segments[0].append + .mockResolvedValueOnce({ ok: true }) + .mockResolvedValueOnce(null); + segments[0].stop.mockRejectedValue(expiredStreamError()); + async function* stream() { + yield "first\n"; + yield "second\n"; + } + + const result = await adapter.stream(THREAD, stream()); + + expect(segments[1].append).not.toHaveBeenCalled(); + expect(segments[1].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "second\n", + }); + expect(result).toMatchObject({ id: "1234567890.1" }); + }); + + it("returns the finalized message when the last segment expired with everything delivered", async () => { + const { adapter, segments, chatStream } = setup({ + feedbackButtons: true, + }); + segments[0].stop.mockRejectedValue(expiredStreamError()); + async function* stream() { + yield "first\n"; + } + + const result = await adapter.stream(THREAD, stream()); + + // No empty message is posted just to carry the feedback buttons. + expect(chatStream).toHaveBeenCalledTimes(1); + expect(result).toMatchObject({ id: "1234567890.0" }); + expect(mockLogger.warn).toHaveBeenCalledWith( + expect.stringContaining("stream-end blocks skipped"), + expect.objectContaining({ skippedBlocks: 1 }) + ); + }); + + it("still propagates non-expiry failures during rotation", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + segments[0].stop.mockRejectedValue(new Error("rotate boom")); + async function* stream() { + yield "a\n"; + tick(MAX_AGE + 1); + yield "b\n\nc\n"; + } + + await expect(adapter.stream(THREAD, stream())).rejects.toThrow( + "rotate boom" + ); + }); + }); + + it("replays the plan and open task cards into the new segment", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + const plan = { type: "plan_update" as const, title: "Plan" }; + const oneInProgress = { + type: "task_update" as const, + id: "t1", + title: "One", + status: "in_progress" as const, + }; + const oneComplete = { ...oneInProgress, status: "complete" as const }; + const twoComplete = { + type: "task_update" as const, + id: "t2", + title: "Two", + status: "complete" as const, + }; + async function* stream() { + yield plan; + yield oneInProgress; + yield twoComplete; + tick(MAX_AGE + 1); + // A structured chunk is a block boundary: rotate right away. + yield oneComplete; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[0].stop).toHaveBeenCalledWith({ token: TOKEN }); + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + chunks: [plan, oneInProgress], + token: TOKEN, + }); + expect(segments[1].append).toHaveBeenNthCalledWith(2, { + chunks: [oneComplete], + token: TOKEN, + }); + }); + }); + + it("repeats the table header when a table continues in the new segment", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + async function* stream() { + yield "| a | b |\n|---|---|\n| 1 | 2 |\n"; + tick(PAST_GRACE); + yield "| 3 | 4 |\n"; + yield "| 5 | 6 |\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[0].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "| 3 | 4 |\n", + }); + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "| a | b |\n|---|---|\n| 5 | 6 |\n", + token: TOKEN, + chunks: [], + }); + }); + }); + + it("does not repeat the table header when the new segment starts with prose", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + async function* stream() { + yield "| a | b |\n|---|---|\n| 1 | 2 |\n"; + tick(PAST_GRACE); + yield "| 3 | 4 |\n"; + yield "\nSummary.\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "\nSummary.\n", + token: TOKEN, + chunks: [], + }); + }); + }); + + it("closes a tilde fence with tildes even when backtick fences appear inside it", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + async function* stream() { + yield "~~~\n```js\nconst x = 1;\n```\n"; + tick(MAX_AGE + 1); + yield "still literal\n\nmore\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[0].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "still literal\n\n~~~", + }); + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "~~~\nmore\n", + token: TOKEN, + chunks: [], + }); + }); + }); + + it("finishes a fenced block in the old segment when the pending text closes it", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + async function* stream() { + yield "```ts\nconst a = 1;\n"; + tick(MAX_AGE + 1); + yield "```\n\nAfter.\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[0].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "```\n\n", + }); + // No empty reopened block at the top of the new segment. + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "After.\n", + token: TOKEN, + chunks: [], + }); + }); + }); + + it("treats a pending partial closing fence as the block's end", async () => { + await withClock(async (tick) => { + const { adapter, segments } = setup(); + async function* stream() { + yield "```ts\nconst a = 1;\n"; + tick(PAST_GRACE); + yield "```"; + yield "\nAfter.\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(segments[0].stop).toHaveBeenCalledWith({ + token: TOKEN, + markdown_text: "```", + }); + expect(segments[1].append).toHaveBeenNthCalledWith(1, { + markdown_text: "\nAfter.\n", + token: TOKEN, + chunks: [], + }); + }); + }); + + it("does not rotate a segment before Slack has started it", async () => { + await withClock(async (tick) => { + const { adapter, chatStream, segments } = setup(); + segments[0].append.mockResolvedValue(null); + async function* stream() { + yield "a\n"; + tick(PAST_GRACE); + yield "b\n\nc\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(1); + expect(segments[0].stop).toHaveBeenCalledTimes(1); + }); + }); + + it.each([ + 0, + -5, + Number.NaN, + ])("falls back to the default max age for streamSegmentMaxAgeMs %s", async (value) => { + await withClock(async (tick) => { + const { adapter, chatStream } = setup({ + streamSegmentMaxAgeMs: value, + }); + async function* stream() { + yield "a\n"; + tick(239_999); + yield "b\n\nc\n"; + expect(chatStream).toHaveBeenCalledTimes(1); + tick(240_000); + yield "d\n\ne\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(2); + }); + }); + + it("never rotates when streamSegmentMaxAgeMs is Infinity", async () => { + await withClock(async (tick) => { + const { adapter, chatStream } = setup({ + streamSegmentMaxAgeMs: Number.POSITIVE_INFINITY, + }); + async function* stream() { + yield "a\n"; + tick(10_000_000); + yield "b\n\nc\n"; + } + + await adapter.stream(THREAD, stream()); + + expect(chatStream).toHaveBeenCalledTimes(1); + }); + }); +}); + 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 d1ac08af..399f7fcc 100644 --- a/packages/adapter-slack/src/index.ts +++ b/packages/adapter-slack/src/index.ts @@ -43,12 +43,14 @@ import type { OptionsLoadResult, PlanContent, PlanModel, + PlanUpdateChunk, RawMessage, ReactionEvent, ScheduledMessage, SelectOptionElement, StreamChunk, StreamOptions, + TaskUpdateChunk, ThreadInfo, ThreadSummary, TypingOptions, @@ -301,6 +303,127 @@ import type { SlackSuggestedPromptsContext, } from "./types"; +/** Default lifetime of one native stream segment before it is rotated. */ +const DEFAULT_STREAM_SEGMENT_MAX_AGE_MS = 240_000; +/** + * Once a segment is past its max age, rotation waits up to this long for a + * paragraph break so a paragraph, list, or table is not split across two + * messages. The default max age plus this grace stays below Slack's roughly + * five-minute stream expiry. + */ +const STREAM_SEGMENT_ROTATION_GRACE_MS = 30_000; +/** Slack's error for appending to or stopping a stream it already expired. */ +const STREAM_EXPIRED_ERROR = "message_not_in_streaming_state"; +/** Fenced code block delimiter: up to three spaces, then 3+ backticks or tildes. */ +const FENCE_LINE_PATTERN = /^ {0,3}(`{3,}|~{3,})(.*)$/; +const TABLE_ROW_PATTERN = /^\|.*\|$/; +const TABLE_SEPARATOR_PATTERN = /^\|[\s:]*-+[\s:]*(?:\|[\s:]*-+[\s:]*)*\|$/; + +interface OpenFence { + /** The fence run that opened the block, e.g. "```" or "~~~~". */ + marker: string; + /** The full opening line (marker plus info string), used to reopen it. */ + opening: string; +} + +/** + * Tracks fenced code block state line by line, CommonMark style: a fence + * closes only on a run of the same character at least as long as its opener + * with nothing but whitespace after it, and other fence-looking lines inside + * the block are literal content. + */ +class FenceTracker { + open: OpenFence | undefined; + + /** + * Feed one complete line (without its newline). Returns true when the line + * opened or closed a fence. + */ + feed(line: string): boolean { + const match = FENCE_LINE_PATTERN.exec(line); + if (!match) { + return false; + } + const [, marker, info] = match; + if (this.open) { + if ( + marker[0] === this.open.marker[0] && + marker.length >= this.open.marker.length && + info.trim() === "" + ) { + this.open = undefined; + return true; + } + return false; + } + // A backtick fence's info string cannot contain backticks. + if (marker[0] === "`" && info.includes("`")) { + return false; + } + this.open = { marker, opening: line.trimEnd() }; + return true; + } +} + +/** + * The code fence left open at the end of `text`, if any. A trailing partial + * line is fed too: the streaming renderer only commits partial lines inside a + * fence, so it is either literal content or a closing delimiter. + */ +function openFenceIn(text: string): OpenFence | undefined { + const tracker = new FenceTracker(); + for (const line of text.split("\n")) { + tracker.feed(line); + } + return tracker.open; +} + +/** Whether `line`, appended after a line break, would close `fence`. */ +function closesFence(fence: OpenFence, line: string): boolean { + const tracker = new FenceTracker(); + tracker.open = fence; + return tracker.feed(line); +} + +/** + * Index in `text` to cut a stream segment at: after the last paragraph break + * when there is one, otherwise after the last line break (0 when none). + */ +function segmentCutIndex(text: string): number { + const paragraphBreak = text.lastIndexOf("\n\n"); + if (paragraphBreak !== -1) { + return paragraphBreak + 2; + } + return text.lastIndexOf("\n") + 1; +} + +/** + * When `text` ends inside a confirmed GFM table, the table's header and + * separator rows (each newline-terminated) so the rows that follow in a new + * segment still render as a table. Empty otherwise. + */ +function tableContinuation(text: string): string { + if (!text.endsWith("\n")) { + return ""; + } + const lines = text.split("\n"); + lines.pop(); + const rows: string[] = []; + for (let i = lines.length - 1; i >= 0; i--) { + if (!TABLE_ROW_PATTERN.test(lines[i].trim())) { + break; + } + rows.unshift(lines[i]); + } + const separatorAt = rows.findIndex((row) => + TABLE_SEPARATOR_PATTERN.test(row.trim()) + ); + if (separatorAt < 1) { + return ""; + } + return `${rows[separatorAt - 1]}\n${rows[separatorAt]}\n`; +} + export type { SlackAdapterConfig, SlackAdapterMode, @@ -1114,6 +1237,7 @@ export class SlackAdapter implements Adapter { /** Normalized feedbackButtons config (`true` becomes `{}`). */ protected readonly feedbackButtons?: SlackFeedbackButtonsOptions; protected readonly nativeStreaming: boolean; + protected readonly streamSegmentMaxAgeMs: number; /** * Latched when the workspace rejects native streaming with an error that * won't heal (e.g. `unknown_method` on GovSlack) so later streams skip the @@ -1294,6 +1418,13 @@ export class SlackAdapter implements Adapter { this.suggestedPrompts = config.suggestedPrompts; this.loadingMessages = config.loadingMessages; this.nativeStreaming = config.nativeStreaming ?? true; + // Zero, negative, and NaN would make every flush rotate; Infinity means + // never rotate. + const segmentMaxAge = config.streamSegmentMaxAgeMs; + this.streamSegmentMaxAgeMs = + segmentMaxAge !== undefined && segmentMaxAge > 0 + ? segmentMaxAge + : DEFAULT_STREAM_SEGMENT_MAX_AGE_MS; if (config.feedbackButtons) { this.feedbackButtons = config.feedbackButtons === true ? {} : config.feedbackButtons; @@ -5481,6 +5612,11 @@ export class SlackAdapter implements Adapter { * streaming API as chunk payloads, enabling native task progress cards * and plan displays in the Slack AI Assistant UI. * + * Slack expires a native stream after roughly five minutes, so a reply + * that streams longer than `streamSegmentMaxAgeMs` is finalized and + * continued in a new message. The returned `id` is the last message of + * the reply; earlier segments are already final and are not tracked. + * * Falls back to post-and-edit when the thread lacks native stream context. */ async stream( @@ -5513,21 +5649,61 @@ 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, - ...(options?.recipientUserId && { - recipient_user_id: options.recipientUserId, - }), - ...(options?.recipientTeamId && { - recipient_team_id: options.recipientTeamId, - }), - ...(options?.taskDisplayMode && { - task_display_mode: options.taskDisplayMode, - }), - }); + const createStreamer = () => + this._client.chatStream({ + channel, + thread_ts: threadTs, + ...(options?.recipientUserId && { + recipient_user_id: options.recipientUserId, + }), + ...(options?.recipientTeamId && { + recipient_team_id: options.recipientTeamId, + }), + ...(options?.taskDisplayMode && { + task_display_mode: options.taskDisplayMode, + }), + }); + // One native Slack stream (a "segment") is open at a time. Slack expires + // streams after roughly five minutes, so a long reply is finalized and + // continued in a fresh segment (see rotateSegment). `startedAt` is + // stamped by the first API call that succeeds on the segment, which is + // when Slack's expiry clock starts; until then the ChatStreamer only + // buffers text locally. + const segment: { + streamer: ReturnType; + startedAt: number | undefined; + /** + * Flush the next text append immediately instead of buffering it, so a + * segment opened by rotation becomes visible with its first text. + */ + flushNextAppend: boolean; + /** + * Code fence opening line to send before the segment's first text, + * when the previous segment was cut inside a fenced block. + */ + reopenFence: string; + /** + * Table header and separator rows to send before the segment's first + * text if that text continues the table cut by the previous segment. + */ + tableHeader: string; + } = { + streamer: createStreamer(), + startedAt: undefined, + flushNextAppend: false, + reopenFence: "", + tableHeader: "", + }; + let rotations = 0; + /** Prefix of `resolvedCommitted` handed to the current segment's streamer. */ let lastAppended = ""; + /** + * Prefix of `resolvedCommitted` Slack has confirmed. Text between here + * and `lastAppended` sits in the streamer's local buffer and has to be + * resent if the segment turns out to have expired. + */ + let lastFlushed = ""; const renderer = new StreamingMarkdownRenderer({ wrapTablesForAppend: false, }); @@ -5540,12 +5716,7 @@ export class SlackAdapter implements Adapter { // counterpart and the coordinate space `lastAppended` tracks. let resolvedCommitted = ""; let resolvedSourceDone = 0; - let insideResolvedFence = false; - - const isFenceLine = (line: string): boolean => { - const trimmed = line.trimStart(); - return trimmed.startsWith("```") || trimmed.startsWith("~~~"); - }; + const fences = new FenceTracker(); /** * Extend `resolvedCommitted` with newly committed renderer text, applying @@ -5562,19 +5733,22 @@ export class SlackAdapter implements Adapter { committable.lastIndexOf("\n", resolvedSourceDone - 1) + 1; const newlineAt = committable.indexOf("\n", resolvedSourceDone); const lineEnd = newlineAt === -1 ? committable.length : newlineAt + 1; - const segment = committable.slice(resolvedSourceDone, lineEnd); - const fenceLine = isFenceLine(committable.slice(lineStart, lineEnd)); - if (insideResolvedFence || fenceLine) { + const piece = committable.slice(resolvedSourceDone, lineEnd); + const line = committable.slice( + lineStart, + newlineAt === -1 ? lineEnd : newlineAt + ); + if (fences.open || FENCE_LINE_PATTERN.test(line)) { // Fence delimiters and fenced content are literal. - resolvedCommitted += segment; + resolvedCommitted += piece; } else { resolvedCommitted += await this.resolveOutgoingMentions( - segment, + piece, threadId ); } - if (newlineAt !== -1 && fenceLine) { - insideResolvedFence = !insideResolvedFence; + if (newlineAt !== -1) { + fences.feed(line); } resolvedSourceDone = lineEnd; } @@ -5639,6 +5813,217 @@ export class SlackAdapter implements Adapter { ); }; + const isStreamExpired = (error: unknown): boolean => + slackPlatformErrorCode(error) === STREAM_EXPIRED_ERROR; + + /** Record that Slack accepted a call on the current segment. */ + const markSegmentStarted = (): void => { + segment.startedAt ??= Date.now(); + segment.flushNextAppend = false; + fallback.nativeRendered = true; + }; + + const segmentAgeMs = (): number => + segment.startedAt === undefined ? 0 : Date.now() - segment.startedAt; + + /** + * Whether the current segment must be rotated before more content is + * sent. Past the max age, rotation waits for a block boundary (a + * paragraph break in the pending text, or a structured chunk) for up to + * the grace window, then happens regardless. A segment Slack has not + * started yet has no expiry clock and is never rotated. + */ + const rotationDue = (atBlockBoundary: boolean): boolean => { + if (segment.startedAt === undefined) { + return false; + } + const age = segmentAgeMs(); + if (age < this.streamSegmentMaxAgeMs) { + return false; + } + return ( + atBlockBoundary || + age >= this.streamSegmentMaxAgeMs + STREAM_SEGMENT_ROTATION_GRACE_MS + ); + }; + + // Structured chunk state, replayed into every new segment. Task cards + // and the plan title belong to one Slack message, so without the replay + // an update for a task first shown in an earlier segment would render + // as a brand-new card. Structured chunks may fail if the app lacks the + // Assistant scopes/features; they are then disabled for the rest of the + // stream to avoid repeated failures, and logged once. + let structuredChunksSupported = true; + const openTasks = new Map(); + let currentPlan: PlanUpdateChunk | undefined; + const rememberStructuredChunk = (chunk: StreamChunk): void => { + if (chunk.type === "plan_update") { + currentPlan = chunk; + } else if (chunk.type === "task_update") { + if (chunk.status === "complete" || chunk.status === "error") { + openTasks.delete(chunk.id); + } else { + openTasks.set(chunk.id, chunk); + } + } + }; + const disableStructuredChunks = ( + chunkType: string, + error: unknown + ): void => { + structuredChunksSupported = false; + this.logger.warn( + "Structured streaming chunk failed, falling back to text-only streaming. " + + "Ensure your Slack app manifest includes the agent/assistant feature " + + "and the assistant:write scope", + { chunkType, error } + ); + }; + + /** + * Open a new segment that continues from `sent`, the resolved text Slack + * holds so far. Open task cards and the plan title are replayed right + * away; when `sent` ends inside a code fence or a table, the fence + * opening or the table header is queued to precede the segment's first + * text (see `withSegmentPrefix`) rather than sent on its own. + */ + const startNextSegment = async (sent: string): Promise => { + segment.streamer = createStreamer(); + segment.startedAt = undefined; + segment.flushNextAppend = true; + const fence = openFenceIn(sent); + segment.reopenFence = fence ? `${fence.opening}\n` : ""; + segment.tableHeader = fence ? "" : tableContinuation(sent); + rotations++; + const replay: StreamChunk[] = [ + ...(currentPlan ? [currentPlan] : []), + ...openTasks.values(), + ]; + if (replay.length > 0 && structuredChunksSupported) { + try { + await segment.streamer.append({ + chunks: replay as ChatAppendStreamArguments["chunks"], + token, + }); + markSegmentStarted(); + } catch (error) { + disableStructuredChunks("replay", error); + } + } + }; + + /** + * Prepend whatever a rotation queued for the segment's first text: the + * reopened code fence always (fenced content is literal), the repeated + * table header only when `text` starts with a table row. + */ + const withSegmentPrefix = (text: string): string => { + const firstLine = text.slice( + 0, + text.includes("\n") ? text.indexOf("\n") : text.length + ); + const header = TABLE_ROW_PATTERN.test(firstLine.trim()) + ? segment.tableHeader + : ""; + const prefixed = segment.reopenFence + header + text; + segment.reopenFence = ""; + segment.tableHeader = ""; + return prefixed; + }; + + /** + * Finalize the current segment and continue in a new one. `delta` is the + * resolved text about to be sent: the part up to its last paragraph + * break (or last line break) travels with the old segment's stop call so + * the cut lands on a block boundary, and a code fence still open there + * is closed. Returns the text to send on the new segment. + */ + const rotateSegment = async (delta: string): Promise => { + const cutAt = segmentCutIndex(delta); + let head = delta.slice(0, cutAt); + let tail = delta.slice(cutAt); + let sent = lastAppended + head; + let fence = openFenceIn(sent); + if (fence && sent.endsWith("\n") && closesFence(fence, tail)) { + // The pending partial line is the closing delimiter: finish the + // block in the old segment instead of reopening it in the new one. + head += tail; + sent += tail; + tail = ""; + fence = undefined; + } + const closer = fence + ? `${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. + const finalText = + (head.length > 0 ? withSegmentPrefix(head) : "") + closer; + const ageMs = segmentAgeMs(); + try { + const result = await segment.streamer.stop({ + token, + ...(finalText.length > 0 ? { markdown_text: finalText } : {}), + // Keep the agent session marked busy: the reply continues in the + // next segment, and chat.stopStream defaults to "active". + ...(this.agentView ? { session_status: "processing" } : {}), + } as ChatStopStreamArguments & { + session_status?: AgentSessionStatus; + }); + lastFlushed = sent; + this.logger.debug("Slack: rotated stream segment", { + channel, + messageId: result.ts, + ageMs, + }); + } catch (error) { + if (!isStreamExpired(error)) { + throw error; + } + // Slack expired the segment during an idle gap. Nothing after the + // last confirmed flush reached it, so that text moves to the new + // segment instead of failing the reply. + this.logger.warn( + "Slack: stream segment expired before rotation, continuing in a new message", + { channel, ageMs } + ); + tail = lastAppended.slice(lastFlushed.length) + delta; + sent = lastFlushed; + } + await startNextSegment(sent); + return tail; + }; + + /** + * 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 + * out with the old segment, so only what it returns goes to the new one. + */ + const sendDelta = async ( + delta: string, + atBlockBoundary: boolean + ): Promise => { + const text = rotationDue(atBlockBoundary || delta.includes("\n\n")) + ? await rotateSegment(delta) + : delta; + if (text.length > 0) { + // 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: [] } : {}), + }); + if (response) { + markSegmentStarted(); + lastFlushed = resolvedCommitted; + } + } + lastAppended = resolvedCommitted; + }; + /** * Flush committed renderer text: as a mention-resolved markdown_text * delta on the native stream, or as a throttled post/edit in fallback @@ -5656,17 +6041,7 @@ export class SlackAdapter implements Adapter { return; } try { - // 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. - const response = await streamer.append({ - markdown_text: delta, - token, - }); - if (response) { - fallback.nativeRendered = true; - } - lastAppended = resolvedCommitted; + await sendDelta(delta, false); } catch (error) { if (fallback.nativeRendered) { // A native call succeeded earlier; content is already rendering @@ -5681,15 +6056,10 @@ export class SlackAdapter implements Adapter { /** * Helper to send a structured chunk (task_update, plan_update, etc.) * directly to Slack's streaming API. Any buffered markdown text is - * flushed first to maintain correct ordering. - * - * If the Slack API rejects the chunk (e.g. missing assistant:write scope - * or Assistant features not enabled in the app manifest), the error is - * logged and the chunk is silently skipped. Text streaming continues - * unaffected. In fallback mode structured chunks are skipped — task - * cards only exist on the native streaming surface. + * flushed first to maintain correct ordering. In fallback mode + * structured chunks are skipped — task cards only exist on the native + * streaming surface. */ - let structuredChunksSupported = true; const sendStructuredChunk = async (chunk: StreamChunk): Promise => { // Flush any buffered markdown before sending the structured chunk await flushCommitted(); @@ -5702,23 +6072,21 @@ export class SlackAdapter implements Adapter { return; } + // A structured chunk is a block boundary, so a segment past its max + // age rotates here instead of waiting for a paragraph break. + await sendDelta("", true); + try { - await streamer.append({ + await segment.streamer.append({ chunks: [chunk] as ChatAppendStreamArguments["chunks"], token, }); - fallback.nativeRendered = true; + markSegmentStarted(); + // Sending chunks flushes the streamer's text buffer as well. + lastFlushed = lastAppended; + rememberStructuredChunk(chunk); } catch (error) { - // Structured chunks may fail if the app doesn't have the required - // Assistant scopes/features. Disable for the rest of this stream - // to avoid repeated failures and log once. - structuredChunksSupported = false; - this.logger.warn( - "Structured streaming chunk failed, falling back to text-only streaming. " + - "Ensure your Slack app manifest includes the agent/assistant feature " + - "and the assistant:write scope", - { chunkType: chunk.type, error } - ); + disableStructuredChunks(chunk.type, error); } }; @@ -5764,38 +6132,71 @@ export class SlackAdapter implements Adapter { ? [buildFeedbackButtonsBlock(this.feedbackButtons)] : []), ]; - let result: Awaited>; + const stopArgs = { + token, + ...(this.agentView + ? { session_status: options?.sessionStatus ?? "active" } + : {}), + ...(stopBlocks.length > 0 + ? { blocks: stopBlocks as ChatStopStreamArguments["blocks"] } + : {}), + } as ChatStopStreamArguments & { + session_status?: AgentSessionStatus; + }; + let result: Awaited>; try { - result = await streamer.stop({ - token, - ...(this.agentView - ? { session_status: options?.sessionStatus ?? "active" } - : {}), - ...(stopBlocks.length > 0 - ? { blocks: stopBlocks as ChatStopStreamArguments["blocks"] } - : {}), - } as ChatStopStreamArguments & { - session_status?: AgentSessionStatus; - }); + result = await segment.streamer.stop(stopArgs); } catch (error) { - if (fallback.nativeRendered) { + if (!fallback.nativeRendered) { + // Short streams can buffer every delta in the streamer, making stop() + // the FIRST real API call — on an unsupported workspace this is where + // the failure lands, so the post-and-edit fallback must engage here + // too. Stream-end blocks are skipped, as in any fallback. + switchToFallback(error); + await flushFallback(true); + this.logger.debug("Slack: fallback stream complete", { + messageId: fallback.message?.id, + }); + await this.endTyping(threadId, options?.sessionStatus ?? "active"); + return fallback.message; + } + if (!isStreamExpired(error)) { throw error; } - // Short streams can buffer every delta in the streamer, making stop() - // the FIRST real API call — on an unsupported workspace this is where - // the failure lands, so the post-and-edit fallback must engage here - // too. Stream-end blocks are skipped, as in any fallback. - switchToFallback(error); - await flushFallback(true); - this.logger.debug("Slack: fallback stream complete", { - messageId: fallback.message?.id, + // Slack expired the last segment during a trailing idle gap and has + // already finalized that message. + const unconfirmed = lastAppended.slice(lastFlushed.length); + if (unconfirmed.length === 0) { + // Every delta was delivered, so the reply is complete; only the + // stream-end blocks are lost. Report the finalized message rather + // than posting an empty one just to carry the blocks. + const expiredTs = segment.streamer.ts; + if (!expiredTs) { + throw error; + } + this.logger.warn( + "Slack: stream expired before stop, stream-end blocks skipped", + { channel, messageId: expiredTs, skippedBlocks: stopBlocks.length } + ); + await this.endTyping(threadId, options?.sessionStatus ?? "active"); + return { id: expiredTs, threadId, raw: { ts: expiredTs } }; + } + this.logger.warn( + "Slack: stream expired before stop, delivering the rest in a new message", + { channel } + ); + await startNextSegment(lastFlushed); + result = await segment.streamer.stop({ + ...stopArgs, + markdown_text: withSegmentPrefix(unconfirmed), }); - await this.endTyping(threadId, options?.sessionStatus ?? "active"); - return fallback.message; } const messageTs = (result.message?.ts ?? result.ts) as string; - this.logger.debug("Slack: stream complete", { messageId: messageTs }); + this.logger.debug("Slack: stream complete", { + messageId: messageTs, + segments: rotations + 1, + }); return { id: messageTs, @@ -6997,6 +7398,7 @@ export function createSlackAdapter(config?: SlackAdapterConfig): SlackAdapter { loadingMessages: config?.loadingMessages, logger: config?.logger ?? new ConsoleLogger("info").child("slack"), nativeStreaming: config?.nativeStreaming, + streamSegmentMaxAgeMs: config?.streamSegmentMaxAgeMs, sessionTitle: config?.sessionTitle, suggestedPrompts: config?.suggestedPrompts, socketForwardingSecret: diff --git a/packages/adapter-slack/src/types.ts b/packages/adapter-slack/src/types.ts index 2c0cfa77..7b6109bd 100644 --- a/packages/adapter-slack/src/types.ts +++ b/packages/adapter-slack/src/types.ts @@ -194,6 +194,12 @@ export interface SlackAdapterConfig { signingSecret?: string; /** Shared secret for authenticating forwarded socket mode events. Auto-detected from SLACK_SOCKET_FORWARDING_SECRET. Falls back to appToken if not set. */ socketForwardingSecret?: string; + /** + * Maximum lifetime of one native Slack stream segment before the adapter + * finalizes it and continues in a new segment. Defaults to 240 seconds, + * safely below Slack's roughly five-minute stream expiry. + */ + streamSegmentMaxAgeMs?: number; /** * Suggested prompts to pin automatically when an assistant/agent thread * opens. Applied on `assistant_thread_started` (legacy `assistant_view`)