fix(workflows): keep long-running calls active (#7579)

* fix(workflows): keep long-running calls active

* chore(helm): bump chart version

* fix(workflows): preserve stream compatibility

* refactor(streaming): import SSE helpers directly

Author: waleedlatif1

apps/docs/openapi-v2-workflows.json

@@ -2502,7 +2502,7 @@
       "post": {
         "operationId": "executeWorkflowV2",
         "summary": "Execute Workflow",
-        "description": "Execute the deployment; `run.source: \"manual\"` uses draft state. Manual runs require a personal key or OAuth write access; workspace keys, anonymous callers, and async are rejected. Start at a runnable trigger, or resume from `sourceRunId` using the same-workflow snapshot. Public deployments allow anonymous sync or streaming; async requires credentials. Sync timeouts return `200` with failed status and `TIMEOUT`. Each option carries the modes it requires and the modes that reject it; a violated combination is a 400.\n\nOAuth scope: `api:write`.",
+        "description": "Execute a deployment or use `run.source: \"manual\"` for draft state. Manual runs require personal or OAuth write access and reject workspace keys, anonymous callers, and async. Start at a trigger or resume from same-workflow `sourceRunId`. Public deployments allow anonymous sync or streaming. Request `application/x-ndjson` for 15-second heartbeats and final resource. Timeouts return `200` with failed status and `TIMEOUT`. Each option carries the modes it requires and the modes that reject it; a violated combination is a 400.\n\nOAuth scope: `api:write`.",
         "x-sim-operation": "workflows.execute",
         "x-oauth-scope": "api:write",
         "tags": ["Workflows"],
@@ -2566,7 +2566,7 @@
         },
         "responses": {
           "200": {
-            "description": "A synchronous run result or Server-Sent Event stream.",
+            "description": "A synchronous run result, heartbeat-delimited NDJSON result stream, or Server-Sent Event stream.",
             "headers": {
               "X-Run-Id": {
                 "$ref": "#/components/headers/X-Run-Id"
@@ -2587,6 +2587,11 @@
                   "$ref": "#/components/schemas/ExecuteWorkflowSyncResponse"
                 }
               },
+              "application/x-ndjson": {
+                "schema": {
+                  "type": "string"
+                }
+              },
               "text/event-stream": {
                 "schema": {
                   "type": "string"

apps/sim/app/api/copilot/chat/stream/route.test.ts

@@ -38,7 +38,6 @@ vi.mock('@/lib/copilot/request/session', () => ({
   }),
   encodeSSEEnvelope: (event: Record<string, unknown>) =>
     new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`),
-  encodeSSEComment: (comment: string) => new TextEncoder().encode(`: ${comment}\n\n`),
   SSE_RESPONSE_HEADERS: {
     'Content-Type': 'text/event-stream',
   },

apps/sim/app/api/copilot/chat/stream/route.ts

@@ -22,13 +22,13 @@ import { getCopilotTracer, markSpanForError } from '@/lib/copilot/request/otel'
 import {
   checkForReplayGap,
   createEvent,
-  encodeSSEComment,
   encodeSSEEnvelope,
   readEvents,
   readFilePreviewSessions,
   SSE_RESPONSE_HEADERS,
 } from '@/lib/copilot/request/session'
 import { toStreamBatchEvent } from '@/lib/copilot/request/session/types'
+import { encodeSSEComment } from '@/lib/core/utils/sse'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 
 export const maxDuration = 3600

apps/sim/app/api/mcp/serve/[serverId]/route.test.ts

@@ -95,7 +95,7 @@ vi.mock('@/lib/auth/internal', () => ({
 }))
 
 vi.mock('@/lib/core/execution-limits', () => ({
-  getMaxExecutionTimeout: () => 10_000,
+  getMaxExecutionTimeout: () => 60_000,
 }))
 
 vi.mock('@/lib/workflows/executor/execute-service', () => ({
@@ -309,7 +309,10 @@ describe('MCP Serve Route', () => {
 
     const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
       method: 'POST',
-      headers: { 'X-API-Key': 'pk_test_123' },
+      headers: {
+        'X-API-Key': 'pk_test_123',
+        Accept: 'application/json, text/event-stream;q=0',
+      },
       body: JSON.stringify({
         jsonrpc: '2.0',
         id: 1,
@@ -320,6 +323,7 @@ describe('MCP Serve Route', () => {
     const response = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
 
     expect(response.status).toBe(200)
+    expect(response.headers.get('content-type')).toContain('application/json')
     expect(mockExecuteWorkflowService).toHaveBeenCalledTimes(1)
     expect(mockExecuteWorkflowService).toHaveBeenCalledWith(
       expect.objectContaining({
@@ -338,6 +342,153 @@ describe('MCP Serve Route', () => {
     })
   })
 
+  it('keeps a Streamable HTTP tool call active and ends with its JSON-RPC response', async () => {
+    vi.useFakeTimers()
+    try {
+      dbChainMockFns.limit
+        .mockResolvedValueOnce([
+          {
+            id: 'server-1',
+            name: 'Public Server',
+            workspaceId: 'ws-1',
+            isPublic: true,
+            createdBy: 'owner-1',
+          },
+        ])
+        .mockResolvedValueOnce([{ toolName: 'tool_a', workflowId: 'wf-1' }])
+        .mockResolvedValueOnce([{ workspaceId: 'ws-1', deploymentVersionId: 'deployment-1' }])
+
+      let finishExecution!: (result: unknown) => void
+      mockExecuteWorkflowService.mockReturnValueOnce(
+        new Promise((resolve) => {
+          finishExecution = resolve
+        })
+      )
+
+      const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
+        method: 'POST',
+        headers: { accept: 'application/json, text/event-stream' },
+        body: JSON.stringify({
+          jsonrpc: '2.0',
+          id: 1,
+          method: 'tools/call',
+          params: { name: 'tool_a', arguments: { q: 'test' } },
+        }),
+      })
+      const response = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
+
+      expect(response.status).toBe(200)
+      expect(response.headers.get('content-type')).toContain('text/event-stream')
+      if (!response.body) throw new Error('Expected MCP event stream')
+      const reader = response.body.getReader()
+      const decoder = new TextDecoder()
+      expect(decoder.decode((await reader.read()).value)).toBe(': keepalive\n\n')
+      await vi.advanceTimersByTimeAsync(15_000)
+      expect(decoder.decode((await reader.read()).value)).toBe(': keepalive\n\n')
+
+      finishExecution({
+        ok: true,
+        executionId: 'exec-1',
+        workflowId: 'wf-1',
+        status: 'completed',
+        aborted: null,
+        output: { ok: true },
+        error: null,
+        hasResponseBlock: false,
+        resolvedSecretTraceProvenance: createResolvedSecretTraceProvenance('owner-1'),
+      })
+
+      const event = decoder.decode((await reader.read()).value)
+      expect(JSON.parse(event.replace(/^data: /, '').trim())).toMatchObject({
+        jsonrpc: '2.0',
+        id: 1,
+        result: { content: [{ type: 'text' }], isError: false },
+      })
+      expect((await reader.read()).done).toBe(true)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('serves metadata when standalone SSE GET is explicitly rejected', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      {
+        id: 'server-1',
+        name: 'Public Server',
+        workspaceId: 'ws-1',
+        isPublic: true,
+        createdBy: 'owner-1',
+      },
+    ])
+
+    const request = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
+      headers: { accept: 'application/json, text/event-stream;q=0' },
+    })
+    const response = await GET(request, { params: Promise.resolve({ serverId: 'server-1' }) })
+
+    expect(response.status).toBe(200)
+    expect(await response.json()).toMatchObject({
+      name: 'Public Server',
+      capabilities: { tools: {} },
+    })
+  })
+
+  it('cancels the workflow when an MCP event-stream consumer disconnects', async () => {
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([
+        {
+          id: 'server-1',
+          name: 'Public Server',
+          workspaceId: 'ws-1',
+          isPublic: true,
+          createdBy: 'owner-1',
+        },
+      ])
+      .mockResolvedValueOnce([{ toolName: 'tool_a', workflowId: 'wf-1' }])
+      .mockResolvedValueOnce([{ workspaceId: 'ws-1', deploymentVersionId: 'deployment-1' }])
+
+    let executionSignal: AbortSignal | undefined
+    mockExecuteWorkflowService.mockImplementationOnce(
+      ({ abortSignal }: { abortSignal: AbortSignal }) =>
+        new Promise((resolve) => {
+          executionSignal = abortSignal
+          const finish = () =>
+            resolve({
+              ok: true,
+              executionId: 'exec-1',
+              workflowId: 'wf-1',
+              status: 'cancelled',
+              aborted: 'client',
+              output: undefined,
+              error: { message: 'Client cancelled request', code: 'CANCELLED' },
+              hasResponseBlock: false,
+            })
+          if (abortSignal.aborted) finish()
+          else abortSignal.addEventListener('abort', finish, { once: true })
+        })
+    )
+
+    const req = new NextRequest('http://localhost:3000/api/mcp/serve/server-1', {
+      method: 'POST',
+      headers: { accept: 'application/json, text/event-stream' },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 1,
+        method: 'tools/call',
+        params: { name: 'tool_a', arguments: { q: 'test' } },
+      }),
+    })
+    const response = await POST(req, { params: Promise.resolve({ serverId: 'server-1' }) })
+    if (!response.body) throw new Error('Expected MCP event stream')
+    const reader = response.body.getReader()
+    await reader.read()
+    await vi.waitFor(() => expect(executionSignal).toBeDefined())
+
+    await reader.cancel('client disconnected')
+
+    expect(executionSignal?.aborted).toBe(true)
+  })
+
   it('rejects a personal api key when the workspace disallows personal api keys', async () => {
     dbChainMockFns.limit.mockResolvedValueOnce([
       {