fix(slack): bound each long-answer append request

Author: Sg312

apps/sim/lib/webhooks/slack-execution-stream.test.ts

@@ -40,6 +40,7 @@ vi.mock('@/lib/webhooks/slack-stream-sessions', () => ({
 }))
 
 import { ExecuteEventProjection } from '@/lib/mothership/request/lifecycle/execute-events'
+import type { SlackStreamChunk } from '@/lib/webhooks/slack-agent-api'
 import { SlackExecutionStreamController } from '@/lib/webhooks/slack-execution-stream'
 import type { SlackStreamResponseConfig } from '@/lib/webhooks/slack-stream-config'
 import { type AgentStreamEvent, createAgentEventReadableStream } from '@/providers/stream-events'
@@ -598,6 +599,65 @@ describe('SlackExecutionStreamController', () => {
     controller.assertSucceeded()
   })
 
+  it.each(['live', 'settled'] as const)(
+    'delivers a long %s answer with each Slack append inside the request limit',
+    async (delivery) => {
+      const answer = `${'x'.repeat(11_999)}🚀${'y'.repeat(17_000)}The complete ending.`
+      mockAppendSlackAgentStream.mockImplementation(
+        async (_token, _channel, _ts, chunks: SlackStreamChunk[]) => {
+          const text = chunks
+            .filter((chunk) => chunk.type === 'markdown_text')
+            .map((chunk) => chunk.text)
+            .join('')
+          if (text.length > 12_000) throw new Error('Slack chat.appendStream: msg_too_long')
+          expect(text.isWellFormed()).toBe(true)
+        }
+      )
+      const { controller } = await deliver(
+        delivery === 'live'
+          ? [
+              { type: 'text_delta', text: answer, turn: 'pending' },
+              { type: 'turn_end', turn: 'final' },
+            ]
+          : [],
+        answer
+      )
+      controller.assertSucceeded()
+      expect(sentText()).toBe(answer)
+      expect(mockAppendSlackAgentStream).toHaveBeenCalledTimes(3)
+      expect(mockStartSlackAgentStream).toHaveBeenCalledTimes(1)
+      expect(mockStopSlackAgentStream).toHaveBeenCalledTimes(1)
+    }
+  )
+
+  it('does not replay acknowledged long-answer chunks after a later append fails', async () => {
+    mockAppendSlackAgentStream.mockResolvedValueOnce(undefined)
+    mockAppendSlackAgentStream.mockRejectedValueOnce(new Error('append acknowledgment lost'))
+    const answer = `${'x'.repeat(25_000)}The complete ending.`
+    const { controller } = await deliver(
+      [
+        { type: 'text_delta', text: answer, turn: 'pending' },
+        { type: 'turn_end', turn: 'final' },
+      ],
+      answer
+    )
+    expect(() => controller.assertSucceeded()).toThrow('append acknowledgment lost')
+    expect(mockAppendSlackAgentStream).toHaveBeenCalledTimes(2)
+    expect(mockAppendSlackAgentStream.mock.calls.map((call) => call[3])).toEqual([
+      [{ type: 'markdown_text', text: answer.slice(0, 12_000) }],
+      [{ type: 'markdown_text', text: answer.slice(12_000, 24_000) }],
+    ])
+    expect(mockStopSlackAgentStream).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      'suspended',
+      undefined,
+      undefined,
+      [expect.objectContaining({ status: 'error' })]
+    )
+  })
+
   it('reconciles a settled suffix without duplicating acknowledged text', async () => {
     const { controller } = await deliver(
       [

apps/sim/lib/webhooks/slack-execution-stream.ts

@@ -87,13 +87,14 @@ export function resolveSlackReplyTarget(triggerInput: Record<string, unknown>):
   }
 }
 
-function splitMarkdown(text: string): SlackStreamChunk[] {
-  const chunks: SlackStreamChunk[] = []
-  for (let offset = 0; offset < text.length; offset += SLACK_MARKDOWN_LIMIT) {
-    chunks.push({
-      type: 'markdown_text',
-      text: text.slice(offset, offset + SLACK_MARKDOWN_LIMIT),
-    })
+function splitMarkdown(text: string): string[] {
+  const chunks: string[] = []
+  for (let offset = 0; offset < text.length; ) {
+    let end = Math.min(offset + SLACK_MARKDOWN_LIMIT, text.length)
+    const lastCodeUnit = text.charCodeAt(end - 1)
+    if (end < text.length && lastCodeUnit >= 0xd800 && lastCodeUnit <= 0xdbff) end--
+    chunks.push(text.slice(offset, end))
+    offset = end
   }
   return chunks
 }
@@ -156,6 +157,14 @@ class SlackInvocationStream {
     await appendSlackAgentStream(this.token, this.channel!, this.ts!, chunks, this.signal)
   }
 
+  /** Slack limits the whole append request, not each chunk within its array. */
+  private async appendAnswerText(text: string): Promise<void> {
+    for (const chunk of splitMarkdown(text)) {
+      await this.append([{ type: 'markdown_text', text: chunk }])
+      this.acknowledgedAnswer += chunk
+    }
+  }
+
   private async flushAnswer(force: boolean): Promise<void> {
     if (
       this.failure ||
@@ -165,8 +174,7 @@ class SlackInvocationStream {
       return
     const projected = await this.projectLiveText(this.answerBuffer)
     if (projected === null) return
-    if (projected) await this.append(splitMarkdown(projected))
-    this.acknowledgedAnswer += projected
+    await this.appendAnswerText(projected)
     this.answerBuffer = ''
   }
 
@@ -267,8 +275,7 @@ class SlackInvocationStream {
         throw new Error('Slack delivery failed: settled output differs from acknowledged answer')
       }
       const remaining = text.slice(this.acknowledgedAnswer.length)
-      if (remaining) await this.append(splitMarkdown(remaining))
-      this.acknowledgedAnswer = text
+      await this.appendAnswerText(remaining)
       this.answerBuffer = ''
     })
   }