fix(workflows): preserve stream compatibility

Author: waleedlatif1

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

@@ -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({
@@ -406,6 +410,29 @@ describe('MCP Serve Route', () => {
     }
   })
 
+  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([

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

@@ -45,6 +45,7 @@ import {
   resolveBillingAttribution,
 } from '@/lib/billing/core/billing-attribution'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
+import { acceptsMediaType } from '@/lib/core/utils/media-types'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { encodeSSE, encodeSSEComment, SSE_HEADERS } from '@/lib/core/utils/sse'
 import {
@@ -141,7 +142,7 @@ function callerAbortedJsonRpcResponse(
 }
 
 function acceptsEventStream(request: NextRequest): boolean {
-  return request.headers.get('accept')?.includes('text/event-stream') === true
+  return acceptsMediaType(request.headers.get('accept'), 'text/event-stream')
 }
 
 /**
@@ -530,7 +531,7 @@ export const GET = withRouteHandler(
       const authResult = await authorizeMcpServeRequest(request, server)
       if (authResult.response) return authResult.response
 
-      if (request.headers.get('accept')?.includes('text/event-stream')) {
+      if (acceptsEventStream(request)) {
         return unsupportedSseGetResponse()
       }
 

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

@@ -24,6 +24,7 @@ import { requestExplicitStreamAbort } from '@/lib/copilot/request/session/explic
 import type { StreamEvent } from '@/lib/copilot/request/types'
 import { normalizeSecretMountPolicy } from '@/lib/copilot/secret-mount-policy'
 import { isDocSandboxEnabled } from '@/lib/core/config/env-flags'
+import { acceptsMediaType } from '@/lib/core/utils/media-types'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import {
@@ -82,7 +83,7 @@ function isAbortError(error: unknown): boolean {
 function wantsStreamedExecuteResponse(req: NextRequest): boolean {
   return (
     req.headers.get(MOTHERSHIP_EXECUTE_STREAM_HEADER) === MOTHERSHIP_EXECUTE_STREAM_VALUE ||
-    req.headers.get('accept')?.includes(MOTHERSHIP_EXECUTE_STREAM_CONTENT_TYPE) === true
+    acceptsMediaType(req.headers.get('accept'), MOTHERSHIP_EXECUTE_STREAM_CONTENT_TYPE)
   )
 }
 

apps/sim/app/api/v2/chat/route.ts

@@ -49,6 +49,7 @@ import {
   type WorkspaceAuthorizationContext,
 } from '@/lib/core/application'
 import { isDocSandboxEnabled } from '@/lib/core/config/env-flags'
+import { acceptsMediaType } from '@/lib/core/utils/media-types'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import { CAPABILITY_RULES } from '@/lib/permission-groups/capabilities'
@@ -128,7 +129,7 @@ function isAbortError(error: unknown): boolean {
 function wantsStreamedChatResponse(req: NextRequest): boolean {
   return (
     req.headers.get(CHAT_STREAM_HEADER) === CHAT_STREAM_VALUE ||
-    req.headers.get('accept')?.includes(CHAT_STREAM_CONTENT_TYPE) === true
+    acceptsMediaType(req.headers.get('accept'), CHAT_STREAM_CONTENT_TYPE)
   )
 }
 

apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts

@@ -421,6 +421,18 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => {
     }
   })
 
+  it('keeps the JSON response when NDJSON is explicitly rejected', async () => {
+    const response = await callExecute(
+      { input: { hello: 'world' } },
+      { Accept: 'application/json, application/x-ndjson;q=0' }
+    )
+
+    expect(response.headers.get('content-type')).toContain('application/json')
+    expect(await response.json()).toMatchObject({
+      data: { runId: 'execution-123', status: 'completed' },
+    })
+  })
+
   it('uses the heartbeat result transport for a manual draft run', async () => {
     authenticatePersonalKey()
     let finishExecution!: (result: unknown) => void

apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts

@@ -25,6 +25,7 @@ import { getWorkspaceBilledAccountUserId } from '@/lib/billing/core/billing-attr
 import { tryAdmit } from '@/lib/core/admission/gate'
 import { ADMISSION_ERROR_DESCRIPTOR } from '@/lib/core/admission/transient-failure'
 import type { ForbiddenDetailCode } from '@/lib/core/application'
+import { acceptsMediaType } from '@/lib/core/utils/media-types'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
@@ -110,7 +111,7 @@ function serviceFailureResponse(failure: ExecuteWorkflowServiceFailure) {
 }
 
 function wantsResultStream(req: NextRequest): boolean {
-  return req.headers.get('accept')?.includes(WORKFLOW_RESULT_STREAM_CONTENT_TYPE) === true
+  return acceptsMediaType(req.headers.get('accept'), WORKFLOW_RESULT_STREAM_CONTENT_TYPE)
 }
 
 function encodeNdjson(value: unknown): Uint8Array {

apps/sim/lib/core/utils/media-types.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, it } from 'vitest'
+import { acceptsMediaType } from '@/lib/core/utils/media-types'
+
+describe('acceptsMediaType', () => {
+  it.each([
+    ['application/json, application/x-ndjson', true],
+    ['application/x-ndjson; q=0.5', true],
+    ['Application/X-Ndjson;Q=1.000', true],
+    ['application/json, application/x-ndjson;q=0', false],
+    ['application/x-ndjson;q=0.000', false],
+    ['application/x-ndjson;q=1.1', false],
+    ['application/x-ndjson;q=invalid', false],
+    ['application/x-ndjson;q=1;q=0', false],
+    ['application/x-ndjson;profile="one,two";q=0', false],
+    ['application/x-ndjson;profile="one;two";q=0.5', true],
+    ['application/x-ndjson;profile="unterminated;q=0', false],
+    ['application/json', false],
+    ['*/*', false],
+    ['', false],
+  ])('parses %s', (header, expected) => {
+    expect(acceptsMediaType(header, 'application/x-ndjson')).toBe(expected)
+  })
+
+  it('rejects a missing Accept header', () => {
+    expect(acceptsMediaType(null, 'application/x-ndjson')).toBe(false)
+  })
+})

