From b1d0cfd0043e914b1076cf7aace7437db432345d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 7 Sep 2026 18:25:04 -0700 Subject: [PATCH 1/4] fix(workflows): keep long-running calls active --- apps/docs/openapi-v2-workflows.json | 9 +- .../api/mcp/serve/[serverId]/route.test.ts | 126 +++++- .../sim/app/api/mcp/serve/[serverId]/route.ts | 124 ++++- .../[workflowId]/execute/route.test.ts | 117 +++++ .../workflows/[workflowId]/execute/route.ts | 178 +++++++- .../lib/api/contracts/v2/openapi/workflows.ts | 7 +- apps/sim/lib/copilot/request/session/sse.ts | 6 +- apps/sim/lib/core/utils/sse.ts | 5 + .../application/execute-manual-workflow.ts | 2 +- .../workflows/application/execute-workflow.ts | 2 +- .../lib/workflows/executor/execute-service.ts | 424 ++++++++++-------- .../lib/workflows/streaming/streaming.test.ts | 34 ++ apps/sim/lib/workflows/streaming/streaming.ts | 23 +- helm/sim/examples/values-aws.yaml | 1 + .../sim-cli/src/commands/protocol/chat.ts | 68 +-- .../protocol/workflow-run-follow.test.ts | 121 ++++- .../commands/protocol/workflow-run-follow.ts | 79 +++- packages/sim-cli/src/http/client.test.ts | 14 + packages/sim-cli/src/http/client.ts | 27 +- packages/sim-cli/src/http/ndjson.ts | 47 ++ packages/sim-cli/src/runtime/execute.ts | 2 +- scripts/openapi/documents.test.ts | 6 +- 22 files changed, 1121 insertions(+), 301 deletions(-) create mode 100644 packages/sim-cli/src/http/ndjson.ts diff --git a/apps/docs/openapi-v2-workflows.json b/apps/docs/openapi-v2-workflows.json index 93898c40177..d23a31aac73 100644 --- a/apps/docs/openapi-v2-workflows.json +++ b/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" diff --git a/apps/sim/app/api/mcp/serve/[serverId]/route.test.ts b/apps/sim/app/api/mcp/serve/[serverId]/route.test.ts index 2f17fb5ea34..6b60e88531b 100644 --- a/apps/sim/app/api/mcp/serve/[serverId]/route.test.ts +++ b/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', () => ({ @@ -338,6 +338,130 @@ 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('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([ { diff --git a/apps/sim/app/api/mcp/serve/[serverId]/route.ts b/apps/sim/app/api/mcp/serve/[serverId]/route.ts index 6ec14605ad8..bf1b17a81bd 100644 --- a/apps/sim/app/api/mcp/serve/[serverId]/route.ts +++ b/apps/sim/app/api/mcp/serve/[serverId]/route.ts @@ -27,6 +27,7 @@ import { workspace, } from '@sim/db/schema' import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' import { isRecordLike } from '@sim/utils/object' import { and, asc, eq, gt, isNull, sql } from 'drizzle-orm' import { type NextRequest, NextResponse } from 'next/server' @@ -45,6 +46,7 @@ import { } from '@/lib/billing/core/billing-attribution' import { getMaxExecutionTimeout } from '@/lib/core/execution-limits' import { generateRequestId } from '@/lib/core/utils/request' +import { encodeSSE, encodeSSEComment, SSE_HEADERS } from '@/lib/core/utils/sse' import { assertContentLengthWithinLimit, assertKnownSizeWithinLimit, @@ -75,6 +77,7 @@ const MAX_MCP_WORKFLOW_REQUEST_BYTES = 10 * 1024 * 1024 const MAX_MCP_TOOL_RESULT_TEXT_BYTES = 10 * 1024 * 1024 const MAX_MCP_TOOLS_LIST_COUNT = MAX_MCP_TOOLS_PER_SERVER const MAX_MCP_TOOLS_LIST_SCHEMA_BYTES = MAX_MCP_PARAMETER_SCHEMA_BYTES +const MCP_STREAM_KEEPALIVE_INTERVAL_MS = 15_000 const MB = 1024 * 1024 function negotiateProtocolVersion(rpcParams: unknown): string { @@ -137,6 +140,95 @@ function callerAbortedJsonRpcResponse( return abortSignal?.isCallerAborted() ? clientCancelledJsonRpcResponse(id) : null } +function acceptsEventStream(request: NextRequest): boolean { + return request.headers.get('accept')?.includes('text/event-stream') === true +} + +/** + * Sends a Streamable HTTP response as SSE so a long tool call can keep the + * connection active before its terminal JSON-RPC message is available. + */ +function streamJsonRpcResponse( + id: RequestId, + requestSignal: AbortSignal, + run: (signal: AbortSignal) => Promise +): Response { + const executionController = new AbortController() + let cancelled = false + let keepaliveId: ReturnType | undefined + + const stopKeepalive = () => { + if (keepaliveId) { + clearInterval(keepaliveId) + keepaliveId = undefined + } + } + const abortExecution = (reason?: unknown) => { + if (!executionController.signal.aborted) { + executionController.abort(reason ?? new Error('MCP client disconnected')) + } + } + const abortFromRequest = () => abortExecution(requestSignal.reason) + + if (requestSignal.aborted) { + abortFromRequest() + } else { + requestSignal.addEventListener('abort', abortFromRequest, { once: true }) + } + + const stream = new ReadableStream({ + start(controller) { + const send = (chunk: Uint8Array): boolean => { + if (cancelled) return false + try { + controller.enqueue(chunk) + return true + } catch { + cancelled = true + stopKeepalive() + abortExecution() + return false + } + } + + if (send(encodeSSEComment('keepalive'))) { + keepaliveId = setInterval(() => { + send(encodeSSEComment('keepalive')) + }, MCP_STREAM_KEEPALIVE_INTERVAL_MS) + } + + void run(executionController.signal) + .then(async (response) => { + const message: unknown = await response.json() + send(encodeSSE(message)) + }) + .catch((error) => { + logger.error('MCP response stream failed', { error: getErrorMessage(error) }) + send(encodeSSE(createError(id, ErrorCode.InternalError, 'Internal error'))) + }) + .finally(() => { + stopKeepalive() + requestSignal.removeEventListener('abort', abortFromRequest) + if (!cancelled) controller.close() + }) + }, + cancel(reason) { + cancelled = true + stopKeepalive() + requestSignal.removeEventListener('abort', abortFromRequest) + abortExecution(reason) + }, + }) + + return new Response(stream, { + headers: { + ...SSE_HEADERS, + 'Cache-Control': 'no-cache, no-transform', + Vary: 'Accept', + }, + }) +} + function limitMessage(label: string, maxBytes: number): string { return `${label} exceeds maximum size of ${Math.round(maxBytes / MB)}MB` } @@ -406,12 +498,12 @@ async function authorizeMcpServeRequest( } } -function unsupportedSseTransportResponse(): NextResponse { +function unsupportedSseGetResponse(): NextResponse { return NextResponse.json( { error: { code: 'unsupported_transport', - message: 'SSE transport is not supported for workflow MCP servers', + message: 'Standalone SSE GET transport is not supported for workflow MCP servers', supportedTransports: ['streamable-http'], allowedMethods: ['GET', 'POST', 'DELETE'], }, @@ -439,7 +531,7 @@ export const GET = withRouteHandler( if (authResult.response) return authResult.response if (request.headers.get('accept')?.includes('text/event-stream')) { - return unsupportedSseTransportResponse() + return unsupportedSseGetResponse() } return NextResponse.json({ @@ -557,16 +649,22 @@ export const POST = withRouteHandler( ) } - return handleToolsCall( - id, - serverId, - server.workspaceId, - paramsValidation.data, - executeAuthContext, - server.isPublic ? server.createdBy : undefined, - request.headers.get(SIM_VIA_HEADER), - request.signal - ) + const callTool = (signal: AbortSignal) => + handleToolsCall( + id, + serverId, + server.workspaceId, + paramsValidation.data, + executeAuthContext, + server.isPublic ? server.createdBy : undefined, + request.headers.get(SIM_VIA_HEADER), + signal + ) + + if (acceptsEventStream(request)) { + return streamJsonRpcResponse(id, request.signal, callTool) + } + return callTool(request.signal) } default: diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts index e1890d05097..bae20236ba0 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts @@ -20,6 +20,7 @@ import { WorkspaceApiKeyAuthorizationError } from '@/lib/core/application' const { MockV2ApiKeyUnauthenticatedError, + mockAdmissionRelease, mockAuthenticateV2ApiKey, mockClaimExecutionId, mockCheckOperationRate, @@ -35,6 +36,7 @@ const { mockValidatePublicApiAllowed, } = vi.hoisted(() => ({ MockV2ApiKeyUnauthenticatedError: class MockV2ApiKeyUnauthenticatedError extends Error {}, + mockAdmissionRelease: vi.fn(), mockAuthenticateV2ApiKey: vi.fn(), mockClaimExecutionId: vi.fn(), mockCheckOperationRate: vi.fn(), @@ -50,6 +52,10 @@ const { mockValidatePublicApiAllowed: vi.fn(), })) +vi.mock('@/lib/core/admission/gate', () => ({ + tryAdmit: vi.fn(() => ({ release: mockAdmissionRelease })), +})) + vi.mock('@/lib/workflows/application/execute-manual-workflow', () => ({ executeManualWorkflowOperation: { execute: mockExecuteManualTrigger }, executeManualWorkflowFromBlockOperation: { execute: mockExecuteManualFromBlock }, @@ -353,6 +359,116 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => { }) }) + it('streams an immediate heartbeat and the same sync result when NDJSON is accepted', async () => { + vi.useFakeTimers() + try { + let finishExecution!: (result: unknown) => void + mockExecuteWorkflowCore.mockReturnValueOnce( + new Promise((resolve) => { + finishExecution = resolve + }) + ) + + const response = await callExecute( + { input: { hello: 'world' } }, + { Accept: 'application/x-ndjson' } + ) + + expect(response.status).toBe(200) + expect(response.headers.get('content-type')).toContain('application/x-ndjson') + expect(response.headers.get('X-Run-Id')).toBe('execution-123') + if (!response.body) throw new Error('Expected NDJSON response body') + const reader = response.body.getReader() + const decoder = new TextDecoder() + expect(JSON.parse(decoder.decode((await reader.read()).value))).toMatchObject({ + type: 'heartbeat', + }) + expect(mockAdmissionRelease).not.toHaveBeenCalled() + expect(mockReleaseExecutionIdClaim).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(15_000) + expect(JSON.parse(decoder.decode((await reader.read()).value))).toMatchObject({ + type: 'heartbeat', + }) + + finishExecution({ + success: true, + output: { result: 'done' }, + metadata: { + duration: 42, + startTime: '2026-07-31T00:00:00.000Z', + endTime: '2026-07-31T00:00:01.000Z', + }, + }) + + expect(JSON.parse(decoder.decode((await reader.read()).value))).toEqual({ + type: 'final', + data: { + runId: 'execution-123', + workflowId: 'workflow-1', + status: 'completed', + output: { result: 'done' }, + error: null, + startedAt: '2026-07-31T00:00:00.000Z', + endedAt: '2026-07-31T00:00:01.000Z', + durationMs: 42, + }, + }) + expect((await reader.read()).done).toBe(true) + expect(mockAdmissionRelease).toHaveBeenCalledTimes(1) + expect(mockReleaseExecutionIdClaim).toHaveBeenCalledTimes(1) + } finally { + vi.useRealTimers() + } + }) + + it('uses the heartbeat result transport for a manual draft run', async () => { + authenticatePersonalKey() + let finishExecution!: (result: unknown) => void + const pending = new Promise((resolve) => { + finishExecution = resolve + }) + mockExecuteManualTrigger.mockResolvedValueOnce({ + ok: true, + executionId: 'execution-123', + pending, + cancel: vi.fn(), + }) + + const response = await callExecute( + { run: { source: 'manual' }, input: { hello: 'world' } }, + { Accept: 'application/x-ndjson' } + ) + + expect(response.headers.get('content-type')).toContain('application/x-ndjson') + expect(mockExecuteManualTrigger).toHaveBeenCalledWith( + expect.objectContaining({ + input: expect.objectContaining({ mode: 'sync-result-stream' }), + }) + ) + if (!response.body) throw new Error('Expected NDJSON response body') + const reader = response.body.getReader() + await reader.read() + + finishExecution({ + ok: true, + executionId: 'execution-123', + workflowId: 'workflow-1', + status: 'completed', + aborted: null, + output: { result: 'manual done' }, + error: null, + hasResponseBlock: false, + }) + + const final = JSON.parse(new TextDecoder().decode((await reader.read()).value)) + expect(final).toMatchObject({ + type: 'final', + data: { runId: 'execution-123', output: { result: 'manual done' } }, + }) + expect((await reader.read()).done).toBe(true) + expect(mockAdmissionRelease).toHaveBeenCalledTimes(1) + }) + it('returns status failed with a structured error instead of an HTTP error', async () => { const error = new Error('Send Email: Invalid credentials') Object.assign(error, { blockId: 'block-9', blockName: 'Send Email', blockType: 'gmail' }) @@ -548,6 +664,7 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => { expect(response.status).toBe(200) expect(response.headers.get('Content-Type')).toContain('text/event-stream') + expect(response.headers.get('X-Run-Id')).toBe('execution-123') expect(await response.text()).toBe('data: "[DONE]"\n\n') expect(mockExecuteManualTrigger).toHaveBeenCalledWith( expect.objectContaining({ input: expect.objectContaining({ mode: 'stream' }) }) diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts index ad737bbe6dc..426294c3dee 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts @@ -43,7 +43,9 @@ import { executeWorkflowOperation } from '@/lib/workflows/application/execute-wo import { workflowOperations } from '@/lib/workflows/application/operations' import { type ExecuteWorkflowServiceFailure, + type ExecuteWorkflowServicePendingRun, type ExecuteWorkflowServiceResult, + type ExecuteWorkflowServiceRun, executeWorkflowService, } from '@/lib/workflows/executor/execute-service' import { @@ -60,6 +62,10 @@ import { const logger = createLogger('V2WorkflowExecuteAPI') +const WORKFLOW_RESULT_STREAM_CONTENT_TYPE = 'application/x-ndjson' +const WORKFLOW_RESULT_HEARTBEAT_INTERVAL_MS = 15_000 +const ndjsonEncoder = new TextEncoder() + export const runtime = 'nodejs' export const dynamic = 'force-dynamic' @@ -103,6 +109,124 @@ function serviceFailureResponse(failure: ExecuteWorkflowServiceFailure) { }) } +function wantsResultStream(req: NextRequest): boolean { + return req.headers.get('accept')?.includes(WORKFLOW_RESULT_STREAM_CONTENT_TYPE) === true +} + +function encodeNdjson(value: unknown): Uint8Array { + return ndjsonEncoder.encode(`${JSON.stringify(value)}\n`) +} + +function presentRun(result: ExecuteWorkflowServiceRun) { + return { + runId: result.executionId, + workflowId: result.workflowId, + status: result.status, + output: result.output ?? null, + error: result.error, + startedAt: result.startedAt, + endedAt: result.endedAt, + durationMs: result.durationMs, + } +} + +/** + * Presents an ordinary synchronous run as heartbeat-delimited NDJSON. The + * execution promise is the same one used by the JSON path; only its response + * framing changes, so manual draft selection and cancellation keep their + * existing semantics. + */ +function streamPendingRun( + pendingRun: ExecuteWorkflowServicePendingRun, + requestId: string, + onSettled: () => void +): Response { + let cancelled = false + let heartbeatId: ReturnType | undefined + const stopHeartbeat = () => { + if (heartbeatId) { + clearInterval(heartbeatId) + heartbeatId = undefined + } + } + + const stream = new ReadableStream({ + start(controller) { + const cancel = () => { + if (cancelled) return + cancelled = true + stopHeartbeat() + pendingRun.cancel() + } + const send = (event: unknown): boolean => { + if (cancelled) return false + try { + controller.enqueue(encodeNdjson(event)) + return true + } catch { + cancel() + return false + } + } + + if (send({ type: 'heartbeat', timestamp: new Date().toISOString() })) { + heartbeatId = setInterval(() => { + send({ type: 'heartbeat', timestamp: new Date().toISOString() }) + }, WORKFLOW_RESULT_HEARTBEAT_INTERVAL_MS) + } + + void pendingRun.pending + .then((result) => { + if (!result.ok) { + send({ + type: 'error', + error: result.failure.message, + code: result.failure.code, + status: result.failure.statusCode, + }) + return + } + if (result.aborted === 'client') { + send({ + type: 'error', + error: 'Client cancelled request', + code: 'CLIENT_CLOSED_REQUEST', + status: 499, + }) + return + } + send({ type: 'final', data: presentRun(result) }) + }) + .catch((error) => { + logger.error(`[${requestId}] v2 execute result stream failed`, { + error: getErrorMessage(error, 'Unknown error'), + }) + send({ type: 'error', error: 'Internal server error', status: 500 }) + }) + .finally(() => { + stopHeartbeat() + onSettled() + if (!cancelled) controller.close() + }) + }, + cancel() { + cancelled = true + stopHeartbeat() + pendingRun.cancel() + }, + }) + + return new Response(stream, { + headers: { + 'Content-Type': `${WORKFLOW_RESULT_STREAM_CONTENT_TYPE}; charset=utf-8`, + 'Cache-Control': 'no-cache, no-transform', + 'X-Accel-Buffering': 'no', + Vary: 'Accept', + [V2_WORKFLOW_RUN_ID_HEADER]: pendingRun.executionId, + }, + }) +} + /** * Path parameters read straight from the Next context, typed from the contract * rather than restated inline. @@ -129,7 +253,9 @@ type V2ExecuteWorkflowRouteContext = { * - `async: true` (body flag — v2 has no mode headers) → 202 * `{ data: { runId, statusUrl } }`; poll the v2 runs resource. * - `stream: true` → SSE passthrough (no `{data}` envelope on event frames). - * - Sync → 200 run resource with the status enum and structured error; + * - Sync → 200 run resource with the status enum and structured error. A caller + * that accepts `application/x-ndjson` receives heartbeat frames followed by + * the same resource in a `final` frame; * an in-band run failure is `status: 'failed'`, never an HTTP error. A * Response block's declared payload stays inside `output` — v2 never lets a * workflow author control response status or headers on this origin. @@ -216,6 +342,7 @@ export const POST = withRouteHandler( }) } + let ticketTransferred = false try { const parsed = await parseRequest(v2ExecuteWorkflowContract, req, context, { ...V2_PARSE_DEFAULTS, @@ -298,6 +425,8 @@ export const POST = withRouteHandler( ) } + const resultStream = !body.async && !body.stream && wantsResultStream(req) + /** Caller-supplied run IDs are a keyed-caller feature; anonymous callers must not probe the claim table. */ let requestedExecutionId: string | undefined const runIdHeader = parsed.data.headers['x-run-id'] @@ -326,7 +455,7 @@ export const POST = withRouteHandler( input: { ...commonInput, input: body.input, - mode: body.stream ? 'stream' : 'sync', + mode: body.stream ? 'stream' : resultStream ? 'sync-result-stream' : 'sync', blockId: manualRun.entry.blockId, sourceRunId: manualRun.entry.sourceRunId, }, @@ -338,7 +467,7 @@ export const POST = withRouteHandler( input: { ...commonInput, input: body.input, - mode: body.stream ? 'stream' : 'sync', + mode: body.stream ? 'stream' : resultStream ? 'sync-result-stream' : 'sync', triggerBlockId: manualRun.entry?.blockId, useMockPayload: manualRun.entry?.useMockPayload === true, }, @@ -351,7 +480,13 @@ export const POST = withRouteHandler( ...commonInput, input: body.input ?? {}, requestedTimeoutSeconds: body.executionTimeoutSeconds, - mode: body.async ? 'async' : body.stream ? 'stream' : 'sync', + mode: body.async + ? 'async' + : body.stream + ? 'stream' + : resultStream + ? 'sync-result-stream' + : 'sync', }, request: req, }) @@ -400,7 +535,7 @@ export const POST = withRouteHandler( selectedOutputs: body.selectedOutputs, rateLimitCounter: 'sync', abortSignal: req.signal, - mode: body.stream ? 'stream' : 'sync', + mode: body.stream ? 'stream' : resultStream ? 'sync-result-stream' : 'sync', requestHeaders: req.headers, includeThinking: body.includeThinking, includeToolCalls: body.includeToolCalls, @@ -412,9 +547,20 @@ export const POST = withRouteHandler( return serviceFailureResponse(result.failure) } + if ('pending' in result) { + const response = streamPendingRun(result, requestId, ticket.release) + ticketTransferred = true + return response + } + if ('stream' in result) { - // SSE: pass the stream through byte-for-byte with its own headers. - return result.stream + const headers = new Headers(result.stream.headers) + headers.set(V2_WORKFLOW_RUN_ID_HEADER, result.executionId) + return new Response(result.stream.body, { + status: result.stream.status, + statusText: result.stream.statusText, + headers, + }) } if ('queued' in result) { @@ -433,19 +579,9 @@ export const POST = withRouteHandler( }) } - return v2Data( - { - runId: result.executionId, - workflowId: result.workflowId, - status: result.status, - output: result.output ?? null, - error: result.error, - startedAt: result.startedAt, - endedAt: result.endedAt, - durationMs: result.durationMs, - }, - { headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId } } - ) + return v2Data(presentRun(result), { + headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId }, + }) } catch (error) { const classified = v2WorkflowErrorPolicies.concealWorkflowAuthorization.render(error) if (classified) return classified @@ -455,7 +591,7 @@ export const POST = withRouteHandler( }) return v2Error('INTERNAL_ERROR', 'Internal server error') } finally { - ticket.release() + if (!ticketTransferred) ticket.release() } }, { diff --git a/apps/sim/lib/api/contracts/v2/openapi/workflows.ts b/apps/sim/lib/api/contracts/v2/openapi/workflows.ts index f60b9495fcf..af71fe09dce 100644 --- a/apps/sim/lib/api/contracts/v2/openapi/workflows.ts +++ b/apps/sim/lib/api/contracts/v2/openapi/workflows.ts @@ -1119,7 +1119,7 @@ const declaredRoutes = [ applicationOperation: workflowOperations.execute, 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\`. ${EXECUTE_OPTION_CONSTRAINTS}`, + 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\`. ${EXECUTE_OPTION_CONSTRAINTS}`, errors: [ 'BadRequest', 'Unauthorized', @@ -1137,9 +1137,10 @@ const declaredRoutes = [ success: { byStatus: { 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', ...RATE_LIMIT_HEADERS], - additionalContentTypes: ['text/event-stream'], + additionalContentTypes: ['application/x-ndjson', 'text/event-stream'], }, 202: { description: 'The asynchronous run was queued.', diff --git a/apps/sim/lib/copilot/request/session/sse.ts b/apps/sim/lib/copilot/request/session/sse.ts index 74b11454047..c19bd553fd1 100644 --- a/apps/sim/lib/copilot/request/session/sse.ts +++ b/apps/sim/lib/copilot/request/session/sse.ts @@ -1,15 +1,13 @@ import { SSE_HEADERS } from '@/lib/core/utils/sse' +export { encodeSSEComment } from '@/lib/core/utils/sse' + const encoder = new TextEncoder() export function encodeSSEEnvelope(envelope: unknown): Uint8Array { return encoder.encode(`data: ${JSON.stringify(envelope)}\n\n`) } -export function encodeSSEComment(comment: string): Uint8Array { - return encoder.encode(`: ${comment}\n\n`) -} - export const SSE_RESPONSE_HEADERS = { ...SSE_HEADERS, 'Content-Encoding': 'none', diff --git a/apps/sim/lib/core/utils/sse.ts b/apps/sim/lib/core/utils/sse.ts index 2651147136e..0d8f74183c4 100644 --- a/apps/sim/lib/core/utils/sse.ts +++ b/apps/sim/lib/core/utils/sse.ts @@ -20,6 +20,11 @@ export function encodeSSE(data: any): Uint8Array { return new TextEncoder().encode(`data: ${JSON.stringify(data)}\n\n`) } +/** Encodes an SSE comment, which clients ignore while intermediaries observe response activity. */ +export function encodeSSEComment(comment: string): Uint8Array { + return new TextEncoder().encode(`: ${comment}\n\n`) +} + /** * The sentinel value servers emit to signal end-of-stream. Lines carrying this * payload are skipped before reaching the consumer's `onEvent` callback. diff --git a/apps/sim/lib/workflows/application/execute-manual-workflow.ts b/apps/sim/lib/workflows/application/execute-manual-workflow.ts index 998b3faebe2..2ba1c8e5afb 100644 --- a/apps/sim/lib/workflows/application/execute-manual-workflow.ts +++ b/apps/sim/lib/workflows/application/execute-manual-workflow.ts @@ -19,7 +19,7 @@ import { interface ManualExecutionInput extends Omit { input?: unknown - mode: 'sync' | 'stream' + mode: 'sync' | 'stream' | 'sync-result-stream' } export interface ExecuteManualWorkflowInput extends ManualExecutionInput { diff --git a/apps/sim/lib/workflows/application/execute-workflow.ts b/apps/sim/lib/workflows/application/execute-workflow.ts index 48b77ce9306..d7c8f74aed4 100644 --- a/apps/sim/lib/workflows/application/execute-workflow.ts +++ b/apps/sim/lib/workflows/application/execute-workflow.ts @@ -18,7 +18,7 @@ export interface ExecuteWorkflowInput { selectedOutputs?: string[] requestedTimeoutSeconds?: number abortSignal?: AbortSignal - mode: 'sync' | 'async' | 'stream' + mode: 'sync' | 'async' | 'stream' | 'sync-result-stream' requestHeaders: Headers includeThinking?: boolean includeToolCalls?: boolean diff --git a/apps/sim/lib/workflows/executor/execute-service.ts b/apps/sim/lib/workflows/executor/execute-service.ts index 7edae7f8b2c..be5d94c4fae 100644 --- a/apps/sim/lib/workflows/executor/execute-service.ts +++ b/apps/sim/lib/workflows/executor/execute-service.ts @@ -104,8 +104,10 @@ export interface ExecuteWorkflowServiceParams { * `async`: enqueue and return the queue receipt. * `stream`: return an SSE Response (agent-stream protocol negotiated from * `requestHeaders`). + * `sync-result-stream`: start the same work as `sync` and return its pending + * result so an HTTP surface can emit transport heartbeats while it runs. */ - mode?: 'sync' | 'async' | 'stream' + mode?: 'sync' | 'async' | 'stream' | 'sync-result-stream' /** Original request headers — stream-protocol negotiation only. */ requestHeaders?: Headers includeThinking?: boolean @@ -162,13 +164,38 @@ export interface ExecuteWorkflowServiceStream { executionId: string } -export type ExecuteWorkflowServiceResult = +/** + * A prepared synchronous run that a streaming surface can present while the + * exact same execution promise used by `sync` is pending. + */ +export interface ExecuteWorkflowServicePendingRun { + ok: true + executionId: string + pending: Promise + cancel: () => void +} + +export interface ExecuteWorkflowServiceFailureResult { + ok: false + failure: ExecuteWorkflowServiceFailure +} + +export type ExecuteWorkflowServiceTerminalResult = | ExecuteWorkflowServiceRun + | ExecuteWorkflowServiceFailureResult + +export type ExecuteWorkflowServiceResult = + | ExecuteWorkflowServiceTerminalResult | ExecuteWorkflowServiceQueued | ExecuteWorkflowServiceStream - | { ok: false; failure: ExecuteWorkflowServiceFailure } + | ExecuteWorkflowServicePendingRun -function failure(f: ExecuteWorkflowServiceFailure): ExecuteWorkflowServiceResult { +export type ExecuteWorkflowServiceImmediateResult = Exclude< + ExecuteWorkflowServiceResult, + ExecuteWorkflowServicePendingRun +> + +function failure(f: ExecuteWorkflowServiceFailure): ExecuteWorkflowServiceFailureResult { return { ok: false, failure: f } } @@ -206,6 +233,15 @@ async function compactServiceOutput( return compacted } +export function executeWorkflowService( + params: ExecuteWorkflowServiceParams & { mode: 'sync-result-stream' } +): Promise +export function executeWorkflowService( + params: ExecuteWorkflowServiceParams & { mode?: 'sync' | 'async' | 'stream' } +): Promise +export function executeWorkflowService( + params: ExecuteWorkflowServiceParams +): Promise export async function executeWorkflowService( params: ExecuteWorkflowServiceParams ): Promise { @@ -267,6 +303,32 @@ export async function executeWorkflowService( let executionIdClaim: ExecutionIdClaim | null = null let executionIdClaimCommitted = false + let executionIdClaimTransferred = false + + const settleExecutionIdClaim = async () => { + if (!executionIdClaim || executionIdClaimCommitted) return + + try { + executionIdClaimCommitted = await hasDurableExecutionOwner(executionId) + } catch (error) { + executionIdClaimCommitted = true + reqLogger.warn('Unable to verify execution ID ownership; retaining claim', { + error: toError(error).message, + executionId, + }) + } + + if (executionIdClaimCommitted) return + try { + await releaseExecutionIdClaim(executionIdClaim) + executionIdClaim = null + } catch (error) { + reqLogger.warn('Failed to release pre-start execution ID claim', { + error: toError(error).message, + executionId, + }) + } + } try { try { @@ -622,191 +684,214 @@ export async function executeWorkflowService( rejectLargeInlineOutput, } - try { - const snapshot = new ExecutionSnapshot( - metadata, - workflow, - processedInput, - workflowVariables, - selectedOutputs - ) - - const result = await executeWorkflowCore({ - snapshot, - callbacks: {}, - loggingSession, - includeFileBase64, - base64MaxBytes, - abortSignal: timeoutController.signal, - runFromBlock, - }) + const runSynchronousWorkflow = async (): Promise => { + try { + const snapshot = new ExecutionSnapshot( + metadata, + workflow, + processedInput, + workflowVariables, + selectedOutputs + ) - await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession }) + const result = await executeWorkflowCore({ + snapshot, + callbacks: {}, + loggingSession, + includeFileBase64, + base64MaxBytes, + abortSignal: timeoutController.signal, + runFromBlock, + }) - if (result.status === 'cancelled' && isRequestAborted() && !timeoutController.isTimedOut()) { - reqLogger.info('Execution cancelled by client disconnect') - await loggingSession.markAsFailed('Client cancelled request') - return { - ok: true, - executionId, - workflowId, - status: 'cancelled', - aborted: 'client', - output: undefined, - error: { message: 'Client cancelled request', code: 'CANCELLED' }, - resolvedSecretTraceProvenance: result.executionState?.resolvedSecretTraceProvenance, - hasResponseBlock: false, + await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession }) + + if ( + result.status === 'cancelled' && + isRequestAborted() && + !timeoutController.isTimedOut() + ) { + reqLogger.info('Execution cancelled by client disconnect') + await loggingSession.markAsFailed('Client cancelled request') + return { + ok: true, + executionId, + workflowId, + status: 'cancelled', + aborted: 'client', + output: undefined, + error: { message: 'Client cancelled request', code: 'CANCELLED' }, + resolvedSecretTraceProvenance: result.executionState?.resolvedSecretTraceProvenance, + hasResponseBlock: false, + } } - } - if ( - result.status === 'cancelled' && - timeoutController.isTimedOut() && - timeoutController.timeoutMs - ) { - const timeoutErrorMessage = getTimeoutErrorMessage(null, timeoutController.timeoutMs) - reqLogger.info('Execution timed out', { timeoutMs: timeoutController.timeoutMs }) - await loggingSession.markAsFailed(timeoutErrorMessage) - const compactTimeoutOutput = await compactServiceOutput(result.output, compactionContext) + if ( + result.status === 'cancelled' && + timeoutController.isTimedOut() && + timeoutController.timeoutMs + ) { + const timeoutErrorMessage = getTimeoutErrorMessage(null, timeoutController.timeoutMs) + reqLogger.info('Execution timed out', { timeoutMs: timeoutController.timeoutMs }) + await loggingSession.markAsFailed(timeoutErrorMessage) + const compactTimeoutOutput = await compactServiceOutput(result.output, compactionContext) + return { + ok: true, + executionId, + workflowId, + status: 'failed', + aborted: 'timeout', + output: compactTimeoutOutput, + error: { message: timeoutErrorMessage, code: 'TIMEOUT' }, + resolvedSecretTraceProvenance: result.executionState?.resolvedSecretTraceProvenance, + hasResponseBlock: false, + startedAt: result.metadata?.startTime, + endedAt: result.metadata?.endTime, + durationMs: result.metadata?.duration, + } + } + + const outputWithBase64 = + includeFileBase64 && !rejectLargeInlineOutput + ? ((await hydrateUserFilesWithBase64(result.output, { + requestId, + workspaceId, + workflowId, + executionId, + largeValueExecutionIds: [executionId], + largeValueKeys: result.metadata?.largeValueKeys ?? [], + fileKeys: result.metadata?.fileKeys ?? [], + allowLargeValueWorkflowScope: false, + userId: actorUserId, + principal, + maxBytes: base64MaxBytes, + preserveLargeValueMetadata: true, + })) as NormalizedBlockOutput) + : result.output + + const compactOutput = await compactServiceOutput(outputWithBase64, compactionContext) + + const status: ExecuteWorkflowServiceRun['status'] = + result.status === 'paused' + ? 'paused' + : result.status === 'cancelled' + ? 'cancelled' + : result.success + ? 'completed' + : 'failed' + return { ok: true, executionId, workflowId, - status: 'failed', - aborted: 'timeout', - output: compactTimeoutOutput, - error: { message: timeoutErrorMessage, code: 'TIMEOUT' }, + status, + aborted: null, + output: compactOutput, + error: + status === 'failed' || (status === 'cancelled' && result.error) + ? classifyExecutionError(result.error ? new Error(result.error) : undefined, result) + : null, resolvedSecretTraceProvenance: result.executionState?.resolvedSecretTraceProvenance, - hasResponseBlock: false, + hasResponseBlock: workflowHasResponseBlock(result), startedAt: result.metadata?.startTime, endedAt: result.metadata?.endTime, durationMs: result.metadata?.duration, } - } + } catch (error: unknown) { + const errorMessage = getErrorMessage(error, 'Unknown error') + const executionResult = hasExecutionResult(error) ? error.executionResult : undefined + + if (isRequestAborted() && !timeoutController.isTimedOut()) { + reqLogger.info('Execution aborted after client disconnect') + return { + ok: true, + executionId, + workflowId, + status: 'cancelled', + aborted: 'client', + output: undefined, + error: { message: 'Client cancelled request', code: 'CANCELLED' }, + resolvedSecretTraceProvenance: + executionResult?.executionState?.resolvedSecretTraceProvenance, + hasResponseBlock: false, + } + } - const outputWithBase64 = - includeFileBase64 && !rejectLargeInlineOutput - ? ((await hydrateUserFilesWithBase64(result.output, { - requestId, - workspaceId, - workflowId, - executionId, - largeValueExecutionIds: [executionId], - largeValueKeys: result.metadata?.largeValueKeys ?? [], - fileKeys: result.metadata?.fileKeys ?? [], - allowLargeValueWorkflowScope: false, - userId: actorUserId, - principal, - maxBytes: base64MaxBytes, - preserveLargeValueMetadata: true, - })) as NormalizedBlockOutput) - : result.output - - const compactOutput = await compactServiceOutput(outputWithBase64, compactionContext) - - const status: ExecuteWorkflowServiceRun['status'] = - result.status === 'paused' - ? 'paused' - : result.status === 'cancelled' - ? 'cancelled' - : result.success - ? 'completed' - : 'failed' + if ( + error instanceof PayloadSizeLimitError && + rejectLargeInlineOutput && + error.label === 'Workflow execution response' + ) { + return failure({ + kind: 'output_too_large', + message: 'Workflow execution response exceeds maximum size', + statusCode: 413, + code: 'workflow_response_too_large', + executionId, + }) + } - return { - ok: true, - executionId, - workflowId, - status, - aborted: null, - output: compactOutput, - error: - status === 'failed' || (status === 'cancelled' && result.error) - ? classifyExecutionError(result.error ? new Error(result.error) : undefined, result) - : null, - resolvedSecretTraceProvenance: result.executionState?.resolvedSecretTraceProvenance, - hasResponseBlock: workflowHasResponseBlock(result), - startedAt: result.metadata?.startTime, - endedAt: result.metadata?.endTime, - durationMs: result.metadata?.duration, - } - } catch (error: unknown) { - const errorMessage = getErrorMessage(error, 'Unknown error') - const executionResult = hasExecutionResult(error) ? error.executionResult : undefined + reqLogger.error(`Execution failed: ${errorMessage}`) + + let compactErrorOutput: NormalizedBlockOutput | undefined + if (executionResult && Object.hasOwn(executionResult, 'output')) { + try { + compactErrorOutput = await compactServiceOutput( + executionResult.output, + compactionContext + ) + } catch (compactError) { + if ( + compactError instanceof PayloadSizeLimitError && + rejectLargeInlineOutput && + compactError.label === 'Workflow execution response' + ) { + return failure({ + kind: 'output_too_large', + message: 'Workflow execution response exceeds maximum size', + statusCode: 413, + code: 'workflow_response_too_large', + executionId, + }) + } + throw compactError + } + } - if (isRequestAborted() && !timeoutController.isTimedOut()) { - reqLogger.info('Execution aborted after client disconnect') return { ok: true, executionId, workflowId, - status: 'cancelled', - aborted: 'client', - output: undefined, - error: { message: 'Client cancelled request', code: 'CANCELLED' }, + status: 'failed', + aborted: null, + output: compactErrorOutput, + error: classifyExecutionError(error, executionResult), resolvedSecretTraceProvenance: executionResult?.executionState?.resolvedSecretTraceProvenance, hasResponseBlock: false, + startedAt: executionResult?.metadata?.startTime, + endedAt: executionResult?.metadata?.endTime, + durationMs: executionResult?.metadata?.duration, } + } finally { + abortSignal?.removeEventListener('abort', abortFromRequest) + timeoutController.cleanup() } + } - if ( - error instanceof PayloadSizeLimitError && - rejectLargeInlineOutput && - error.label === 'Workflow execution response' - ) { - return failure({ - kind: 'output_too_large', - message: 'Workflow execution response exceeds maximum size', - statusCode: 413, - code: 'workflow_response_too_large', - executionId, - }) - } - - reqLogger.error(`Execution failed: ${errorMessage}`) - - let compactErrorOutput: NormalizedBlockOutput | undefined - if (executionResult && Object.hasOwn(executionResult, 'output')) { - try { - compactErrorOutput = await compactServiceOutput(executionResult.output, compactionContext) - } catch (compactError) { - if ( - compactError instanceof PayloadSizeLimitError && - rejectLargeInlineOutput && - compactError.label === 'Workflow execution response' - ) { - return failure({ - kind: 'output_too_large', - message: 'Workflow execution response exceeds maximum size', - statusCode: 413, - code: 'workflow_response_too_large', - executionId, - }) - } - throw compactError - } - } - + const pending = runSynchronousWorkflow() + if (mode === 'sync-result-stream') { + /** Keep the claim until the pending run can prove whether durable ownership exists. */ + executionIdClaimTransferred = true return { ok: true, executionId, - workflowId, - status: 'failed', - aborted: null, - output: compactErrorOutput, - error: classifyExecutionError(error, executionResult), - resolvedSecretTraceProvenance: - executionResult?.executionState?.resolvedSecretTraceProvenance, - hasResponseBlock: false, - startedAt: executionResult?.metadata?.startTime, - endedAt: executionResult?.metadata?.endTime, - durationMs: executionResult?.metadata?.duration, + pending: pending.finally(settleExecutionIdClaim), + cancel: abortFromRequest, } - } finally { - abortSignal?.removeEventListener('abort', abortFromRequest) - timeoutController.cleanup() } + + return await pending } catch (error) { reqLogger.error('Failed to start workflow execution', { error: toError(error).message }) if (executionId) await releaseExecutionSlot(executionId) @@ -816,28 +901,7 @@ export async function executeWorkflowService( statusCode: 500, }) } finally { - if (executionIdClaim && !executionIdClaimCommitted) { - try { - executionIdClaimCommitted = await hasDurableExecutionOwner(executionId) - } catch (error) { - executionIdClaimCommitted = true - reqLogger.warn('Unable to verify execution ID ownership; retaining claim', { - error: toError(error).message, - executionId, - }) - } - } - - if (executionIdClaim && !executionIdClaimCommitted) { - try { - await releaseExecutionIdClaim(executionIdClaim) - } catch (error) { - reqLogger.warn('Failed to release pre-start execution ID claim', { - error: toError(error).message, - executionId, - }) - } - } + if (!executionIdClaimTransferred) await settleExecutionIdClaim() } } diff --git a/apps/sim/lib/workflows/streaming/streaming.test.ts b/apps/sim/lib/workflows/streaming/streaming.test.ts index 1400be73a1a..b695e4757c2 100644 --- a/apps/sim/lib/workflows/streaming/streaming.test.ts +++ b/apps/sim/lib/workflows/streaming/streaming.test.ts @@ -9,6 +9,7 @@ import { agentStreamProtocolResponseHeaders, createStreamingResponse, } from '@/lib/workflows/streaming/streaming' +import type { ExecutionResult } from '@/executor/types' import type { AgentStreamSink } from '@/providers/stream-events' const workflowStreamingLoggerCallIndex = loggerMock.createLogger.mock.calls.findIndex( @@ -102,6 +103,39 @@ describe('createStreamingResponse', () => { clearLargeValueCacheForTests() }) + it('emits an immediate keepalive and repeats it while execution is silent', async () => { + vi.useFakeTimers() + let finishExecution!: (result: ExecutionResult) => void + try { + const stream = await createStreamingResponse({ + requestId: 'request-keepalive', + executionId: 'execution-1', + streamConfig: {}, + executeFn: async () => + await new Promise((resolve) => { + finishExecution = resolve + }), + }) + const reader = stream.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({ + success: true, + status: 'completed', + output: {}, + logs: [], + metadata: { duration: 1 }, + }) + while (!(await reader.read()).done) {} + } finally { + vi.useRealTimers() + } + }) + it('forwards raw execution state to terminal logging', async () => { const safeComplete = vi.fn().mockResolvedValue(undefined) const executionState = { diff --git a/apps/sim/lib/workflows/streaming/streaming.ts b/apps/sim/lib/workflows/streaming/streaming.ts index 7c57eb35586..cf508bf34d1 100644 --- a/apps/sim/lib/workflows/streaming/streaming.ts +++ b/apps/sim/lib/workflows/streaming/streaming.ts @@ -8,7 +8,7 @@ import { extractPathFromOutputId, parseOutputContentSafely, } from '@/lib/core/utils/response-format' -import { encodeSSE } from '@/lib/core/utils/sse' +import { encodeSSE, encodeSSEComment } from '@/lib/core/utils/sse' import { getInlineJsonByteLength, materializeInlineExecutionValue, @@ -47,6 +47,7 @@ import { DEFAULT_MAX_THINKING_CHARS } from '@/providers/stream-pump' const logger = createLogger('WorkflowStreaming') const DANGEROUS_KEYS = ['__proto__', 'constructor', 'prototype'] +const STREAM_KEEPALIVE_INTERVAL_MS = 15_000 const SELECTED_OUTPUT_TOO_LARGE_MESSAGE = 'Selected output is too large to inline; select a nested field or use pagination/preview.' @@ -531,8 +532,26 @@ export async function createStreamingResponse( options.requestSignal?.removeEventListener('abort', onRequestAbort) } + let keepaliveId: ReturnType | undefined + const stopKeepalive = () => { + if (keepaliveId) { + clearInterval(keepaliveId) + keepaliveId = undefined + } + } + return new ReadableStream({ async start(controller) { + /** Flush headers promptly and keep silent blocks alive through idle-limited proxies. */ + controller.enqueue(encodeSSEComment('keepalive')) + keepaliveId = setInterval(() => { + try { + controller.enqueue(encodeSSEComment('keepalive')) + } catch { + stopKeepalive() + } + }, STREAM_KEEPALIVE_INTERVAL_MS) + const state: StreamingState = { streamedChunks: new Map(), processedOutputs: new Set(), @@ -930,6 +949,7 @@ export async function createStreamingResponse( controller.close() } finally { + stopKeepalive() cleanupRequestAbort() timeoutController.cleanup() } @@ -940,6 +960,7 @@ export async function createStreamingResponse( projectResolvedSecretDiagnosticError(reason, undefined) ) requestAborted = true + stopKeepalive() timeoutController.abort() cleanupRequestAbort() timeoutController.cleanup() diff --git a/helm/sim/examples/values-aws.yaml b/helm/sim/examples/values-aws.yaml index 011b9d92f0a..8f0c907270b 100644 --- a/helm/sim/examples/values-aws.yaml +++ b/helm/sim/examples/values-aws.yaml @@ -249,6 +249,7 @@ ingress: alb.ingress.kubernetes.io/target-type: ip alb.ingress.kubernetes.io/ssl-redirect: "443" alb.ingress.kubernetes.io/certificate-arn: "arn:aws:acm:us-west-2:123456789012:certificate/your-cert-arn" + alb.ingress.kubernetes.io/load-balancer-attributes: idle_timeout.timeout_seconds=3600 # Main application app: diff --git a/packages/sim-cli/src/commands/protocol/chat.ts b/packages/sim-cli/src/commands/protocol/chat.ts index a27139031fc..dcb0caabbb9 100644 --- a/packages/sim-cli/src/commands/protocol/chat.ts +++ b/packages/sim-cli/src/commands/protocol/chat.ts @@ -3,6 +3,7 @@ import type { Command } from 'commander' import { clientFrom } from '../../context' import { type ChatResponse, V2_OPERATIONS } from '../../generated/v2-api' import { SimApiError } from '../../http/client' +import { readNdjson } from '../../http/ndjson' import { sanitize } from '../../output/render' import { printProtocolResult } from './result' @@ -25,17 +26,6 @@ interface ChatOptions { conversation?: string } -function parseChatStreamLine(line: string): ChatStreamEvent | undefined { - const trimmed = line.trim() - if (!trimmed) return undefined - - try { - return JSON.parse(trimmed) as ChatStreamEvent - } catch { - throw new SimApiError('Chat stream returned malformed data', 0) - } -} - /** * Consumes the chat NDJSON stream to its final payload. * @@ -50,23 +40,16 @@ async function readChatStream( response: Response, onChunk: (content: string) => void ): Promise { - if (!response.body) { - throw new SimApiError('Chat stream ended without a response body', 0) - } - - const reader = response.body.getReader() - const decoder = new TextDecoder() - let buffer = '' - let finalResult: ChatResult | undefined - - /** Reports whether the line ended the turn, so reading can stop there. */ - const processLine = (line: string): boolean => { - const event = parseChatStreamLine(line) - if (!event || event.type === 'heartbeat') return false + for await (const value of readNdjson(response.body, 'Chat stream')) { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new SimApiError('Chat stream returned an unknown event', 0) + } + const event = value as ChatStreamEvent + if (event.type === 'heartbeat') continue if (event.type === 'chunk') { if (event.content) onChunk(sanitize(event.content)) - return false + continue } if (event.type === 'error') { @@ -74,43 +57,12 @@ async function readChatStream( } if (event.type === 'final') { - finalResult = event.data - return true + return event.data } throw new SimApiError('Chat stream returned an unknown event', 0) } - - try { - let ended = false - while (!ended) { - const { done, value } = await reader.read() - if (done) break - - buffer += decoder.decode(value, { stream: true }) - const lines = buffer.split('\n') - buffer = lines.pop() ?? '' - for (const line of lines) { - if (processLine(line)) { - ended = true - break - } - } - } - - if (!ended) { - buffer += decoder.decode() - processLine(buffer) - } - - if (!finalResult) { - throw new SimApiError('Chat stream ended without a final result', 0) - } - - return finalResult - } finally { - void reader.cancel().catch(() => undefined) - } + throw new SimApiError('Chat stream ended without a final result', 0) } /** diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts b/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts index f0220dab97c..39e06ac7d2d 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts +++ b/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts @@ -62,6 +62,20 @@ function streamResponse(body: ReadableStream): Response { } as unknown as Response } +function jsonResponse(data: Record): Response { + return new Response(JSON.stringify({ data }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) +} + +function ndjsonResponse(...events: Array>): Response { + return new Response(bodyOf(events.map((event) => `${JSON.stringify(event)}\n`)), { + status: 200, + headers: { 'content-type': 'application/x-ndjson; charset=utf-8' }, + }) +} + function options(overrides: Partial[1]> = {}) { return { includeThinking: false, includeToolCalls: false, stderr: writer(), ...overrides } } @@ -70,6 +84,15 @@ beforeEach(() => { output.format = 'json' request.mockReset() requestRaw.mockReset() + requestRaw.mockResolvedValue( + jsonResponse({ + runId: 'run-1', + workflowId: WORKFLOW_ID, + status: 'completed', + output: {}, + error: null, + }) + ) }) afterEach(() => { @@ -322,19 +345,31 @@ describe('sim workflows run --follow', () => { }) }) - it('leaves the generated non-streaming path untouched', async () => { - request.mockResolvedValue({ data: { success: true, output: {} } }) + it('negotiates a heartbeat result stream for an ordinary sync run', async () => { vi.spyOn(console, 'log').mockImplementation(() => {}) await run(WORKFLOW_ID, '--input', '{"topic":"otters"}') - expect(requestRaw).not.toHaveBeenCalled() + expect(request).not.toHaveBeenCalled() + expect(requestRaw).toHaveBeenCalledTimes(1) + expect(requestRaw.mock.calls[0][1]).toMatchObject({ + body: { input: { topic: 'otters' } }, + headers: { accept: 'application/x-ndjson' }, + }) + }) + + 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(() => {}) + + await run(WORKFLOW_ID, '--async') + expect(request).toHaveBeenCalledTimes(1) - expect(request.mock.calls[0][1].body).toEqual({ input: { topic: 'otters' } }) + expect(requestRaw).not.toHaveBeenCalled() + expect(request.mock.calls[0][1].body).toEqual({ async: true }) }) it('projects manual trigger flags into one nested run selector', async () => { - request.mockResolvedValue({ data: { success: true, output: {} } }) vi.spyOn(console, 'log').mockImplementation(() => {}) await run( @@ -346,19 +381,18 @@ describe('sim workflows run --follow', () => { '{"event":"created"}' ) - expect(request.mock.calls[0][1].body).toEqual({ + expect(requestRaw.mock.calls[0][1].body).toEqual({ input: { event: 'created' }, run: { source: 'manual', entry: { type: 'trigger', blockId: 'slack-trigger' } }, }) }) it('lets --from-block imply manual and requires an exact source run', async () => { - request.mockResolvedValue({ data: { success: true, output: {} } }) vi.spyOn(console, 'log').mockImplementation(() => {}) await run(WORKFLOW_ID, '--from-block', 'agent-1', '--source-run', 'run-1') - expect(request.mock.calls[0][1].body).toEqual({ + expect(requestRaw.mock.calls[0][1].body).toEqual({ run: { source: 'manual', entry: { type: 'block', blockId: 'agent-1', sourceRunId: 'run-1' }, @@ -379,6 +413,77 @@ describe('sim workflows run --follow', () => { }) }) + it('consumes heartbeats and prints the final NDJSON result', async () => { + requestRaw.mockResolvedValue( + ndjsonResponse( + { type: 'heartbeat', timestamp: '2026-09-07T00:00:00.000Z' }, + { + type: 'final', + data: { + runId: 'run-1', + workflowId: WORKFLOW_ID, + status: 'completed', + output: { answer: 42 }, + error: null, + }, + } + ) + ) + const stdout = vi.spyOn(console, 'log').mockImplementation(() => {}) + + await run(WORKFLOW_ID, '--manual') + + expect(JSON.parse(String(stdout.mock.calls[0][0]))).toMatchObject({ + runId: 'run-1', + status: 'completed', + output: { answer: 42 }, + }) + }) + + it('accepts an older server JSON response without a content type', async () => { + requestRaw.mockResolvedValue( + new Response( + JSON.stringify({ + data: { + runId: 'run-1', + workflowId: WORKFLOW_ID, + status: 'completed', + output: { answer: 42 }, + error: null, + }, + }), + { status: 200 } + ) + ) + const stdout = vi.spyOn(console, 'log').mockImplementation(() => {}) + + await run(WORKFLOW_ID) + + expect(JSON.parse(String(stdout.mock.calls[0][0]))).toMatchObject({ + runId: 'run-1', + status: 'completed', + }) + }) + + it('prints then fails for a failed NDJSON run just like the JSON path', async () => { + requestRaw.mockResolvedValue( + ndjsonResponse({ + type: 'final', + data: { + runId: 'run-1', + workflowId: WORKFLOW_ID, + status: 'failed', + output: { partial: true }, + error: { message: 'Agent failed', code: 'BLOCK_EXECUTION_FAILED' }, + }, + }) + ) + const stdout = vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect(run(WORKFLOW_ID)).rejects.toThrow('Agent failed') + expect(stdout).toHaveBeenCalled() + }) + it('fails fast on invalid manual flag combinations', async () => { await expect(run(WORKFLOW_ID, '--trigger', 'trigger-1')).rejects.toThrow(/require --manual/) await expect(run(WORKFLOW_ID, '--from-block', 'agent-1')).rejects.toThrow( diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts index 256e0567f75..ddf45f7947c 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts +++ b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts @@ -4,8 +4,9 @@ import { clientFrom } from '../../context' import { CLI_CONTRACT } from '../../contract/commands' import { V2_OPERATIONS } from '../../generated/v2-api' import { SimApiError } from '../../http/client' +import { readNdjson } from '../../http/ndjson' import { safeOneLine, sanitize } from '../../output/render' -import { executeOperation } from '../../runtime/execute' +import { executeOperation, runFailureMessage } from '../../runtime/execute' import { buildRequest } from '../../runtime/request' import { renderResult } from '../../runtime/result' import type { OperationSpec } from '../../runtime/types' @@ -23,6 +24,7 @@ import type { OperationSpec } from '../../runtime/types' */ const AGENT_STREAM_PROTOCOL_HEADER = 'x-sim-stream-protocol' const AGENT_STREAM_PROTOCOL_V1 = 'agent-events-v1' +const WORKFLOW_RESULT_STREAM_CONTENT_TYPE = 'application/x-ndjson' /** Terminal marker. Sent JSON-encoded, so the raw payload carries its quotes. */ const DONE_SENTINEL = '[DONE]' @@ -105,6 +107,70 @@ function stringField(frame: Record, key: string): string | null return typeof value === 'string' ? value : null } +/** + * Reads the heartbeat-delimited result transport used for long synchronous + * runs. A server that predates the transport ignores `Accept` and returns its + * ordinary JSON envelope, which is consumed without retrying the run. + */ +async function readWorkflowResult(response: Response): Promise> { + const contentType = (response.headers.get('content-type') ?? '').toLowerCase() + if (!contentType.includes(WORKFLOW_RESULT_STREAM_CONTENT_TYPE)) { + let envelope: unknown + try { + envelope = await response.json() + } catch { + throw new SimApiError( + `Workflow run returned malformed JSON${contentType ? ` as ${contentType}` : ''}`, + response.status + ) + } + if (!isRecord(envelope)) { + throw new SimApiError('Workflow run returned an invalid result envelope', response.status) + } + return isRecord(envelope.data) ? envelope.data : envelope + } + + for await (const value of readNdjson(response.body, 'Workflow result stream')) { + if (!isRecord(value) || typeof value.type !== 'string') { + throw new SimApiError('Workflow result stream returned an unknown event', response.status) + } + if (value.type === 'heartbeat') continue + if (value.type === 'error') { + throw new SimApiError( + safeOneLine(typeof value.error === 'string' ? value.error : 'Workflow run failed'), + typeof value.status === 'number' ? value.status : 0, + typeof value.code === 'string' ? value.code : null + ) + } + if (value.type === 'final' && isRecord(value.data)) return value.data + throw new SimApiError('Workflow result stream returned an unknown event', response.status) + } + + throw new SimApiError('Workflow result stream ended without a final result', response.status) +} + +/** Runs synchronously while keeping idle-limited HTTP paths active. */ +async function runWithResultStream(workflowId: string, command: Command): Promise { + const flags = command.optsWithGlobals() as Record + 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) + + renderResult('executeWorkflow', profile.output, payload, CLI_CONTRACT.executeWorkflow ?? {}, { + expandedTrace: flags.trace === true, + }) + + const failure = runFailureMessage('executeWorkflow', payload) + if (failure) throw new SimApiError(failure, 0) +} + /** * Yields the payload of every `data:` line in an SSE body. * @@ -361,8 +427,11 @@ function followOrDelegate(previous: ((args: unknown[]) => unknown) | null) { 0 ) } - // Whatever was installed before wins, so a second augmentation of the - // same leaf composes with this one instead of replacing it. + if (flags.async !== true) { + await runWithResultStream(workflowId, command) + return + } + if (previous) { await previous(command.processedArgs) return @@ -389,8 +458,8 @@ function followOrDelegate(previous: ((args: unknown[]) => unknown) | null) { * would have to restate every one of them and then drift. * * Commander offers no way to read the action it already holds, so the existing - * handler is captured and delegated to — every non-`--follow` invocation still - * runs the generated path byte for byte. + * handler is captured for async execution. Synchronous execution uses the same + * generated request and result builders with a heartbeat-capable response. */ export function attachWorkflowRunFollow(workflows: Command): void { const run = workflows.commands.find((command) => command.name() === 'run') diff --git a/packages/sim-cli/src/http/client.test.ts b/packages/sim-cli/src/http/client.test.ts index ace92c62804..24fcf8d552e 100644 --- a/packages/sim-cli/src/http/client.test.ts +++ b/packages/sim-cli/src/http/client.test.ts @@ -321,6 +321,20 @@ describe('non-JSON responses', () => { }) describe('a request that never answers', () => { + it('reports the nested Undici reason behind fetch failed', async () => { + const socketError = Object.assign(new Error('other side closed'), { code: 'UND_ERR_SOCKET' }) + vi.stubGlobal( + 'fetch', + vi.fn().mockRejectedValue(new TypeError('fetch failed', { cause: socketError })) + ) + + await expect(client().request('/api/v2/workflows')).rejects.toMatchObject({ + message: + 'Could not reach https://sim.example: fetch failed: other side closed (UND_ERR_SOCKET)', + status: 0, + }) + }) + it('bounds a request by default, above every timeout the server itself applies', async () => { // A synchronous workflow run is allowed 3000s on a paid plan, so a tighter // default would abort real work and report it as a transport failure. What diff --git a/packages/sim-cli/src/http/client.ts b/packages/sim-cli/src/http/client.ts index 509405c0349..d710dda5311 100644 --- a/packages/sim-cli/src/http/client.ts +++ b/packages/sim-cli/src/http/client.ts @@ -187,6 +187,31 @@ function truncate(value: string, max: number): string { return value.length <= max ? value : `${value.slice(0, max)}…` } +/** + * Keeps the useful nested reason from Node/Undici transport failures without + * serializing request options, headers, socket objects, or credentials. + */ +function transportErrorMessage(error: unknown): string { + const messages: string[] = [] + const seen = new Set() + let current: unknown = error + + while (current && typeof current === 'object' && messages.length < 4 && !seen.has(current)) { + seen.add(current) + const candidate = current as { message?: unknown; code?: unknown; cause?: unknown } + const message = + typeof candidate.message === 'string' + ? truncate(candidate.message.replace(/\s+/g, ' ').trim(), 300) + : '' + const code = typeof candidate.code === 'string' ? candidate.code : '' + const detail = `${message}${code && !message.includes(code) ? ` (${code})` : ''}` + if (detail && messages.at(-1) !== detail) messages.push(detail) + current = candidate.cause + } + + return messages.join(': ') || 'Unknown network error' +} + /** * Whether this is the refusal a workspace-scoped key gets from an operation only * a personal key may perform, under either code that expresses it. @@ -600,7 +625,7 @@ export class SimClient { ) } throw new SimApiError( - `Could not reach ${this.profile.endpoint}: ${(cause as Error).message}`, + `Could not reach ${this.profile.endpoint}: ${transportErrorMessage(cause)}`, 0 ) } diff --git a/packages/sim-cli/src/http/ndjson.ts b/packages/sim-cli/src/http/ndjson.ts new file mode 100644 index 00000000000..51e96a79f68 --- /dev/null +++ b/packages/sim-cli/src/http/ndjson.ts @@ -0,0 +1,47 @@ +import { SimApiError } from './client' + +/** + * Parses a newline-delimited JSON response incrementally and releases its + * reader when the consumer reaches a terminal event or stops early. + */ +export async function* readNdjson( + body: ReadableStream | null, + protocol: string +): AsyncGenerator { + if (!body) { + throw new SimApiError(`${protocol} ended without a response body`, 0) + } + + const reader = body.getReader() + const decoder = new TextDecoder() + let buffer = '' + + const parse = (line: string): unknown => { + const trimmed = line.trim() + if (!trimmed) return undefined + try { + return JSON.parse(trimmed) + } catch { + throw new SimApiError(`${protocol} returned malformed data`, 0) + } + } + + try { + while (true) { + const { done, value } = await reader.read() + buffer += done ? decoder.decode() : decoder.decode(value, { stream: true }) + + const lines = buffer.split('\n') + buffer = done ? '' : (lines.pop() ?? '') + for (const line of lines) { + const event = parse(line) + if (event !== undefined) yield event + } + + if (done) return + } + } finally { + void reader.cancel().catch(() => undefined) + reader.releaseLock() + } +} diff --git a/packages/sim-cli/src/runtime/execute.ts b/packages/sim-cli/src/runtime/execute.ts index 77b652c4346..db3f2a864eb 100644 --- a/packages/sim-cli/src/runtime/execute.ts +++ b/packages/sim-cli/src/runtime/execute.ts @@ -59,7 +59,7 @@ const RUN_OUTCOME_OPERATIONS: Readonly< } /** The one-line explanation of an in-band run failure, or `null` if there is none. */ -function runFailureMessage(operation: V2OperationName, payload: unknown): string | null { +export function runFailureMessage(operation: V2OperationName, payload: unknown): string | null { const failureMessages = RUN_OUTCOME_OPERATIONS[operation] if (!failureMessages) return null if (!payload || typeof payload !== 'object' || Array.isArray(payload)) return null diff --git a/scripts/openapi/documents.test.ts b/scripts/openapi/documents.test.ts index bdba2062cd2..483c0b6bee4 100644 --- a/scripts/openapi/documents.test.ts +++ b/scripts/openapi/documents.test.ts @@ -325,7 +325,11 @@ describe('generated OpenAPI documents', () => { ]) expect(execute.tags).toEqual(['Workflows']) expect(execute.security).toEqual([{ apiKey: [] }, { oauthBearer: [] }, {}]) - expect(Object.keys(executeOkContent).sort()).toEqual(['application/json', 'text/event-stream']) + expect(Object.keys(executeOkContent).sort()).toEqual([ + 'application/json', + 'application/x-ndjson', + 'text/event-stream', + ]) expect(Object.keys(executeQueuedContent)).toEqual(['application/json']) const resume = getOperation(spec, '/api/v2/workflows/{workflowId}/runs/{runId}/resume', 'post') From 1304b6ddf20e45606259dd4731d7008eedbbe3b5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 7 Sep 2026 18:27:45 -0700 Subject: [PATCH 2/4] chore(helm): bump chart version --- helm/sim/Chart.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helm/sim/Chart.yaml b/helm/sim/Chart.yaml index 8781e11ba72..bc45e6a3cb2 100644 --- a/helm/sim/Chart.yaml +++ b/helm/sim/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: v2 name: sim description: A Helm chart for Sim - the open-source AI workspace where teams build, deploy, and manage AI agents type: application -version: 1.10.0 +version: 1.10.1 appVersion: "v0.8.24" kubeVersion: ">=1.25.0-0" home: https://sim.ai From 9f9c93c69dfc2a6fe7a49a4fff88244825627a62 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 7 Sep 2026 18:53:07 -0700 Subject: [PATCH 3/4] fix(workflows): preserve stream compatibility --- .../api/mcp/serve/[serverId]/route.test.ts | 29 +++++++- .../sim/app/api/mcp/serve/[serverId]/route.ts | 5 +- apps/sim/app/api/mothership/execute/route.ts | 3 +- apps/sim/app/api/v2/chat/route.ts | 3 +- .../[workflowId]/execute/route.test.ts | 12 ++++ .../workflows/[workflowId]/execute/route.ts | 3 +- apps/sim/lib/core/utils/media-types.test.ts | 27 ++++++++ apps/sim/lib/core/utils/media-types.ts | 66 +++++++++++++++++++ .../protocol/workflow-run-follow.test.ts | 10 +++ .../commands/protocol/workflow-run-follow.ts | 35 ++++++---- 10 files changed, 173 insertions(+), 20 deletions(-) create mode 100644 apps/sim/lib/core/utils/media-types.test.ts create mode 100644 apps/sim/lib/core/utils/media-types.ts diff --git a/apps/sim/app/api/mcp/serve/[serverId]/route.test.ts b/apps/sim/app/api/mcp/serve/[serverId]/route.test.ts index 6b60e88531b..5a95ce3523d 100644 --- a/apps/sim/app/api/mcp/serve/[serverId]/route.test.ts +++ b/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([ diff --git a/apps/sim/app/api/mcp/serve/[serverId]/route.ts b/apps/sim/app/api/mcp/serve/[serverId]/route.ts index bf1b17a81bd..007de1cf4c5 100644 --- a/apps/sim/app/api/mcp/serve/[serverId]/route.ts +++ b/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() } diff --git a/apps/sim/app/api/mothership/execute/route.ts b/apps/sim/app/api/mothership/execute/route.ts index e01f727f591..4c2c1093142 100644 --- a/apps/sim/app/api/mothership/execute/route.ts +++ b/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) ) } diff --git a/apps/sim/app/api/v2/chat/route.ts b/apps/sim/app/api/v2/chat/route.ts index f4f09bbaaaa..0740600766f 100644 --- a/apps/sim/app/api/v2/chat/route.ts +++ b/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) ) } diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts index bae20236ba0..323a43fd62d 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts +++ b/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 diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts index 426294c3dee..db03a7bd2db 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts +++ b/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 { diff --git a/apps/sim/lib/core/utils/media-types.test.ts b/apps/sim/lib/core/utils/media-types.test.ts new file mode 100644 index 00000000000..6d317a858f5 --- /dev/null +++ b/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) + }) +}) diff --git a/apps/sim/lib/core/utils/media-types.ts b/apps/sim/lib/core/utils/media-types.ts new file mode 100644 index 00000000000..69abb6f7d57 --- /dev/null +++ b/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 + }) +} diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts b/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts index 39e06ac7d2d..1e74f4dd4c0 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts +++ b/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(() => {}) diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts index ddf45f7947c..67566760bc0 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts +++ b/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 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) + } } /** From f13696fe595a9d15eea1d04e623cae03c7515e21 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 7 Sep 2026 19:00:06 -0700 Subject: [PATCH 4/4] refactor(streaming): import SSE helpers directly --- apps/sim/app/api/copilot/chat/stream/route.test.ts | 1 - apps/sim/app/api/copilot/chat/stream/route.ts | 2 +- apps/sim/lib/copilot/request/session/index.ts | 2 +- apps/sim/lib/copilot/request/session/sse.ts | 2 -- apps/sim/lib/copilot/request/session/writer.ts | 3 ++- 5 files changed, 4 insertions(+), 6 deletions(-) diff --git a/apps/sim/app/api/copilot/chat/stream/route.test.ts b/apps/sim/app/api/copilot/chat/stream/route.test.ts index aa7c85b250f..24d7a99cfef 100644 --- a/apps/sim/app/api/copilot/chat/stream/route.test.ts +++ b/apps/sim/app/api/copilot/chat/stream/route.test.ts @@ -38,7 +38,6 @@ vi.mock('@/lib/copilot/request/session', () => ({ }), encodeSSEEnvelope: (event: Record) => 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', }, diff --git a/apps/sim/app/api/copilot/chat/stream/route.ts b/apps/sim/app/api/copilot/chat/stream/route.ts index bd7f5465685..47b1f65c79c 100644 --- a/apps/sim/app/api/copilot/chat/stream/route.ts +++ b/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 diff --git a/apps/sim/lib/copilot/request/session/index.ts b/apps/sim/lib/copilot/request/session/index.ts index c0ecb7a1716..8e62b70bb2e 100644 --- a/apps/sim/lib/copilot/request/session/index.ts +++ b/apps/sim/lib/copilot/request/session/index.ts @@ -71,6 +71,6 @@ export { isFilePreviewSession, } from './file-preview-session-contract' export { checkForReplayGap, type ReplayGapResult } from './recovery' -export { encodeSSEComment, encodeSSEEnvelope, SSE_RESPONSE_HEADERS } from './sse' +export { encodeSSEEnvelope, SSE_RESPONSE_HEADERS } from './sse' export type { StreamBatchEvent } from './types' export { StreamWriter, type StreamWriterOptions } from './writer' diff --git a/apps/sim/lib/copilot/request/session/sse.ts b/apps/sim/lib/copilot/request/session/sse.ts index c19bd553fd1..99b8b58fe12 100644 --- a/apps/sim/lib/copilot/request/session/sse.ts +++ b/apps/sim/lib/copilot/request/session/sse.ts @@ -1,7 +1,5 @@ import { SSE_HEADERS } from '@/lib/core/utils/sse' -export { encodeSSEComment } from '@/lib/core/utils/sse' - const encoder = new TextEncoder() export function encodeSSEEnvelope(envelope: unknown): Uint8Array { diff --git a/apps/sim/lib/copilot/request/session/writer.ts b/apps/sim/lib/copilot/request/session/writer.ts index 8699b790c71..a56502066cc 100644 --- a/apps/sim/lib/copilot/request/session/writer.ts +++ b/apps/sim/lib/copilot/request/session/writer.ts @@ -1,10 +1,11 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { MothershipStreamV1EventType } from '@/lib/copilot/generated/mothership-stream-v1' +import { encodeSSEComment } from '@/lib/core/utils/sse' import { appendEvents } from './buffer' import type { PersistedStreamEventEnvelope } from './contract' import { createEvent } from './event' -import { encodeSSEComment, encodeSSEEnvelope } from './sse' +import { encodeSSEEnvelope } from './sse' import type { StreamEvent } from './types' const logger = createLogger('StreamWriter')