fix(slack): preserve text order around tool progress

Author: TheodoreSpeaks

apps/sim/lib/slack-search/assistant-stream.test.ts

@@ -111,6 +111,101 @@ function toolResult(
 }
 
 describe('Slack tool progress', () => {
+  it('flushes a batched sentence before starting tool progress', async () => {
+    vi.spyOn(Date, 'now').mockReturnValue(1000)
+    try {
+      const { stream } = setup()
+      await stream.start()
+      await stream.onEvent({
+        type: 'text',
+        payload: { channel: 'assistant', text: "I'll search " },
+      })
+      await stream.onEvent({
+        type: 'text',
+        payload: { channel: 'assistant', text: 'the connected sources for the handbook.' },
+      })
+      await stream.onEvent(toolCall('search_workspace'))
+      const chunks = api.append.mock.calls.flatMap((call) => call[3])
+      expect(chunks).toEqual([
+        { type: 'markdown_text', text: "I'll search " },
+        {
+          type: 'markdown_text',
+          text: 'the connected sources for the handbook.\n\n',
+        },
+        {
+          type: 'task_update',
+          id: expect.any(String),
+          title: 'Searching documents…',
+          status: 'in_progress',
+        },
+      ])
+    } finally {
+      vi.restoreAllMocks()
+    }
+  })
+
+  it('keeps text contiguous across preparatory and hidden tool events', async () => {
+    const { stream } = setup()
+    await stream.start()
+    await stream.onEvent({
+      type: 'text',
+      payload: { channel: 'assistant', text: "I'll search" },
+    })
+    for (const attributes of [
+      { partial: true },
+      { status: 'generating' as const },
+      { ui: { hidden: true } },
+      { ui: { internal: true } },
+    ]) {
+      const event = toolCall('search_workspace')
+      await stream.onEvent({ ...event, payload: { ...event.payload, ...attributes } })
+    }
+    await stream.onEvent({
+      type: 'text',
+      payload: { channel: 'assistant', text: ' the connected sources.' },
+    })
+    await stream.finish(result)
+    expect(deliveredText()).toBe("I'll search the connected sources.")
+    expect(
+      api.append.mock.calls
+        .flatMap((call) => call[3])
+        .every((chunk) => chunk.type === 'markdown_text')
+    ).toBe(true)
+  })
+
+  it('serializes concurrent text and tool events without duplicating buffered text', async () => {
+    const { stream } = setup()
+    await stream.start()
+    let releaseAppend!: () => void
+    api.append.mockImplementationOnce(
+      () =>
+        new Promise<void>((resolve) => {
+          releaseAppend = resolve
+        })
+    )
+    const text = stream.onEvent({
+      type: 'text',
+      payload: { channel: 'assistant', text: "I'll search the connected sources. " },
+    })
+    await vi.waitFor(() => expect(api.append).toHaveBeenCalledOnce(), { interval: 1 })
+    const call = stream.onEvent(toolCall('search_workspace'))
+    const completed = stream.onEvent(toolResult('search_workspace'))
+    const finished = stream.finish(result)
+    expect(api.append).toHaveBeenCalledOnce()
+    expect(api.stop).not.toHaveBeenCalled()
+    releaseAppend()
+    await Promise.all([text, call, completed, finished])
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(deliveredText()).toBe("I'll search the connected sources. \n\n")
+    expect(chunks.map((chunk) => chunk.type)).toEqual([
+      'markdown_text',
+      'markdown_text',
+      'task_update',
+      'task_update',
+    ])
+    expect(chunks[3]).toEqual({ ...chunks[2], status: 'complete' })
+  })
+
   it.each([
     ['list_integrations', 'Listing connected integrations…'],
     ['search_workspace', 'Searching documents…'],

apps/sim/lib/slack-search/assistant-stream.ts

@@ -123,7 +123,7 @@ export class SlackSearchAssistantStream {
   private failure?: Error
   private closed = false
   private closeAttempted = false
-  private separateNextText = false
+  private pendingEvents: Promise<void> = Promise.resolve()
   private evidence = new Map<string, Record<string, unknown>>()
   private toolProgress = new Map<string, { toolName: string; chunk: ToolProgress }>()
   constructor(private readonly options: AssistantStreamOptions) {}
@@ -160,7 +160,12 @@ export class SlackSearchAssistantStream {
     })
   }
 
-  async onEvent(event: StreamEvent) {
+  onEvent(event: StreamEvent): Promise<void> {
+    this.pendingEvents = this.pendingEvents.then(() => this.handleEvent(event))
+    return this.pendingEvents
+  }
+
+  private async handleEvent(event: StreamEvent) {
     if (this.failure) throw this.failure
     if (event.type === 'tool' && 'phase' in event.payload && event.payload.phase === 'result') {
       const { toolName, success, status, output } = event.payload
@@ -175,7 +180,6 @@ export class SlackSearchAssistantStream {
       ])
     }
     if (event.type === 'tool' && !event.scope) {
-      this.separateNextText = true
       if (
         'phase' in event.payload &&
         (event.payload.phase === 'call' || event.payload.phase === 'result')
@@ -184,8 +188,6 @@ export class SlackSearchAssistantStream {
       }
     }
     if (event.type !== 'text' || event.payload.channel !== 'assistant' || event.scope) return
-    if (this.separateNextText && this.text) this.text += '\n\n'
-    this.separateNextText = false
     this.text += event.payload.text
     if (this.text.length > 128_000) throw new Error('Slack answer exceeds the supported size')
     if (Date.now() - this.lastSentAt >= 750) await this.flush(false)
@@ -208,6 +210,9 @@ export class SlackSearchAssistantStream {
         (payload.status !== undefined && payload.status !== 'executing')
       )
         return
+      /** Close the preceding text segment so batching cannot place its tail after the task. */
+      if (this.text && !this.text.endsWith('\n\n')) this.text += '\n\n'
+      await this.flush(false)
       chunk = { type: 'task_update', id: generateId(), title, status: 'in_progress' }
       this.toolProgress.set(payload.toolCallId, { toolName: payload.toolName, chunk })
     } else {
@@ -298,6 +303,7 @@ export class SlackSearchAssistantStream {
   }
 
   async finish(result: OrchestratorResult) {
+    await this.pendingEvents
     if (this.failure) throw this.failure
     this.collectSources(result.contentBlocks)
     const projection = projectResolvedSecretDiagnosticContent(