apps/sim/lib/core/utils/media-types.ts

@@ -0,0 +1,66 @@
+const QUALITY_VALUE = /^(?:0(?:\.\d{0,3})?|1(?:\.0{0,3})?)$/
+
+/** Splits an HTTP list or parameter list without cutting inside quoted strings. */
+function splitOutsideQuotes(value: string, separator: ',' | ';'): string[] | null {
+  const parts: string[] = []
+  let start = 0
+  let quoted = false
+  let escaped = false
+
+  for (let index = 0; index < value.length; index++) {
+    const character = value[index]
+    if (escaped) {
+      escaped = false
+      continue
+    }
+    if (quoted && character === '\\') {
+      escaped = true
+      continue
+    }
+    if (character === '"') {
+      quoted = !quoted
+      continue
+    }
+    if (!quoted && character === separator) {
+      parts.push(value.slice(start, index))
+      start = index + 1
+    }
+  }
+
+  if (quoted || escaped) return null
+  parts.push(value.slice(start))
+  return parts
+}
+
+/**
+ * Whether an Accept header explicitly permits a media type.
+ *
+ * Wildcards do not opt callers into a streaming protocol, and a matching range
+ * with an invalid or zero quality value is not acceptable.
+ */
+export function acceptsMediaType(acceptHeader: string | null, mediaType: string): boolean {
+  if (!acceptHeader) return false
+  const normalizedMediaType = mediaType.trim().toLowerCase()
+  const ranges = splitOutsideQuotes(acceptHeader, ',')
+  if (!ranges) return false
+
+  return ranges.some((range) => {
+    const parts = splitOutsideQuotes(range, ';')
+    if (!parts) return false
+    const [type, ...parameters] = parts
+    if (type.trim().toLowerCase() !== normalizedMediaType) return false
+
+    const qualityParameters = parameters.filter((parameter) => {
+      const separator = parameter.indexOf('=')
+      const name = separator === -1 ? parameter : parameter.slice(0, separator)
+      return name.trim().toLowerCase() === 'q'
+    })
+    if (qualityParameters.length === 0) return true
+    if (qualityParameters.length > 1) return false
+
+    const qualityParameter = qualityParameters[0]
+    const separator = qualityParameter.indexOf('=')
+    const quality = separator === -1 ? '' : qualityParameter.slice(separator + 1).trim()
+    return QUALITY_VALUE.test(quality) && Number(quality) > 0
+  })
+}

packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts

@@ -358,6 +358,16 @@ describe('sim workflows run --follow', () => {
     })
   })
 
+  it('translates API field names in synchronous run errors', async () => {
+    requestRaw.mockRejectedValue(
+      new SimApiError('executionTimeoutSeconds must be less than or equal to 3000', 400)
+    )
+
+    await expect(run(WORKFLOW_ID)).rejects.toThrow(
+      '--execution-timeout-seconds must be less than or equal to 3000'
+    )
+  })
+
   it('keeps async runs on the generated JSON path', async () => {
     request.mockResolvedValue({ data: { runId: 'run-1', statusUrl: '/runs/run-1' } })
     vi.spyOn(console, 'log').mockImplementation(() => {})

packages/sim-cli/src/commands/protocol/workflow-run-follow.ts

@@ -7,6 +7,7 @@ import { SimApiError } from '../../http/client'
 import { readNdjson } from '../../http/ndjson'
 import { safeOneLine, sanitize } from '../../output/render'
 import { executeOperation, runFailureMessage } from '../../runtime/execute'
+import { retypeApiError } from '../../runtime/naming'
 import { buildRequest } from '../../runtime/request'
 import { renderResult } from '../../runtime/result'
 import type { OperationSpec } from '../../runtime/types'
@@ -154,21 +155,27 @@ async function runWithResultStream(workflowId: string, command: Command): Promis
   const flags = command.optsWithGlobals() as Record<string, unknown>
   const { client, profile } = clientFrom(command)
   const operation = V2_OPERATIONS.executeWorkflow as OperationSpec
-  const request = buildRequest('executeWorkflow', [workflowId], flags, profile.workspaceId)
-  const response = await client.requestRaw(request.path, {
-    method: operation.method,
-    query: request.query,
-    body: request.body,
-    headers: { ...request.headers, accept: WORKFLOW_RESULT_STREAM_CONTENT_TYPE },
-  })
-  const payload = await readWorkflowResult(response)
+  const commandSpec = CLI_CONTRACT.executeWorkflow ?? {}
 
-  renderResult('executeWorkflow', profile.output, payload, CLI_CONTRACT.executeWorkflow ?? {}, {
-    expandedTrace: flags.trace === true,
-  })
-
-  const failure = runFailureMessage('executeWorkflow', payload)
-  if (failure) throw new SimApiError(failure, 0)
+  try {
+    const request = buildRequest('executeWorkflow', [workflowId], flags, profile.workspaceId)
+    const response = await client.requestRaw(request.path, {
+      method: operation.method,
+      query: request.query,
+      body: request.body,
+      headers: { ...request.headers, accept: WORKFLOW_RESULT_STREAM_CONTENT_TYPE },
+    })
+    const payload = await readWorkflowResult(response)
+
+    renderResult('executeWorkflow', profile.output, payload, commandSpec, {
+      expandedTrace: flags.trace === true,
+    })
+
+    const failure = runFailureMessage('executeWorkflow', payload)
+    if (failure) throw new SimApiError(failure, 0)
+  } catch (error) {
+    throw retypeApiError(error, 'executeWorkflow', commandSpec, operation)
+  }
 }
 
 /**