fix(slack): defer task progress behind withheld text

Author: TheodoreSpeaks

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

@@ -69,6 +69,7 @@ function setup(deliverConnections = vi.fn().mockResolvedValue(undefined)) {
   } as unknown as ResolvedSecretTraceRegistry
   return {
     controller,
+    registry,
     beforeDelivery,
     beforeCleanup,
     stream: new SlackSearchAssistantStream({
@@ -111,6 +112,184 @@ function toolResult(
 }
 
 describe('Slack tool progress', () => {
+  it('preserves task positions when secret projection defers delivery until completion', async () => {
+    const { stream, registry } = setup()
+    vi.spyOn(registry, 'getActiveMatches').mockReturnValue([
+      { plaintext: 'private-token', replacement: '[REDACTED_SECRET]' },
+    ])
+    api.project.mockImplementation((value: unknown) => ({
+      safe: true,
+      value:
+        typeof value === 'string' ? value.replaceAll('private-token', '[REDACTED_SECRET]') : value,
+    }))
+    await stream.start()
+    await stream.onEvent({
+      type: 'text',
+      payload: { channel: 'assistant', text: 'Checking private-token.' },
+    })
+    await stream.onEvent(toolCall('search_workspace'))
+    await stream.onEvent(toolResult('search_workspace'))
+    await stream.onEvent({
+      type: 'text',
+      payload: { channel: 'assistant', text: 'Found a result.' },
+    })
+    expect(api.append).not.toHaveBeenCalled()
+    await stream.finish(result)
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(chunks).toEqual([
+      { type: 'markdown_text', text: 'Checking [REDACTED_SECRET].\n\n' },
+      {
+        type: 'task_update',
+        id: expect.any(String),
+        title: 'Searching documents…',
+        status: 'in_progress',
+      },
+      { type: 'task_update', id: chunks[1].id, title: 'Searching documents…', status: 'complete' },
+      { type: 'markdown_text', text: 'Found a result.' },
+    ])
+    expect(JSON.stringify(api.append.mock.calls)).not.toContain('private-token')
+  })
+
+  it('withholds tasks and following text until preceding citation evidence arrives', async () => {
+    const { stream } = setup()
+    await stream.start()
+    await stream.onEvent({
+      type: 'text',
+      payload: {
+        channel: 'assistant',
+        text: 'Checking <source>{"id":"late"}</source> for details.',
+      },
+    })
+    await stream.onEvent(toolCall('search_workspace'))
+    await stream.onEvent({
+      type: 'text',
+      payload: { channel: 'assistant', text: 'Found a result. ' },
+    })
+    expect(api.append.mock.calls.flatMap((call) => call[3])).toEqual([
+      { type: 'markdown_text', text: 'Checking ' },
+    ])
+    const completed = toolResult('search_workspace')
+    await stream.onEvent({
+      ...completed,
+      payload: {
+        ...completed.payload,
+        output: {
+          data: {
+            results: [
+              {
+                citationId: 'late',
+                citationUrl: 'https://example.com/policy',
+                documentName: 'Policy',
+              },
+            ],
+          },
+        },
+      },
+    })
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(chunks).toEqual([
+      { type: 'markdown_text', text: 'Checking ' },
+      { type: 'markdown_text', text: '[Policy](<https://example.com/policy>) for details.\n\n' },
+      {
+        type: 'task_update',
+        id: expect.any(String),
+        title: 'Searching documents…',
+        status: 'in_progress',
+      },
+      { type: 'task_update', id: chunks[2].id, title: 'Searching documents…', status: 'complete' },
+      { type: 'markdown_text', text: 'Found a result. ' },
+    ])
+    await stream.finish(result)
+    expect(api.append.mock.calls.flatMap((call) => call[3])).toEqual(chunks)
+  })
+
+  it('rejects a tool boundary whose prefix is unsafe in the complete secret projection', async () => {
+    const { stream, registry } = setup()
+    const secret = 'private-\n\ntoken'
+    vi.spyOn(registry, 'getActiveMatches').mockReturnValue([
+      { plaintext: secret, replacement: '[REDACTED_SECRET]' },
+    ])
+    api.project.mockImplementation((value: unknown) => ({
+      safe: true,
+      value: typeof value === 'string' ? value.replaceAll(secret, '[REDACTED_SECRET]') : value,
+    }))
+    await stream.start()
+    await stream.onEvent({ type: 'text', payload: { channel: 'assistant', text: 'private-' } })
+    await stream.onEvent(toolCall('search_workspace'))
+    await stream.onEvent({ type: 'text', payload: { channel: 'assistant', text: 'token' } })
+    await expect(stream.finish(result)).rejects.toThrow(
+      'The safe answer changed at a tool boundary'
+    )
+    expect(api.append).not.toHaveBeenCalled()
+  })
+
+  it('never retries a deferred task after its append fails ambiguously', async () => {
+    const { stream, registry, controller } = setup()
+    vi.spyOn(registry, 'getActiveMatches').mockReturnValue([
+      { plaintext: 'private-token', replacement: '[REDACTED_SECRET]' },
+    ])
+    await stream.start()
+    await stream.onEvent({ type: 'text', payload: { channel: 'assistant', text: 'Checking.' } })
+    await stream.onEvent(toolCall('search_workspace'))
+    expect(api.append).not.toHaveBeenCalled()
+    api.append.mockResolvedValueOnce(undefined).mockRejectedValueOnce(new Error('response lost'))
+    await expect(stream.finish(result)).rejects.toThrow('response lost')
+    expect(controller.signal.aborted).toBe(true)
+    await stream.terminateAfterFailure()
+    await stream.terminateAfterFailure()
+    expect(api.append).toHaveBeenCalledTimes(2)
+    expect(api.stop).toHaveBeenCalledOnce()
+    expect(api.stop.mock.calls[0][6]).toEqual([
+      { ...api.append.mock.calls[1][3][0], status: 'error' },
+    ])
+  })
+
+  it('omits unverified citations at completion without moving tasks ahead of their text', async () => {
+    const { stream } = setup()
+    await stream.start()
+    await stream.onEvent({
+      type: 'text',
+      payload: {
+        channel: 'assistant',
+        text: 'Checking <source>{"id":"missing"}</source> for details.',
+      },
+    })
+    await stream.onEvent(toolCall('search_workspace'))
+    await stream.onEvent(toolResult('search_workspace'))
+    await stream.onEvent({ type: 'text', payload: { channel: 'assistant', text: 'Done.' } })
+    await stream.finish(result)
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(chunks.map((chunk) => chunk.type)).toEqual([
+      'markdown_text',
+      'markdown_text',
+      'task_update',
+      'task_update',
+      'markdown_text',
+    ])
+    expect(chunks[1].text).toBe(' for details.\n\n')
+    expect(chunks[4].text).toBe('Done.')
+    expect(deliveredText()).not.toContain('missing')
+  })
+
+  it('does not introduce a withheld task when delivery is cancelled', async () => {
+    const { stream, controller } = setup()
+    await stream.start()
+    await stream.onEvent({
+      type: 'text',
+      payload: {
+        channel: 'assistant',
+        text: 'Checking <source>{"id":"missing"}</source> for details.',
+      },
+    })
+    await stream.onEvent(toolCall('search_workspace'))
+    controller.abort(new Error('stopped'))
+    await stream.terminateAfterFailure()
+    expect(api.stop.mock.calls[0][6]).toEqual([])
+    expect(api.append.mock.calls.flatMap((call) => call[3])).toEqual([
+      { type: 'markdown_text', text: 'Checking ' },
+    ])
+  })
+
   it('flushes a batched sentence before starting tool progress', async () => {
     vi.spyOn(Date, 'now').mockReturnValue(1000)
     try {

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

@@ -126,6 +126,8 @@ export class SlackSearchAssistantStream {
   private pendingEvents: Promise<void> = Promise.resolve()
   private evidence = new Map<string, Record<string, unknown>>()
   private toolProgress = new Map<string, { toolName: string; chunk: ToolProgress }>()
+  private pendingProgress: { textEnd: number; chunk: ToolProgress }[] = []
+  private deliveredProgress = new Map<string, ToolProgress>()
   constructor(private readonly options: AssistantStreamOptions) {}
 
   private async deliver(action: () => Promise<void>) {
@@ -184,17 +186,18 @@ export class SlackSearchAssistantStream {
         'phase' in event.payload &&
         (event.payload.phase === 'call' || event.payload.phase === 'result')
       ) {
-        await this.updateToolProgress(event.payload)
+        this.queueToolProgress(event.payload)
       }
     }
+    if (event.type === 'tool' && this.pendingProgress.length) await this.flush(false)
     if (event.type !== 'text' || event.payload.channel !== 'assistant' || event.scope) return
     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)
   }
 
   /** Only static labels reach Slack; arguments, account details, and backend errors stay private. */
-  private async updateToolProgress(
+  private queueToolProgress(
     payload: ToolCallStreamEvent['payload'] | ToolResultStreamEvent['payload']
   ) {
     const title = TOOL_PROGRESS_TITLES.get(payload.toolName)
@@ -212,9 +215,7 @@ export class SlackSearchAssistantStream {
         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 {
       if (!existing || existing.chunk.status !== 'in_progress') return
       if (existing.toolName !== payload.toolName)
@@ -227,24 +228,16 @@ export class SlackSearchAssistantStream {
             : 'error',
       }
     }
-    await this.deliver(async () => {
-      if (!this.stream || this.closed) throw new Error('Slack stream is not active')
-      await appendSlackAgentStream(
-        this.options.token,
-        this.stream.channel,
-        this.stream.ts,
-        [chunk],
-        this.options.controller.signal
-      )
-    })
     this.toolProgress.set(payload.toolCallId, { toolName: payload.toolName, chunk })
+    /** Updating an existing task does not introduce a new position in Slack's timeline. */
+    this.pendingProgress.push({ textEnd: payload.phase === 'call' ? this.text.length : 0, chunk })
   }
 
   /** Finalize interrupted tasks in the single stop request, including ambiguous progress sends. */
   private interruptedToolProgress(): ToolProgress[] {
-    return [...this.toolProgress.values()]
-      .filter(({ chunk }) => chunk.status === 'in_progress')
-      .map(({ chunk }) => ({ ...chunk, status: 'error' }))
+    return [...this.deliveredProgress.values()]
+      .filter((chunk) => chunk.status === 'in_progress')
+      .map((chunk) => ({ ...chunk, status: 'error' }))
   }
 
   private collectSources(blocks: readonly RetrievalCitationBlock[]) {
@@ -254,13 +247,10 @@ export class SlackSearchAssistantStream {
   }
 
   private async flush(complete: boolean) {
-    const { registry, token, controller } = this.options
+    const { registry } = this.options
     if (!registry.isComplete()) throw new Error('Answer secret provenance is unavailable')
     /** Active secret literals can straddle deltas; project their complete answer instead. */
     if (!complete && registry.getActiveMatches().length) return
-    const projection = projectResolvedSecretDiagnosticContent(this.text, registry, 512_000)
-    if (!projection.safe || typeof projection.value !== 'string')
-      throw new Error('Answer could not be safely projected')
     const sources = new Map<string, string>()
     for (const [id, source] of this.evidence) {
       const projected = projectResolvedSecretDiagnosticContent(source, registry)
@@ -271,8 +261,46 @@ export class SlackSearchAssistantStream {
           : ''
       )
     }
-    const text = publicSlackAnswer(redactSensitiveContent(projection.value), complete, sources)
+    const text = this.projectAnswer(this.text, complete, sources)
+    if (!text.startsWith(this.sent)) throw new Error('The safe answer changed after delivery')
+    while (this.pendingProgress.length) {
+      const { textEnd, chunk } = this.pendingProgress[0]!
+      const preceding = this.text.slice(0, textEnd)
+      const prefix = this.projectAnswer(preceding, complete, sources)
+      /** A prefix must remain safe when projected as part of the complete answer. */
+      if (!text.startsWith(prefix)) throw new Error('The safe answer changed at a tool boundary')
+      await this.appendText(prefix)
+      /** Unresolved citations and partial markup must not let a task overtake withheld text. */
+      if (!complete && prefix !== this.projectAnswer(preceding, true, sources)) return
+      await this.deliver(async () => {
+        if (!this.stream || this.closed) throw new Error('Slack stream is not active')
+        /** Include an ambiguously started task in failure cleanup, but never an unsent task. */
+        if (!this.deliveredProgress.has(chunk.id)) this.deliveredProgress.set(chunk.id, chunk)
+        await appendSlackAgentStream(
+          this.options.token,
+          this.stream.channel,
+          this.stream.ts,
+          [chunk],
+          this.options.controller.signal
+        )
+      })
+      this.deliveredProgress.set(chunk.id, chunk)
+      this.pendingProgress.shift()
+    }
+    await this.appendText(text)
+  }
+
+  private projectAnswer(text: string, complete: boolean, sources: ReadonlyMap<string, string>) {
+    const projection = projectResolvedSecretDiagnosticContent(text, this.options.registry, 512_000)
+    if (!projection.safe || typeof projection.value !== 'string')
+      throw new Error('Answer could not be safely projected')
+    return publicSlackAnswer(redactSensitiveContent(projection.value), complete, sources)
+  }
+
+  private async appendText(text: string) {
+    if (this.sent.startsWith(text)) return
     if (!text.startsWith(this.sent)) throw new Error('The safe answer changed after delivery')
+    const { token, controller } = this.options
     let pending = text.slice(this.sent.length)
     while (pending.length) {
       let end = Math.min(4000, pending.length)