fix(slack): deliver Sim Chat text and tool progress reliably

Author: Sg312

apps/docs/content/docs/integrations/slack.mdx

@@ -1970,7 +1970,7 @@ Trigger from Slack events, interactions, and slash commands
 | `streamTaskTitle` | string | No | Optional status Slack shows while each selected response is being produced. Leave empty to use Running. |
 | `streamTaskDisplayMode` | string | No | Choose how Slack displays thinking and tool progress. |
 | `streamIncludeThinking` | boolean | No | Show agent thinking as Slack task updates while the response is generated. |
-| `streamIncludeToolCalls` | boolean | No | Show tool execution lifecycle as Slack task updates. |
+| `streamIncludeToolCalls` | boolean | No | Show Agent and Sim Chat tool execution progress as Slack task updates. |
 | `emoji` | string | No | Comma-separated emoji names to restrict to. Leave empty to match any emoji. |
 | `nameContains` | string | No | Only fire when the created channel name contains this text. |
 | `interactionFilter` | string | No | Comma-separated action_ids \(buttons/selects\) or callback_ids \(modals\) to restrict to. Leave empty to fire on any interaction. |

apps/sim/app/api/mothership/execute/route.test.ts

@@ -391,19 +391,18 @@ describe('mothership private trace provenance transport', () => {
       .map((line) => JSON.parse(line))
     expect(response.status).toBe(200)
     expect(events.filter((event) => event.type === 'error')).toEqual([])
-    expect(events.filter((event) => event.type === 'agent_event')).toEqual([
-      { type: 'agent_event', event: { type: 'thinking_delta', text: 'Considering' } },
-      { type: 'agent_event', event: { type: 'turn_end', turn: 'intermediate' } },
-      { type: 'agent_event', event: { type: 'tool_call_start', id: 'tool-1', name: 'Lookup' } },
-      {
-        type: 'agent_event',
-        event: { type: 'tool_call_end', id: 'tool-1', name: 'Lookup', status: 'success' },
-      },
-      { type: 'agent_event', event: { type: 'turn_end', turn: 'final' } },
-    ])
+    expect(events.filter((event) => event.type === 'agent_event')).toEqual(
+      [
+        { type: 'thinking_delta', text: 'Considering' },
+        { type: 'turn_end', turn: 'intermediate' },
+        { type: 'tool_call_start', id: 'tool-1', name: 'Lookup' },
+        { type: 'tool_call_end', id: 'tool-1', name: 'Lookup', status: 'success' },
+        { type: 'turn_end', turn: 'final' },
+      ].map((event) => ({ type: 'agent_event', v: 1, event }))
+    )
     expect(events.filter((event) => event.type === 'chunk')).toEqual([
-      { type: 'chunk', content: 'I will check.', turn: 'pending' },
-      { type: 'chunk', content: 'Answer', turn: 'pending' },
+      { type: 'chunk', v: 1, content: 'I will check.', turn: 'pending' },
+      { type: 'chunk', v: 1, content: 'Answer', turn: 'pending' },
     ])
     expect(text).not.toContain('private-args')
     expect(text).not.toContain('private-result')

apps/sim/app/api/mothership/execute/route.ts

@@ -3,7 +3,10 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { type NextRequest, NextResponse } from 'next/server'
-import { mothershipExecuteContract } from '@/lib/api/contracts/mothership-chats'
+import {
+  type MothershipExecuteStreamEvent,
+  mothershipExecuteContract,
+} from '@/lib/api/contracts/mothership-chats'
 import { parseRequest } from '@/lib/api/server'
 import { checkInternalAuth } from '@/lib/auth/hybrid'
 import { verifyInternalDelegationToken } from '@/lib/auth/internal'
@@ -27,11 +30,8 @@ import {
   createCopilotEnvironmentContext,
 } from '@/lib/mothership/environment-context'
 import { IntegrationCatalogMcpExecution } from '@/lib/mothership/generated/integration-catalog'
-import {
-  MothershipStreamV1EventType,
-  MothershipStreamV1TextChannel,
-} from '@/lib/mothership/generated/mothership-stream-v1'
 import { PROTOCOL_VERSION } from '@/lib/mothership/generated/protocol'
+import { ExecuteEventProjection } from '@/lib/mothership/request/lifecycle/execute-events'
 import { runHeadlessCopilotLifecycle } from '@/lib/mothership/request/lifecycle/headless'
 import { requestExplicitStreamAbort } from '@/lib/mothership/request/session/explicit-abort'
 import type { StreamEvent } from '@/lib/mothership/request/types'
@@ -391,15 +391,19 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
 
       const stream = new ReadableStream<Uint8Array>({
         start(controller) {
-          let lastForwardedSeq = -1
-          const startedTools = new Set<string>()
-          const endedTools = new Set<string>()
-          const send = (event: unknown) => {
+          const send = (event: MothershipExecuteStreamEvent) => {
             if (!cancelled) {
               controller.enqueue(encodeNdjson(event))
             }
           }
 
+          const projection = new ExecuteEventProjection((event) => {
+            /** Keep text frames readable by workers deployed before agent-event versioning. */
+            if (event.type === 'text_delta')
+              send({ type: 'chunk', v: 1, content: event.text, turn: event.turn })
+            else send({ type: 'agent_event', v: 1, event })
+          })
+
           // Flush response headers promptly and keep long headless runs from
           // looking idle to worker/proxy HTTP stacks.
           send({ type: 'heartbeat', timestamp: new Date().toISOString() })
@@ -409,59 +413,11 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
 
           void (async () => {
             try {
-              const result = await runLifecycle(async (event) => {
-                /** Reconnect replays share the same monotone sequence across event types. */
-                if (typeof event.seq === 'number') {
-                  if (event.seq <= lastForwardedSeq) return
-                  lastForwardedSeq = event.seq
-                }
-                if (event.type === MothershipStreamV1EventType.text && event.payload.text) {
-                  if (event.payload.channel === MothershipStreamV1TextChannel.assistant) {
-                    if (event.scope?.lane === 'subagent') return
-                    send({ type: 'chunk', content: event.payload.text, turn: 'pending' })
-                  } else if (event.payload.channel === MothershipStreamV1TextChannel.thinking) {
-                    send({
-                      type: 'agent_event',
-                      event: { type: 'thinking_delta', text: event.payload.text },
-                    })
-                  }
-                } else if (event.type === MothershipStreamV1EventType.tool) {
-                  const tool = event.payload
-                  if (!('phase' in tool)) return
-                  if (tool.phase === 'call' && !startedTools.has(tool.toolCallId)) {
-                    startedTools.add(tool.toolCallId)
-                    if (event.scope?.lane !== 'subagent' && !tool.replay) {
-                      send({
-                        type: 'agent_event',
-                        event: { type: 'turn_end', turn: 'intermediate' },
-                      })
-                    }
-                    send({
-                      type: 'agent_event',
-                      event: { type: 'tool_call_start', id: tool.toolCallId, name: tool.toolName },
-                    })
-                  } else if (tool.phase === 'result' && !endedTools.has(tool.toolCallId)) {
-                    endedTools.add(tool.toolCallId)
-                    send({
-                      type: 'agent_event',
-                      event: {
-                        type: 'tool_call_end',
-                        id: tool.toolCallId,
-                        name: tool.toolName,
-                        status:
-                          tool.status === 'cancelled'
-                            ? 'cancelled'
-                            : tool.success
-                              ? 'success'
-                              : 'error',
-                      },
-                    })
-                  }
-                }
-              })
+              const result = await runLifecycle(async (event) => projection.accept(event))
               allowExplicitAbort = false
 
               if (lifecycleAbortController.signal.aborted) {
+                projection.finish('cancelled')
                 send(
                   withPrivateProvenance(
                     { type: 'error', error: 'Sim execution aborted' },
@@ -473,6 +429,7 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
               }
 
               if (!result.success) {
+                projection.finish('error')
                 logger.error(
                   messageId
                     ? `Mothership execute failed [messageId:${messageId}]`
@@ -499,7 +456,7 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
                 return
               }
 
-              send({ type: 'agent_event', event: { type: 'turn_end', turn: 'final' } })
+              projection.finish('success')
               send({
                 type: 'final',
                 data: withPrivateProvenance(
@@ -509,6 +466,7 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
                 ),
               })
             } catch (error) {
+              projection.finish(lifecycleAbortController.signal.aborted ? 'cancelled' : 'error')
               if (
                 lifecycleAbortController.signal.aborted ||
                 req.signal.aborted ||

apps/sim/background/webhook-execution.test.ts

@@ -75,6 +75,13 @@ vi.mock('@/lib/webhooks/env-resolver', () => ({
   resolveWebhookRecordProviderConfig: mockResolveWebhookRecordProviderConfig,
 }))
 
+const { mockCreateSlackStreamController } = vi.hoisted(() => ({
+  mockCreateSlackStreamController: vi.fn(),
+}))
+vi.mock('@/lib/webhooks/slack-execution-stream', () => ({
+  SlackExecutionStreamController: { create: mockCreateSlackStreamController },
+}))
+
 vi.mock('@/lib/workflows/executor/execution-core', () => ({
   executeWorkflowCore: mockExecuteWorkflowCore,
   wasExecutionFinalizedByCore: mockWasExecutionFinalizedByCore,
@@ -324,6 +331,59 @@ describe('executeWebhookJob fault vs error handling', () => {
     dbChainMockFns.limit.mockResolvedValue([{ id: 'webhook-1' }])
   })
 
+  it.each([true, false])(
+    'enables the shared Agent/Sim Chat event stream and honors tool visibility: %s',
+    async (includeToolCalls) => {
+      const config = {
+        enabled: true,
+        outputConfigs: [{ blockId: 'mship', path: 'content' }],
+        includeToolCalls,
+        includeThinking: false,
+        taskDisplayMode: 'timeline',
+        taskTitle: 'Running',
+      }
+      dbChainMockFns.limit.mockResolvedValue([
+        {
+          id: 'webhook-1',
+          providerConfig: { credentialId: 'bot-1', streamResponseConfig: config },
+        },
+      ])
+      const calls: string[] = []
+      const controller = {
+        selectedOutputs: ['mship_content'],
+        callbacks: { onStream: vi.fn() },
+        finalize: vi.fn(async () => {
+          calls.push('delivery')
+        }),
+        assertSucceeded: vi.fn(() => {
+          calls.push('receipt')
+        }),
+      }
+      mockCreateSlackStreamController.mockResolvedValue(controller)
+      mockExecuteWorkflowCore.mockImplementationOnce(async (options) => {
+        const result = {
+          success: true,
+          status: 'completed',
+          output: { content: 'Complete answer' },
+          logs: [],
+        }
+        expect(options.callbacks).toBe(controller.callbacks)
+        await options.finalizeDelivery(result)
+        calls.push('workflow-result')
+        return result
+      })
+      await executeWebhookJob({ ...legacyPayload, provider: 'slack' })
+      expect(mockExecutionSnapshot.mock.calls[0]?.[0]).toMatchObject({
+        agentEvents: true,
+        includeThinking: false,
+        includeToolCalls,
+      })
+      expect(mockExecutionSnapshot.mock.calls[0]?.[4]).toEqual(['mship_content'])
+      expect(controller.finalize).toHaveBeenCalledTimes(1)
+      expect(calls).toEqual(['delivery', 'receipt', 'workflow-result'])
+    }
+  )
+
   it('restores a legacy queued webhook as its canonical system principal', async () => {
     mockExecuteWorkflowCore.mockResolvedValueOnce({
       success: true,

apps/sim/background/webhook-execution.ts

@@ -1120,6 +1120,12 @@ async function executeWebhookJobInternal(
           })
         : null
 
+      if (slackStreamConfig) {
+        metadata.agentEvents = true
+        metadata.includeThinking = slackStreamConfig.includeThinking
+        metadata.includeToolCalls = slackStreamConfig.includeToolCalls
+      }
+
       const snapshot = new ExecutionSnapshot(
         metadata,
         workflowRecord,
@@ -1134,6 +1140,14 @@ async function executeWebhookJobInternal(
         executionResult = await executeWorkflowCore({
           snapshot,
           callbacks: slackStreamController?.callbacks ?? {},
+          ...(slackStreamController
+            ? {
+                finalizeDelivery: async (result: ExecutionResult) => {
+                  await slackStreamController.finalize(result)
+                  slackStreamController.assertSucceeded()
+                },
+              }
+            : {}),
           loggingSession,
           trustedInitialResolvedSecretTraceProvenance:
             resolvedSecretTraceRegistry.exportProvenanceForValue(triggerInput),
@@ -1151,11 +1165,6 @@ async function executeWebhookJobInternal(
         }
         throw error
       }
-      if (slackStreamController) {
-        await slackStreamController.finalize(executionResult)
-        slackStreamController.assertSucceeded()
-      }
-
       await handleExecutionResult(executionResult, {
         loggingSession,
         timeoutController,

apps/sim/executor/handlers/mothership/mothership-handler.test.ts

@@ -1947,14 +1947,22 @@ describe('MothershipBlockHandler', () => {
       context.metadata.agentEvents = agentEvents
       fetchMock.mockResolvedValue(
         createNdjsonResponse([
-          { type: 'chunk', content: 'I will check.', turn: 'pending' },
+          {
+            type: 'agent_event',
+            v: 1,
+            event: { type: 'text_delta', text: 'I will check.', turn: 'pending' },
+          },
           { type: 'agent_event', event: { type: 'turn_end', turn: 'intermediate' } },
           { type: 'agent_event', event: { type: 'tool_call_start', id: 'lookup', name: 'Lookup' } },
           {
             type: 'agent_event',
             event: { type: 'tool_call_end', id: 'lookup', name: 'Lookup', status: 'success' },
           },
-          { type: 'chunk', content: 'Final answer.', turn: 'pending' },
+          {
+            type: 'agent_event',
+            v: 1,
+            event: { type: 'text_delta', text: 'Final answer.', turn: 'pending' },
+          },
           { type: 'agent_event', event: { type: 'turn_end', turn: 'final' } },
           { type: 'final', data: { content: 'Final answer.' } },
         ])

apps/sim/executor/handlers/mothership/mothership-handler.ts

@@ -1,6 +1,10 @@
 import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
 import { isPlainRecord } from '@sim/utils/object'
+import type {
+  MothershipExecuteResult,
+  MothershipExecuteStreamEvent,
+} from '@/lib/api/contracts/mothership-chats'
 import { generateInternalDelegationToken } from '@/lib/auth/internal'
 import {
   BILLING_ATTRIBUTION_HEADER,
@@ -53,11 +57,7 @@ import type {
   ResolvedSecretInputPath,
   ResolvedSecretTraceRegistry,
 } from '@/executor/utils/resolved-secret-trace-registry'
-import {
-  type AgentStreamEvent,
-  isAgentStreamEvent,
-  type TextDeltaClassification,
-} from '@/providers/stream-events'
+import { type AgentStreamEvent, isAgentStreamEvent } from '@/providers/stream-events'
 import type { SerializedBlock } from '@/serializer/types'
 
 const logger = createLogger('MothershipBlockHandler')
@@ -101,24 +101,6 @@ interface IndexedMothershipSkillContext {
   hasExplicitLabel: boolean
 }
 
-type MothershipExecuteResult = {
-  content?: string
-  model?: string
-  conversationId?: string
-  tokens?: Record<string, unknown>
-  toolCalls?: Array<Record<string, unknown>>
-  cost?: unknown
-} & Partial<Record<typeof RESOLVED_SECRET_PROVENANCE_FIELD, unknown>>
-
-type MothershipExecuteStreamEvent =
-  | { type: 'heartbeat'; timestamp?: string }
-  | { type: 'chunk'; content?: string; turn?: TextDeltaClassification }
-  | { type: 'agent_event'; event: AgentStreamEvent }
-  | { type: 'final'; data: MothershipExecuteResult }
-  | ({ type: 'error'; error?: string } & Partial<
-      Record<typeof RESOLVED_SECRET_PROVENANCE_FIELD, unknown>
-    >)
-
 function selectIndexedMothershipMcpTools(tools: unknown): IndexedMothershipMcpToolSelection[] {
   if (!Array.isArray(tools)) return []
 
@@ -496,7 +478,7 @@ function formatMothershipBlockOutput(
   result: MothershipExecuteResult,
   conversationId: string
 ): NormalizedBlockOutput {
-  const formattedList = (result.toolCalls || []).map((tc: Record<string, unknown>) => ({
+  const formattedList = (result.toolCalls || []).map((tc) => ({
     name: typeof tc.name === 'string' ? tc.name : String(tc.name ?? ''),
     ...(typeof tc.status === 'string' ? { status: tc.status } : {}),
     arguments: (tc.arguments || tc.params || tc.input || {}) as Record<string, unknown>,
@@ -676,11 +658,15 @@ function createMothershipStreamingExecution(
         }
 
         if (event.type === 'agent_event') {
-          if (!isAgentStreamEvent(event.event)) {
+          if ((event.v !== undefined && event.v !== 1) || !isAgentStreamEvent(event.event)) {
             throw new Error('Sim execution stream returned an invalid agent event')
           }
           if (options.agentEvents) controller.enqueue(event.event)
-          else if (event.event.type === 'turn_end') {
+          else if (event.event.type === 'text_delta') {
+            if (event.event.turn === 'pending') pendingText += event.event.text
+            else if (event.event.turn !== 'intermediate')
+              controller.enqueue(encoder.encode(event.event.text))
+          } else if (event.event.type === 'turn_end') {
             if (event.event.turn === 'final' && pendingText)
               controller.enqueue(encoder.encode(pendingText))
             pendingText = ''