fix(slack): recover explicitly rejected size overflows

Author: Sg312

apps/sim/lib/webhooks/slack-agent-api.test.ts

@@ -8,6 +8,7 @@ import {
   startSlackAgentStream,
   stopSlackAgentStream,
 } from '@/lib/webhooks/slack-agent-api'
+import { SlackDeliveryError } from '@/lib/webhooks/slack-delivery-error'
 
 afterEach(() => {
   vi.unstubAllGlobals()
@@ -184,6 +185,26 @@ describe('Slack agent API transport', () => {
     ).rejects.toThrow('missing_scope')
   })
 
+  it('exposes an explicit size rejection to the delivery controller without a transport retry', async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(
+        new Response(JSON.stringify({ ok: false, error: 'msg_too_long' }), { status: 200 })
+      )
+    vi.stubGlobal('fetch', fetchMock)
+    const appended = appendSlackAgentStream('token', 'C1', '1.2', [
+      { type: 'markdown_text', text: 'undelivered suffix' },
+    ])
+    await expect(appended).rejects.toBeInstanceOf(SlackDeliveryError)
+    await expect(appended).rejects.toMatchObject({
+      method: 'chat.appendStream',
+      outcome: 'rejected',
+      code: 'msg_too_long',
+      httpStatus: 200,
+    })
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+  })
+
   it('fails fast when Slack does not recognize the stop-event subscription', async () => {
     vi.stubGlobal(
       'fetch',

apps/sim/lib/webhooks/slack-agent-api.ts

@@ -1,6 +1,7 @@
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
 import { readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits'
+import { SlackDeliveryError } from '@/lib/webhooks/slack-delivery-error'
 
 export type SlackStreamChunk =
   | { type: 'markdown_text'; text: string }
@@ -31,19 +32,6 @@ interface SlackStreamTarget {
   recipientTeamId?: string
 }
 
-/** A missing acknowledgment does not establish that Slack rejected a write. */
-export class SlackDeliveryError extends Error {
-  constructor(
-    readonly method: string,
-    readonly outcome: 'rejected' | 'uncertain',
-    readonly code: string,
-    readonly httpStatus?: number
-  ) {
-    super(`Slack ${method}: ${code} (delivery ${outcome})`)
-    this.name = 'SlackDeliveryError'
-  }
-}
-
 /** Slack acknowledgments can include the full accumulated message, including task cards. */
 const MAX_SLACK_RESPONSE_BYTES = 4 * 1024 * 1024
 

apps/sim/lib/webhooks/slack-delivery-error.ts

@@ -0,0 +1,12 @@
+/** A missing acknowledgment does not establish that Slack rejected a write. */
+export class SlackDeliveryError extends Error {
+  constructor(
+    readonly method: string,
+    readonly outcome: 'rejected' | 'uncertain',
+    readonly code: string,
+    readonly httpStatus?: number
+  ) {
+    super(`Slack ${method}: ${code} (delivery ${outcome})`)
+    this.name = 'SlackDeliveryError'
+  }
+}

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

@@ -41,6 +41,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 { SlackDeliveryError } from '@/lib/webhooks/slack-delivery-error'
 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'
@@ -475,7 +476,7 @@ describe('SlackExecutionStreamController', () => {
       .join('')
   }
 
-  function enforceSlackMessageLimit() {
+  function enforceSlackMessageLimit(limit = 12_000) {
     const messages = new Map<
       string,
       { text: string; chunks: SlackStreamChunk[]; stopped: boolean }
@@ -493,8 +494,8 @@ describe('SlackExecutionStreamController', () => {
           .filter((chunk) => chunk.type === 'markdown_text')
           .map((chunk) => chunk.text)
           .join('')
-        if (message.text.length + text.length > 12_000) {
-          throw new Error('Slack chat.appendStream: msg_too_long')
+        if (message.text.length + text.length > limit) {
+          throw new SlackDeliveryError('chat.appendStream', 'rejected', 'msg_too_long', 200)
         }
         expect(text.isWellFormed()).toBe(true)
         message.text += text
@@ -674,6 +675,100 @@ describe('SlackExecutionStreamController', () => {
     }
   )
 
+  it('continues a definitively rejected 11938 + 62 boundary append without replaying accepted text', async () => {
+    const messages = enforceSlackMessageLimit(11_999)
+    const prefix = 'a'.repeat(11_938)
+    const suffix = `${'b'.repeat(3_000)}Complete ending.`
+    const { controller } = await deliver(
+      [
+        { type: 'tool_call_start', id: 'read', name: 'read' },
+        { type: 'text_delta', text: prefix, turn: 'pending' },
+        { type: 'text_delta', text: suffix, turn: 'pending' },
+        { type: 'tool_call_end', id: 'read', name: 'read', status: 'success' },
+        { type: 'turn_end', turn: 'final' },
+      ],
+      prefix + suffix
+    )
+    controller.assertSucceeded()
+    expect([...messages.values()].map((message) => message.text)).toEqual([prefix, suffix])
+    const rejected = mockAppendSlackAgentStream.mock.calls.filter(
+      (call) =>
+        call[2] === 'message-0' &&
+        call[3].some(
+          (chunk: SlackStreamChunk) =>
+            chunk.type === 'markdown_text' && chunk.text === suffix.slice(0, 62)
+        )
+    )
+    expect(rejected).toHaveLength(1)
+    expect([...messages.values()].every((message) => message.stopped)).toBe(true)
+    const toolChunks = messages
+      .get('message-0')!
+      .chunks.filter((chunk) => chunk.type === 'task_update' && chunk.id.endsWith('-tool-read'))
+    expect(toolChunks.map((chunk) => chunk.status)).toEqual(['in_progress', 'complete'])
+  })
+
+  it('reduces a definitively oversized first append and still delivers all final-only text', async () => {
+    const messages = enforceSlackMessageLimit(11_999)
+    const answer = '🚀'.repeat(13_000)
+    const { controller } = await deliver([], answer)
+    controller.assertSucceeded()
+    expect([...messages.values()].map((message) => message.text).join('')).toBe(answer)
+    expect([...messages.values()].every((message) => message.text && message.stopped)).toBe(true)
+  })
+
+  it.each([
+    new SlackDeliveryError('chat.appendStream', 'uncertain', 'msg_too_long'),
+    new SlackDeliveryError('chat.appendStream', 'rejected', 'ratelimited', 429),
+    new SlackDeliveryError('chat.appendStream', 'rejected', 'missing_scope', 200),
+  ])('does not replay ambiguous or non-size append errors: %s', async (error) => {
+    const messages = enforceSlackMessageLimit()
+    mockAppendSlackAgentStream.mockRejectedValueOnce(error)
+    const { controller } = await deliver([{ type: 'text_delta', text: 'answer' }], 'answer')
+    expect(() => controller.assertSucceeded()).toThrow(error)
+    expect(mockAppendSlackAgentStream).toHaveBeenCalledTimes(1)
+    expect(mockStartSlackAgentStream).toHaveBeenCalledTimes(1)
+    expect(messages.get('message-0')!.stopped).toBe(true)
+  })
+
+  it('bounds size recovery when even a single character is rejected', async () => {
+    const messages = enforceSlackMessageLimit(0)
+    const { controller } = await deliver([], 'abcd')
+    expect(() => controller.assertSucceeded()).toThrow('msg_too_long')
+    expect(mockAppendSlackAgentStream.mock.calls.map((call) => call[3])).toEqual([
+      [{ type: 'markdown_text', text: 'abcd' }],
+      [{ type: 'markdown_text', text: 'ab' }],
+      [{ type: 'markdown_text', text: 'a' }],
+    ])
+    expect(mockStartSlackAgentStream).toHaveBeenCalledTimes(1)
+    expect(messages.get('message-0')!.stopped).toBe(true)
+  })
+
+  it('stops recovery if the continuation append has an uncertain outcome', async () => {
+    const messages = enforceSlackMessageLimit()
+    const prefix = 'a'.repeat(11_938)
+    mockAppendSlackAgentStream
+      .mockImplementationOnce(async () => {
+        messages.get('message-0')!.text = prefix
+      })
+      .mockRejectedValueOnce(
+        new SlackDeliveryError('chat.appendStream', 'rejected', 'msg_too_long')
+      )
+      .mockRejectedValueOnce(
+        new SlackDeliveryError('chat.appendStream', 'uncertain', 'transport_failed')
+      )
+    const { controller } = await deliver(
+      [
+        { type: 'text_delta', text: prefix, turn: 'pending' },
+        { type: 'text_delta', text: 'b'.repeat(150), turn: 'pending' },
+      ],
+      prefix + 'b'.repeat(150)
+    )
+    expect(() => controller.assertSucceeded()).toThrow('transport_failed')
+    expect(mockAppendSlackAgentStream).toHaveBeenCalledTimes(3)
+    expect(mockStartSlackAgentStream).toHaveBeenCalledTimes(2)
+    expect([...messages.values()].every((message) => message.stopped)).toBe(true)
+  })
+
   it('counts commentary across turns and pairs tools on their original message after rollover', async () => {
     const messages = enforceSlackMessageLimit()
     const commentary = 'c'.repeat(11_800)

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

@@ -12,6 +12,7 @@ import {
   startSlackAgentStream,
   stopSlackAgentStream,
 } from '@/lib/webhooks/slack-agent-api'
+import { SlackDeliveryError } from '@/lib/webhooks/slack-delivery-error'
 import type {
   SlackStreamOutputConfig,
   SlackStreamResponseConfig,
@@ -189,10 +190,35 @@ class SlackInvocationStream {
       let end = Math.min(offset + SLACK_MESSAGE_TEXT_LIMIT - message.textLength, text.length)
       const lastCodeUnit = text.charCodeAt(end - 1)
       if (end < text.length && lastCodeUnit >= 0xd800 && lastCodeUnit <= 0xdbff) end--
-      const chunk = text.slice(offset, end)
-      await this.append([{ type: 'markdown_text', text: chunk }], message)
-      this.acknowledgedAnswer += chunk
-      offset = end
+      for (;;) {
+        const chunk = text.slice(offset, end)
+        try {
+          await this.append([{ type: 'markdown_text', text: chunk }], message)
+          this.acknowledgedAnswer += chunk
+          offset = end
+          break
+        } catch (error) {
+          /** Only an explicit size rejection proves this text was not appended. */
+          if (
+            !(error instanceof SlackDeliveryError) ||
+            error.method !== 'chat.appendStream' ||
+            error.outcome !== 'rejected' ||
+            error.code !== 'msg_too_long'
+          ) {
+            throw error
+          }
+          if (message.textLength > 0) {
+            message = await this.ensureStarted(true)
+          } else {
+            /** A rejected first append shrinks on each attempt, stopping at one code point. */
+            let smallerEnd = offset + Math.floor((end - offset) / 2)
+            const lastCodeUnit = text.charCodeAt(smallerEnd - 1)
+            if (lastCodeUnit >= 0xd800 && lastCodeUnit <= 0xdbff) smallerEnd--
+            if (smallerEnd <= offset) throw error
+            end = smallerEnd
+          }
+        }
+      }
     }
   }