From 4a4ca0bfe6fd5251ffb13a2b15903756fe665f78 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Wed, 9 Sep 2026 22:12:53 -0700 Subject: [PATCH 1/2] fix(slack): preserve text order around tool progress --- .../lib/slack-search/assistant-stream.test.ts | 95 +++++++++++++++++++ apps/sim/lib/slack-search/assistant-stream.ts | 16 +++- 2 files changed, 106 insertions(+), 5 deletions(-) diff --git a/apps/sim/lib/slack-search/assistant-stream.test.ts b/apps/sim/lib/slack-search/assistant-stream.test.ts index 4fa5a9b4753..c25576fd017 100644 --- a/apps/sim/lib/slack-search/assistant-stream.test.ts +++ b/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((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…'], diff --git a/apps/sim/lib/slack-search/assistant-stream.ts b/apps/sim/lib/slack-search/assistant-stream.ts index 201e1270f99..1ff9e580abf 100644 --- a/apps/sim/lib/slack-search/assistant-stream.ts +++ b/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 = Promise.resolve() private evidence = new Map>() private toolProgress = new Map() constructor(private readonly options: AssistantStreamOptions) {} @@ -160,7 +160,12 @@ export class SlackSearchAssistantStream { }) } - async onEvent(event: StreamEvent) { + onEvent(event: StreamEvent): Promise { + 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( From 950cda585a97978332c7ae5a6971c97486a9c5e7 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Wed, 9 Sep 2026 22:28:18 -0700 Subject: [PATCH 2/2] fix(slack): defer task progress behind withheld text --- .../lib/slack-search/assistant-stream.test.ts | 179 ++++++++++++++++++ apps/sim/lib/slack-search/assistant-stream.ts | 72 ++++--- 2 files changed, 229 insertions(+), 22 deletions(-) diff --git a/apps/sim/lib/slack-search/assistant-stream.test.ts b/apps/sim/lib/slack-search/assistant-stream.test.ts index c25576fd017..f2cdfef06e3 100644 --- a/apps/sim/lib/slack-search/assistant-stream.test.ts +++ b/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 {"id":"late"} 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]() 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 {"id":"missing"} 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 {"id":"missing"} 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 { diff --git a/apps/sim/lib/slack-search/assistant-stream.ts b/apps/sim/lib/slack-search/assistant-stream.ts index 1ff9e580abf..07767a48d2d 100644 --- a/apps/sim/lib/slack-search/assistant-stream.ts +++ b/apps/sim/lib/slack-search/assistant-stream.ts @@ -126,6 +126,8 @@ export class SlackSearchAssistantStream { private pendingEvents: Promise = Promise.resolve() private evidence = new Map>() private toolProgress = new Map() + private pendingProgress: { textEnd: number; chunk: ToolProgress }[] = [] + private deliveredProgress = new Map() constructor(private readonly options: AssistantStreamOptions) {} private async deliver(action: () => Promise) { @@ -184,9 +186,10 @@ 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') @@ -194,7 +197,7 @@ export class SlackSearchAssistantStream { } /** 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() 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) { + 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)