diff --git a/apps/docs/content/docs/cli/reference.mdx b/apps/docs/content/docs/cli/reference.mdx index 5e6b7136d8e..df99c0861bf 100644 --- a/apps/docs/content/docs/cli/reference.mdx +++ b/apps/docs/content/docs/cli/reference.mdx @@ -7911,7 +7911,7 @@ sim workflows run [options] | --- | --- | --- | | `--input ` | No | Trigger input as JSON (JSON, or @path / @- to read a file or stdin). | | `--async` | No | Queue the run and return immediately. Defaults to false. | -| `--execution-timeout-seconds ` | No | Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true`; otherwise returns `400`. | +| `--execution-timeout-seconds ` | No | Maximum duration of the run in seconds, capped by the plan's execution timeout. Requires --async or --wait-timeout. | | `--select-output ` | No | Return blockName.path values (e.g. agent_1.content), or childWorkflowId.blockName.path for a child workflow (applies to every invocation) — in blockOutputs on a sync run, or from the streamed result with --follow; missing paths are omitted. Not available with --async (space-separated, or @path / @- with one value per line; @@value for a literal leading @). | | `--include-file-base64` | No | Inline eligible output files as base64 content. Defaults to true for the installed CLI; Chat defaults to file references and admits requested bytes only for synchronous runs. Rejected when async is true. | | `--no-include-file-base64` | No | Send --include-file-base64 as false. | @@ -7923,6 +7923,7 @@ sim workflows run [options] | `--from-block ` | No | Run manually from this saved workflow block. | | `--source-run ` | No | Prior run whose persisted state supplies upstream outputs (requires --from-block). | | `--stop-after ` | No | Stop the run after this saved block, failing it if the run takes a path that skips the block; with --from-block on the same block, re-runs only that block (implies --manual). | +| `--wait-timeout ` | No | Wait at most this many seconds for a synchronous run (0 waits until it settles). A run still going then continues in the background under async limits: the command prints its run record and exits 4, and `sim workflows runs wait` keeps waiting. Not with --async or --follow. | | `--follow` | No | Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns. | | `--include-thinking` | No | Show model reasoning while following (requires --follow). | | `--include-tool-calls` | No | Show tool calls while following (requires --follow). | diff --git a/apps/docs/content/docs/cli/workflows.mdx b/apps/docs/content/docs/cli/workflows.mdx index 0e670a1d73f..2a3c7e338d6 100644 --- a/apps/docs/content/docs/cli/workflows.mdx +++ b/apps/docs/content/docs/cli/workflows.mdx @@ -695,7 +695,7 @@ sim workflows run [options] | --- | --- | --- | | `--input ` | No | Trigger input as JSON (JSON, or @path / @- to read a file or stdin). | | `--async` | No | Queue the run and return immediately. Defaults to false. | -| `--execution-timeout-seconds ` | No | Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true`; otherwise returns `400`. | +| `--execution-timeout-seconds ` | No | Maximum duration of the run in seconds, capped by the plan's execution timeout. Requires --async or --wait-timeout. | | `--select-output ` | No | Return blockName.path values (e.g. agent_1.content), or childWorkflowId.blockName.path for a child workflow (applies to every invocation) — in blockOutputs on a sync run, or from the streamed result with --follow; missing paths are omitted. Not available with --async (space-separated, or @path / @- with one value per line; @@value for a literal leading @). | | `--include-file-base64` | No | Inline eligible output files as base64 content. Defaults to true for the installed CLI; Chat defaults to file references and admits requested bytes only for synchronous runs. Rejected when async is true. | | `--no-include-file-base64` | No | Send --include-file-base64 as false. | @@ -707,6 +707,7 @@ sim workflows run [options] | `--from-block ` | No | Run manually from this saved workflow block. | | `--source-run ` | No | Prior run whose persisted state supplies upstream outputs (requires --from-block). | | `--stop-after ` | No | Stop the run after this saved block, failing it if the run takes a path that skips the block; with --from-block on the same block, re-runs only that block (implies --manual). | +| `--wait-timeout ` | No | Wait at most this many seconds for a synchronous run (0 waits until it settles). A run still going then continues in the background under async limits: the command prints its run record and exits 4, and `sim workflows runs wait` keeps waiting. Not with --async or --follow. | | `--follow` | No | Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns. | | `--include-thinking` | No | Show model reasoning while following (requires --follow). | | `--include-tool-calls` | No | Show tool calls while following (requires --follow). | diff --git a/apps/docs/openapi-v2-workflows.json b/apps/docs/openapi-v2-workflows.json index 30bc2c5dc8f..386196d06a8 100644 --- a/apps/docs/openapi-v2-workflows.json +++ b/apps/docs/openapi-v2-workflows.json @@ -3797,6 +3797,15 @@ }, "example5": { "summary": "Request example 5", + "value": { + "run": { + "source": "manual" + }, + "waitTimeoutSeconds": 240 + } + }, + "example6": { + "summary": "Request example 6", "value": { "run": { "source": "manual", @@ -3808,8 +3817,8 @@ } } }, - "example6": { - "summary": "Request example 6", + "example7": { + "summary": "Request example 7", "value": { "run": { "source": "manual", @@ -14245,7 +14254,7 @@ "required": ["runId", "statusUrl"], "additionalProperties": false, "title": "Queued workflow run", - "description": "Receipt returned when a workflow run is queued." + "description": "Receipt returned when a workflow run is queued, or is still running when `waitTimeoutSeconds` elapsed." }, "ExecuteWorkflowQueuedResponse": { "type": "object", @@ -14282,7 +14291,7 @@ } }, "run": { - "description": "Workflow state and entry point to execute. Omit for the active deployment. Manual execution requires OAuth or personal-key write access and supports synchronous or streamed runs only.", + "description": "Workflow state and entry point to execute. Omit for the active deployment. Manual execution requires OAuth or personal-key write access and supports synchronous or streamed runs only; `waitTimeoutSeconds` lets a synchronous one continue in the background.", "oneOf": [ { "type": "object", @@ -14369,11 +14378,17 @@ "type": "boolean" }, "executionTimeoutSeconds": { - "description": "Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true`; otherwise returns `400`.", + "description": "Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true` or `waitTimeoutSeconds`; otherwise returns `400`.", "type": "integer", "minimum": 1, "maximum": 604800 }, + "waitTimeoutSeconds": { + "description": "Wait at most this many seconds for a synchronous run; `0` waits until it settles. A run still going then continues in the background and the response carries its run receipt instead of its result: `202` on a JSON response, or a closing `detached` frame on an `application/x-ndjson` result stream, which keeps its `200`. The receipt is sent only once the run can be read on the runs resource, where it can also be cancelled. The run is admitted under asynchronous limits, and a dropped connection no longer cancels it. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.", + "type": "integer", + "minimum": 0, + "maximum": 604800 + }, "stream": { "default": false, "description": "Return Server-Sent Events instead of JSON when true. Cannot be combined with `async`.", @@ -14435,6 +14450,12 @@ "source": "manual" } }, + { + "run": { + "source": "manual" + }, + "waitTimeoutSeconds": 240 + }, { "run": { "source": "manual", 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 bef2b85433c..45e9ce4748f 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 @@ -4,6 +4,7 @@ import { executionPreprocessingMock, executionPreprocessingMockFns, loggingSessionMock, + loggingSessionMockFns, resetDbChainMock, setEnv, workflowAuthzMockFns, @@ -35,8 +36,11 @@ import { v2RouteMocks, } from '@sim/testing/mocks/v2-route.mock' import { workspaceAuthzMock, workspaceAuthzMockFns } from '@sim/testing/mocks/workspace-authz.mock' +import { NextRequest } from 'next/server' import { beforeEach, describe, expect, it, vi } from 'vitest' import { WorkspaceApiKeyAuthorizationError } from '@/lib/core/application' +import { getRemainingExecutionMs } from '@/lib/core/execution-limits' +import { abortManualExecution } from '@/lib/execution/manual-cancellation' const { mockClaimExecutionId, @@ -414,6 +418,7 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => { ok: true, executionId: 'execution-123', pending, + logged: Promise.resolve(), cancel: vi.fn(), }) @@ -1120,4 +1125,241 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => { }) expect(mockPreprocessExecution).not.toHaveBeenCalled() }) + describe('waitTimeoutSeconds', () => { + const settledCore = { + success: true, + output: { result: 'done' }, + metadata: { + duration: 42, + startTime: '2026-07-31T00:00:00.000Z', + endTime: '2026-07-31T00:00:01.000Z', + }, + } + const receipt = { + runId: 'execution-123', + statusUrl: 'http://localhost:3000/api/v2/workflows/workflow-1/runs/execution-123', + } + + /** A run the test settles by hand, recording the signal the executor was handed. */ + function holdRun() { + const held: { signal?: AbortSignal; finish: (result: unknown) => void } = { + finish: () => {}, + } + mockExecuteWorkflowCore.mockImplementationOnce( + ({ abortSignal }: { abortSignal: AbortSignal }) => + new Promise((resolve) => { + held.signal = abortSignal + held.finish = resolve + abortSignal.addEventListener('abort', () => + resolve({ success: false, status: 'cancelled', output: {}, metadata: {} }) + ) + }) + ) + return held + } + + function readLines(response: Response) { + if (!response.body) throw new Error('Expected NDJSON response body') + const reader = response.body.getReader() + const decoder = new TextDecoder() + return { + reader, + next: async () => { + const chunk = await reader.read() + return chunk.done ? null : JSON.parse(decoder.decode(chunk.value)) + }, + } + } + + it('answers with the ordinary result when the run settles within the wait', async () => { + const response = await callExecute( + { input: {}, waitTimeoutSeconds: 30 }, + { Accept: 'application/x-ndjson' } + ) + const lines = readLines(response) + + expect(await lines.next()).toMatchObject({ type: 'heartbeat' }) + expect(await lines.next()).toMatchObject({ + type: 'final', + data: { runId: 'execution-123', status: 'completed', output: { result: 'done' } }, + }) + expect(await lines.next()).toBeNull() + }) + + it('admits a detachable run as an async run with the async timeout', async () => { + const response = await callExecute({ input: {}, waitTimeoutSeconds: 30 }) + + expect(response.status).toBe(200) + expect((await response.json()).data).toMatchObject({ + runId: 'execution-123', + status: 'completed', + }) + expect(mockPreprocessExecution).toHaveBeenCalledWith( + expect.objectContaining({ rateLimitCounter: 'async', executionType: 'async' }) + ) + const { abortSignal } = mockExecuteWorkflowCore.mock.calls[0][0] + expect(getRemainingExecutionMs(abortSignal)).toBeGreaterThan(60_000) + }) + + it('detaches a run still going at the deadline and lets it finish in the background', async () => { + vi.useFakeTimers() + try { + const held = holdRun() + const response = await callExecute( + { input: {}, waitTimeoutSeconds: 20 }, + { Accept: 'application/x-ndjson' } + ) + const lines = readLines(response) + expect(await lines.next()).toMatchObject({ type: 'heartbeat' }) + + await vi.advanceTimersByTimeAsync(20_000) + const frames = [] + for (let frame = await lines.next(); frame; frame = await lines.next()) frames.push(frame) + expect(frames.filter((frame) => frame.type !== 'heartbeat')).toEqual([ + { type: 'detached', data: receipt }, + ]) + + expect(held.signal?.aborted).toBe(false) + expect(mockAdmissionRelease).not.toHaveBeenCalled() + expect(mockReleaseExecutionIdClaim).not.toHaveBeenCalled() + + held.finish(settledCore) + await vi.advanceTimersByTimeAsync(0) + expect(mockAdmissionRelease).toHaveBeenCalledTimes(1) + } finally { + vi.useRealTimers() + } + }) + + it('withholds the receipt until the run can be found by its id', async () => { + vi.useFakeTimers() + try { + let logStarted!: () => void + loggingSessionMockFns.mockWhenLogStarted.mockReturnValueOnce( + new Promise((resolve) => { + logStarted = resolve + }) + ) + const held = holdRun() + const pending = Promise.resolve(callExecute({ input: {}, waitTimeoutSeconds: 1 })) + await vi.advanceTimersByTimeAsync(5_000) + let answered = false + void pending.then(() => { + answered = true + }) + await vi.advanceTimersByTimeAsync(0) + expect(answered).toBe(false) + + logStarted() + const response = await pending + expect(response.status).toBe(202) + expect((await response.json()).data).toEqual(receipt) + held.finish(settledCore) + } finally { + vi.useRealTimers() + } + }) + + it('answers 202 with the run receipt on the JSON transport', async () => { + vi.useFakeTimers() + try { + const held = holdRun() + const pending = callExecute({ input: {}, waitTimeoutSeconds: 20 }) + await vi.advanceTimersByTimeAsync(20_000) + const response = await pending + + expect(response.status).toBe(202) + expect(response.headers.get('X-Run-Id')).toBe('execution-123') + expect((await response.json()).data).toEqual(receipt) + expect(held.signal?.aborted).toBe(false) + expect(mockAdmissionRelease).not.toHaveBeenCalled() + + held.finish(settledCore) + await vi.advanceTimersByTimeAsync(0) + expect(mockAdmissionRelease).toHaveBeenCalledTimes(1) + } finally { + vi.useRealTimers() + } + }) + + it('keeps a detachable run going when the caller disconnects', async () => { + const held = holdRun() + const disconnect = new AbortController() + const request = new NextRequest(new URL('http://localhost:3000/api/test'), { + method: 'POST', + headers: new Headers({ + 'Content-Type': 'application/json', + 'X-API-Key': 'test-key', + Accept: 'application/x-ndjson', + }), + body: JSON.stringify({ input: {}, waitTimeoutSeconds: 600 }), + signal: disconnect.signal, + }) + const response = await POST(request, createRouteContext({ workflowId: 'workflow-1' })) + const lines = readLines(response) + await lines.next() + + disconnect.abort() + await lines.reader.cancel() + + expect(held.signal?.aborted).toBe(false) + held.finish(settledCore) + }) + + it('stops a detached run through runs cancel in the process that holds it', async () => { + const held = holdRun() + const response = await callExecute( + { input: {}, waitTimeoutSeconds: 600 }, + { Accept: 'application/x-ndjson' } + ) + const lines = readLines(response) + await lines.next() + + expect(abortManualExecution('execution-123')).toBe(true) + expect(await lines.next()).toMatchObject({ + type: 'final', + data: { runId: 'execution-123', status: 'cancelled' }, + }) + expect(held.signal?.aborted).toBe(true) + expect(abortManualExecution('execution-123')).toBe(false) + }) + + it('refuses a wait on runs that never hold the response', async () => { + expect((await callExecute({ waitTimeoutSeconds: 30, async: true })).status).toBe(400) + expect((await callExecute({ waitTimeoutSeconds: 30, stream: true })).status).toBe(400) + + dbChainMockFns.limit.mockReset() + queuePublicWorkflowReads() + expect((await callPublicExecute({ waitTimeoutSeconds: 30 })).status).toBe(400) + expect(mockPreprocessExecution).not.toHaveBeenCalled() + }) + + it('accepts an execution timeout only on a run that may outlive the response', async () => { + expect((await callExecute({ executionTimeoutSeconds: 60 })).status).toBe(400) + + const response = await callExecute({ executionTimeoutSeconds: 60, waitTimeoutSeconds: 30 }) + expect(response.status).toBe(200) + expect(mockPreprocessExecution).toHaveBeenCalledWith( + expect.objectContaining({ requestedTimeoutSeconds: 60 }) + ) + }) + + it('dispatches a detachable manual draft run without tying it to the request', async () => { + authenticatePersonalKey() + + const response = await callExecute({ + run: { source: 'manual', stopAfterBlockId: 'agent-1' }, + waitTimeoutSeconds: 30, + }) + + expect(response.status).toBe(200) + const call = mockExecuteManualTrigger.mock.calls[0][0] + expect(call.input).toMatchObject({ + mode: 'sync-result-stream', + detachable: true, + stopAfterBlockId: 'agent-1', + }) + expect(call.input.abortSignal).toBeUndefined() + }) + }) }) 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 1251ccc2923..3b8a7d30dbe 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts @@ -47,6 +47,7 @@ import { type ExecuteWorkflowServicePendingRun, type ExecuteWorkflowServiceResult, type ExecuteWorkflowServiceRun, + type ExecuteWorkflowServiceTerminalResult, executeWorkflowService, } from '@/lib/workflows/executor/execute-service' import { @@ -132,49 +133,113 @@ function presentRun(result: ExecuteWorkflowServiceRun) { } } +/** Where a run that continues past its response is followed: the v2 runs resource. */ +function runReceipt(workflowId: string, runId: string) { + return { runId, statusUrl: `${getBaseUrl()}/api/v2/workflows/${workflowId}/runs/${runId}` } +} + +/** A run that may outlive its response: past `afterMs` (never, when null) only its receipt is sent. */ +interface RunDetachment { + afterMs: number | null + receipt: ReturnType +} + +function terminalRunResponse(result: ExecuteWorkflowServiceTerminalResult): Response { + if (!result.ok) return serviceFailureResponse(result.failure) + if (result.aborted === 'client') { + return v2Error('CLIENT_CLOSED_REQUEST', 'Client cancelled request', { + details: { runId: result.executionId }, + }) + } + return v2Data(presentRun(result), { + headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId }, + }) +} + +/** + * The run's result, or null once it may detach: `afterMs` has passed (never, when null) + * and the run can be found by its id, so the receipt handed out always resolves. The run + * itself is never cancelled here. + */ +async function settledBeforeDetach( + run: ExecuteWorkflowServicePendingRun, + afterMs: number | null +): Promise { + if (afterMs === null) return run.pending + let timer: ReturnType | undefined + const deadline = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), afterMs) + }) + try { + const settled = await Promise.race([run.pending, deadline]) + if (settled) return settled + } finally { + clearTimeout(timer) + } + return Promise.race([run.pending, run.logged.then(() => null)]) +} + /** * 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. + * existing semantics. A detachable run ends the stream with a `detached` frame + * carrying its receipt once the wait elapses, and a reader that goes away + * leaves it running instead of cancelling it. */ function streamPendingRun( pendingRun: ExecuteWorkflowServicePendingRun, requestId: string, - onSettled: () => void + onSettled: () => void, + detachment?: RunDetachment ): Response { - let cancelled = false + let closed = false let heartbeatId: ReturnType | undefined - const stopHeartbeat = () => { - if (heartbeatId) { - clearInterval(heartbeatId) - heartbeatId = undefined - } + const stopTimers = () => { + clearInterval(heartbeatId) + heartbeatId = undefined + } + const abandon = () => { + if (closed) return + closed = true + stopTimers() + if (!detachment) pendingRun.cancel() } 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 + if (closed) return false try { controller.enqueue(encodeNdjson(event)) return true } catch { - cancel() + abandon() return false } } + const finish = () => { + if (closed) return + closed = true + stopTimers() + controller.close() + } if (send({ type: 'heartbeat', timestamp: new Date().toISOString() })) { heartbeatId = setInterval(() => { send({ type: 'heartbeat', timestamp: new Date().toISOString() }) }, WORKFLOW_RESULT_HEARTBEAT_INTERVAL_MS) + if (detachment) { + const { receipt } = detachment + void settledBeforeDetach(pendingRun, detachment.afterMs).then( + (settled) => { + if (settled) return + send({ type: 'detached', data: receipt }) + finish() + }, + () => {} + ) + } } void pendingRun.pending @@ -206,15 +271,13 @@ function streamPendingRun( send({ type: 'error', error: 'Internal server error', status: 500 }) }) .finally(() => { - stopHeartbeat() + stopTimers() onSettled() - if (!cancelled) controller.close() + finish() }) }, cancel() { - cancelled = true - stopHeartbeat() - pendingRun.cancel() + abandon() }, }) @@ -376,16 +439,27 @@ export const POST = withRouteHandler( return v2Error('BAD_REQUEST', 'input and run.entry.useMockPayload cannot be combined') } + const detachable = body.waitTimeoutSeconds !== undefined + if (detachable && (body.async || body.stream)) { + return v2Error('BAD_REQUEST', 'waitTimeoutSeconds cannot be combined with async or stream') + } + if (detachable && isPublicApiAccess) { + return v2Error( + 'BAD_REQUEST', + 'waitTimeoutSeconds requires an OAuth access token or API key' + ) + } + if (body.async && isPublicApiAccess) { return v2Error('BAD_REQUEST', 'Async execution requires an OAuth access token or API key') } if (body.async && body.stream) { return v2Error('BAD_REQUEST', 'async and stream cannot be combined') } - if (body.executionTimeoutSeconds !== undefined && !body.async) { + if (body.executionTimeoutSeconds !== undefined && !body.async && !detachable) { return v2Error( 'BAD_REQUEST', - 'executionTimeoutSeconds is supported only for async execution' + 'executionTimeoutSeconds is supported only for async execution or with waitTimeoutSeconds' ) } if ( @@ -416,6 +490,8 @@ export const POST = withRouteHandler( } const resultStream = !body.async && !body.stream && wantsResultStream(req) + const syncMode = resultStream || detachable ? 'sync-result-stream' : 'sync' + const detachAfterMs = body.waitTimeoutSeconds ? body.waitTimeoutSeconds * 1000 : null /** Caller-supplied run IDs are a keyed-caller feature; anonymous callers must not probe the claim table. */ let requestedExecutionId: string | undefined @@ -433,7 +509,9 @@ export const POST = withRouteHandler( includeFileBase64: body.includeFileBase64, base64MaxBytes: body.base64MaxBytes, selectedOutputs: body.selectedOutputs, - abortSignal: req.signal, + requestedTimeoutSeconds: body.executionTimeoutSeconds, + abortSignal: detachable ? undefined : req.signal, + detachable, requestHeaders: req.headers, includeThinking: body.includeThinking, includeToolCalls: body.includeToolCalls, @@ -445,7 +523,7 @@ export const POST = withRouteHandler( input: { ...commonInput, input: body.input, - mode: body.stream ? 'stream' : resultStream ? 'sync-result-stream' : 'sync', + mode: body.stream ? 'stream' : syncMode, blockId: manualRun.entry.blockId, sourceRunId: manualRun.entry.sourceRunId, stopAfterBlockId: manualRun.stopAfterBlockId, @@ -458,7 +536,7 @@ export const POST = withRouteHandler( input: { ...commonInput, input: body.input, - mode: body.stream ? 'stream' : resultStream ? 'sync-result-stream' : 'sync', + mode: body.stream ? 'stream' : syncMode, triggerBlockId: manualRun.entry?.blockId, useMockPayload: manualRun.entry?.useMockPayload === true, stopAfterBlockId: manualRun.stopAfterBlockId, @@ -471,14 +549,7 @@ export const POST = withRouteHandler( input: { ...commonInput, input: body.input ?? {}, - requestedTimeoutSeconds: body.executionTimeoutSeconds, - mode: body.async - ? 'async' - : body.stream - ? 'stream' - : resultStream - ? 'sync-result-stream' - : 'sync', + mode: body.async ? 'async' : body.stream ? 'stream' : syncMode, }, request: req, }) @@ -540,9 +611,26 @@ export const POST = withRouteHandler( } if ('pending' in result) { - const response = streamPendingRun(result, requestId, ticket.release) + const detachment: RunDetachment | undefined = detachable + ? { afterMs: detachAfterMs, receipt: runReceipt(workflowId, result.executionId) } + : undefined + if (resultStream) { + const response = streamPendingRun(result, requestId, ticket.release, detachment) + ticketTransferred = true + return response + } + if (!detachment) { + throw new Error('Only a detachable run answers JSON from a pending result') + } + /** The ticket holds this server's capacity for as long as the run, not the response. */ ticketTransferred = true - return response + void result.pending.then(ticket.release, ticket.release) + const settled = await settledBeforeDetach(result, detachment.afterMs) + if (settled) return terminalRunResponse(settled) + return v2Data(detachment.receipt, { + status: 202, + headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId }, + }) } if ('stream' in result) { @@ -556,24 +644,13 @@ export const POST = withRouteHandler( } if ('queued' in result) { - return v2Data( - { - runId: result.executionId, - statusUrl: `${getBaseUrl()}/api/v2/workflows/${workflowId}/runs/${result.executionId}`, - }, - { status: 202, headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId } } - ) - } - - if (result.aborted === 'client') { - return v2Error('CLIENT_CLOSED_REQUEST', 'Client cancelled request', { - details: { runId: result.executionId }, + return v2Data(runReceipt(workflowId, result.executionId), { + status: 202, + headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId }, }) } - return v2Data(presentRun(result), { - headers: { [V2_WORKFLOW_RUN_ID_HEADER]: result.executionId }, - }) + return terminalRunResponse(result) } catch (error) { const classified = v2WorkflowErrorPolicies.concealWorkflowAuthorization.render(error) if (classified) return classified diff --git a/apps/sim/lib/api/contracts/v2/workflows.ts b/apps/sim/lib/api/contracts/v2/workflows.ts index f3ad518de4d..a441518b467 100644 --- a/apps/sim/lib/api/contracts/v2/workflows.ts +++ b/apps/sim/lib/api/contracts/v2/workflows.ts @@ -1357,7 +1357,8 @@ export type V2ExecutionError = z.output * * It deliberately does not enumerate the combinations the route rejects: a * caller reads a constraint where it applies, so each lives on `async`, - * `stream`, `executionTimeoutSeconds`, `includeThinking`, or `includeToolCalls`. + * `stream`, `executionTimeoutSeconds`, `waitTimeoutSeconds`, `includeThinking`, or + * `includeToolCalls`. */ export const EXECUTE_OPTION_CONSTRAINTS = 'Input descriptions specify compatible modes; invalid combinations return `400`.' @@ -1447,7 +1448,7 @@ export const v2ExecuteWorkflowBodySchema = z run: v2WorkflowRunSelectionSchema .optional() .describe( - 'Workflow state and entry point to execute. Omit for the active deployment. Manual execution requires OAuth or personal-key write access and supports synchronous or streamed runs only.' + 'Workflow state and entry point to execute. Omit for the active deployment. Manual execution requires OAuth or personal-key write access and supports synchronous or streamed runs only; `waitTimeoutSeconds` lets a synchronous one continue in the background.' ), async: z .boolean() @@ -1469,7 +1470,21 @@ export const v2ExecuteWorkflowBodySchema = z .max(MAX_WORKFLOW_EXECUTION_TIMEOUT_SECONDS) .optional() .describe( - "Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true`; otherwise returns `400`." + "Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true` or `waitTimeoutSeconds`; otherwise returns `400`." + ), + /** + * Bounds how long the response waits, never the run: past it the run keeps + * going in this server process exactly as it started, so the draft state, + * entry point, and pause handling are the synchronous run's own. + */ + waitTimeoutSeconds: z + .number() + .int() + .min(0) + .max(MAX_WORKFLOW_EXECUTION_TIMEOUT_SECONDS) + .optional() + .describe( + 'Wait at most this many seconds for a synchronous run; `0` waits until it settles. A run still going then continues in the background and the response carries its run receipt instead of its result: `202` on a JSON response, or a closing `detached` frame on an `application/x-ndjson` result stream, which keeps its `200`. The receipt is sent only once the run can be read on the runs resource, where it can also be cancelled. The run is admitted under asynchronous limits, and a dropped connection no longer cancels it. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.' ), stream: z .boolean() @@ -1531,6 +1546,7 @@ export const v2ExecuteWorkflowBodySchema = z { input: { ticketId: 'ticket_123' }, async: true }, { input: { ticketId: 'ticket_123' }, stream: true }, { run: { source: 'manual' } }, + { run: { source: 'manual' }, waitTimeoutSeconds: 240 }, { run: { source: 'manual', @@ -1591,7 +1607,10 @@ export const v2ExecuteWorkflowDataSchema = z }) export type V2ExecuteWorkflowData = z.output -/** 202 receipt for `async: true` — poll `statusUrl` (the v2 runs resource). */ +/** + * 202 receipt for `async: true`, or for a run still going when `waitTimeoutSeconds` + * elapsed — poll `statusUrl` (the v2 runs resource). + */ export const v2ExecuteWorkflowQueuedSchema = z .object({ runId: v2WorkflowRunIdSchema, @@ -1600,7 +1619,8 @@ export const v2ExecuteWorkflowQueuedSchema = z .meta({ id: 'QueuedWorkflowRun', title: 'Queued workflow run', - description: 'Receipt returned when a workflow run is queued.', + description: + 'Receipt returned when a workflow run is queued, or is still running when `waitTimeoutSeconds` elapsed.', }) export type V2ExecuteWorkflowQueued = z.output @@ -1612,7 +1632,8 @@ export const v2ExecuteWorkflowSuccessSchema = z .meta({ id: 'ExecuteWorkflowResponse', title: 'Execute workflow response', - description: 'A completed synchronous run or an asynchronous queue receipt.', + description: + 'A settled synchronous run, or the receipt of a run that continues in the background.', }) export const v2ExecuteWorkflowContract = defineRouteContract({ diff --git a/apps/sim/lib/logs/execution/logging-session.ts b/apps/sim/lib/logs/execution/logging-session.ts index 2f12d944b41..a3e4f8d88c9 100644 --- a/apps/sim/lib/logs/execution/logging-session.ts +++ b/apps/sim/lib/logs/execution/logging-session.ts @@ -232,6 +232,10 @@ export class LoggingSession { private lastCompletionError: string | null = null /** Set once the run's workspace or workflow is found deleted; nothing is left to record. */ private deletedScope: ExecutionScope | null = null + private markLogStarted: () => void = () => {} + private readonly logStarted = new Promise((resolve) => { + this.markLogStarted = resolve + }) constructor( workflowId: string, @@ -1326,12 +1330,26 @@ export class LoggingSession { } } + /** + * Resolves once {@link safeStart} has written the run's log, from which point the run can be + * found by its id; never, when no log could be started. + */ + whenLogStarted(): Promise { + return this.logStarted + } + /** * Starts the run's log, falling back to a minimal log when the full start fails. `false` when * no log could be started; {@link getDeletedScope} then tells whether the run's workspace or * workflow is gone, in which case its admission slot is already released. */ async safeStart(params: SessionStartParams): Promise { + const started = await this.startOrFallback(params) + if (started) this.markLogStarted() + return started + } + + private async startOrFallback(params: SessionStartParams): Promise { if (this.deletedScope) return false try { await this.start(params) diff --git a/apps/sim/lib/mothership/chat/payload.test.ts b/apps/sim/lib/mothership/chat/payload.test.ts index f03b1829049..eeaec1f385e 100644 --- a/apps/sim/lib/mothership/chat/payload.test.ts +++ b/apps/sim/lib/mothership/chat/payload.test.ts @@ -30,7 +30,11 @@ import { } from '@sim/testing/mocks/workspace-file-manager.mock' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { getExposedIntegrationTools } from '@/lib/integrations/tool-catalog' -import { ChatPayloadSchema, TOOL_DEADLINE_CAPABILITY } from '@/lib/mothership/generated/protocol' +import { + ChatPayloadSchema, + TOOL_DEADLINE_CAPABILITY, + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, +} from '@/lib/mothership/generated/protocol' import { searchUsersV2Tool } from '@/tools/github/search_users' import { gmailListLabelsV2Tool } from '@/tools/gmail/list_labels' import { getToolMetadata } from '@/tools/metadata' @@ -430,7 +434,11 @@ describe('buildCopilotRequestPayload', () => { model: '', }) expect(payload.entitlements).toEqual(enabled ? ['dashboards'] : []) - expect(payload.clientCapabilities).toEqual(['workflow-tool-pickup', TOOL_DEADLINE_CAPABILITY]) + expect(payload.clientCapabilities).toEqual([ + 'workflow-tool-pickup', + TOOL_DEADLINE_CAPABILITY, + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + ]) } ) @@ -507,7 +515,10 @@ describe('buildCopilotRequestPayload', () => { 'mothershipTools', ]) expect(payload).not.toHaveProperty(key) - expect(payload.clientCapabilities).toEqual([TOOL_DEADLINE_CAPABILITY]) + expect(payload.clientCapabilities).toEqual([ + TOOL_DEADLINE_CAPABILITY, + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + ]) if (owner === 'organization') expect(payload).not.toHaveProperty('integrationTools') } ) @@ -844,7 +855,10 @@ describe('Assistant payload', () => { terminalCapable: true, }) expect(payload).not.toHaveProperty('desktop') - expect(payload.clientCapabilities).toEqual([TOOL_DEADLINE_CAPABILITY]) + expect(payload.clientCapabilities).toEqual([ + TOOL_DEADLINE_CAPABILITY, + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + ]) expect(payload.mode).toBe('assistant') expect(payload.assistantSearch).toEqual({ source: 'slack', documentIds: ['document-1'] }) for (const field of ['context', 'commands', 'mothershipTools', 'workflowId']) { diff --git a/apps/sim/lib/mothership/chat/payload.ts b/apps/sim/lib/mothership/chat/payload.ts index a3cbb5a2e00..114851ad8e7 100644 --- a/apps/sim/lib/mothership/chat/payload.ts +++ b/apps/sim/lib/mothership/chat/payload.ts @@ -38,6 +38,7 @@ import { type ChatRequest, type ModelSelection, TOOL_DEADLINE_CAPABILITY, + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, } from '@/lib/mothership/generated/protocol' import type { VfsSnapshotV1 } from '@/lib/mothership/generated/vfs-snapshot-v1' import { getToolEntry } from '@/lib/mothership/tool-executor/router' @@ -562,11 +563,13 @@ export async function buildCopilotRequestPayload(params: BuildPayloadParams): Pr : {}), // The mounted chat view executes client-routed workflow tools (run panel UX), so the // UI declares that capability explicitly; headless callers omit it and the server runs - // those tools immediately. The deadline token is this server's own: it bounds every - // handed call by the deadline its frame names. + // those tools immediately. The deadline and run-wait tokens are this server's own: it + // bounds every handed call by the deadline its frame names, and its CLI lets a + // synchronous `workflows run` outlast `--wait-timeout`. clientCapabilities: [ ...(isAssistant ? [] : ['workflow-tool-pickup']), TOOL_DEADLINE_CAPABILITY, + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, ], } } diff --git a/apps/sim/lib/mothership/generated/protocol.ts b/apps/sim/lib/mothership/generated/protocol.ts index 93526ed0d19..2b252886dde 100644 --- a/apps/sim/lib/mothership/generated/protocol.ts +++ b/apps/sim/lib/mothership/generated/protocol.ts @@ -315,6 +315,13 @@ export const TOOL_DEADLINE_CAPABILITY = "tool-deadline"; */ export const WORKSPACE_FILE_COPY_CAPABILITY = "workspace-file-copy"; +/** + * The `clientCapabilities` token a Sim sends when its CLI's synchronous `workflows run` takes + * `--wait-timeout`: a run still going then continues in the background and the command prints + * its run record and exits 4, so the worker waits on it in slices. + */ +export const WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY = "workflow-run-wait-timeout"; + /** POST /api/mothership — the chat request sim sends. */ export interface ChatRequest extends StreamResponseReceipt { fileOwnerProtocolVersion?: 1 | undefined; @@ -360,8 +367,8 @@ export interface ChatRequest extends StreamResponseReceipt { * grace windows and run tools server-side immediately. ABSENT = older/unknown caller — * dispatch keeps its conservative grace (deploy-skew safe: a stale tab that predates * this field still gets waited on). Known capability: "workflow-tool-pickup". Sim also - * declares its own {@link TOOL_DEADLINE_CAPABILITY} and {@link WORKSPACE_FILE_COPY_CAPABILITY} - * here. + * declares its own {@link TOOL_DEADLINE_CAPABILITY}, {@link WORKSPACE_FILE_COPY_CAPABILITY} and + * {@link WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY} here. */ clientCapabilities?: string[] | undefined; /** diff --git a/apps/sim/lib/workflows/application/execute-manual-workflow.ts b/apps/sim/lib/workflows/application/execute-manual-workflow.ts index 5f51a6a5939..f742270bddc 100644 --- a/apps/sim/lib/workflows/application/execute-manual-workflow.ts +++ b/apps/sim/lib/workflows/application/execute-manual-workflow.ts @@ -23,8 +23,7 @@ import { validateTriggerInput, } from '@/lib/workflows/triggers/run-options' -interface ManualExecutionInput - extends Omit { +interface ManualExecutionInput extends Omit { input?: unknown mode: 'sync' | 'stream' | 'sync-result-stream' stopAfterBlockId?: string @@ -146,8 +145,10 @@ function executionServiceInput(params: { selectedOutputs: params.input.selectedOutputs, stopAfterBlockId: params.input.stopAfterBlockId, rateLimitCounter: 'sync' as const, + requestedTimeoutSeconds: params.input.requestedTimeoutSeconds, abortSignal: params.input.abortSignal, mode: params.input.mode, + detachable: params.input.detachable, requestHeaders: params.input.requestHeaders, includeThinking: params.input.includeThinking, includeToolCalls: params.input.includeToolCalls, diff --git a/apps/sim/lib/workflows/application/execute-workflow.ts b/apps/sim/lib/workflows/application/execute-workflow.ts index 5f50a2d665d..125f393f787 100644 --- a/apps/sim/lib/workflows/application/execute-workflow.ts +++ b/apps/sim/lib/workflows/application/execute-workflow.ts @@ -23,6 +23,8 @@ export interface ExecuteWorkflowInput { requestedTimeoutSeconds?: number abortSignal?: AbortSignal mode: 'sync' | 'async' | 'stream' | 'sync-result-stream' + /** The caller may stop waiting and leave the run going; see `executeWorkflowService`. */ + detachable?: boolean requestHeaders: Headers includeThinking?: boolean includeToolCalls?: boolean @@ -63,6 +65,7 @@ export const executeWorkflowOperation = defineAuthorizedWorkflowUseCase({ requestedTimeoutSeconds: input.requestedTimeoutSeconds, abortSignal: input.abortSignal, mode: input.mode, + detachable: input.detachable, requestHeaders: input.requestHeaders, includeThinking: input.includeThinking, includeToolCalls: input.includeToolCalls, diff --git a/apps/sim/lib/workflows/executor/execute-service.ts b/apps/sim/lib/workflows/executor/execute-service.ts index 262f9cdfd84..5438f6a0776 100644 --- a/apps/sim/lib/workflows/executor/execute-service.ts +++ b/apps/sim/lib/workflows/executor/execute-service.ts @@ -11,6 +11,10 @@ import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core import { SSE_HEADERS } from '@/lib/core/utils/sse' import { PayloadSizeLimitError } from '@/lib/core/utils/stream-limits' import { validateCallChain } from '@/lib/execution/call-chain' +import { + registerManualExecutionAborter, + unregisterManualExecutionAborter, +} from '@/lib/execution/manual-cancellation' import { containsLargeValueRef } from '@/lib/execution/payloads/large-value-ref' import { compactExecutionPayload } from '@/lib/execution/payloads/serializer' import { preprocessExecution } from '@/lib/execution/preprocessing' @@ -106,6 +110,15 @@ export interface ExecuteWorkflowServiceParams { * result so an HTTP surface can emit transport heartbeats while it runs. */ mode?: 'sync' | 'async' | 'stream' | 'sync-result-stream' + /** + * The caller may stop waiting before the run settles, leaving it running in this + * process. Such a run is admitted like an async one (async rate bucket, reservation, + * and timeout), never ends with its caller (so it takes no `abortSignal`), and + * registers an in-process aborter so a run cancel reaches it without Redis. Its + * execution is otherwise the sync run's own, so a pause resumes as a sync run's does. + * Requires `sync-result-stream`, whose pending result the caller stops awaiting. + */ + detachable?: boolean /** Original request headers — stream-protocol negotiation only. */ requestHeaders?: Headers includeThinking?: boolean @@ -182,6 +195,8 @@ export interface ExecuteWorkflowServicePendingRun { ok: true executionId: string pending: Promise + /** Resolves once the run can be found by its id; never, when its log could not start. */ + logged: Promise cancel: () => void } @@ -284,6 +299,7 @@ export async function executeWorkflowService( triggerBlockId, runFromBlock, stopAfterBlockId, + detachable = false, } = params let reqLogger = logger.withMetadata({ requestId, workflowId, userId }) @@ -307,6 +323,10 @@ export async function executeWorkflowService( if (draftState && !useDraftState) { throw new Error('A preloaded draft state requires manual execution state') } + if (detachable && (mode !== 'sync-result-stream' || abortSignal)) { + throw new Error('A detachable run returns its pending result and outlives its request') + } + const executionType = mode === 'async' || detachable ? 'async' : 'sync' if (callChain) { const chainError = validateCallChain(callChain) @@ -396,12 +416,12 @@ export async function executeWorkflowService( executionId, requestId, checkDeployment: !useDraftState, - rateLimitCounter, + rateLimitCounter: detachable ? 'async' : rateLimitCounter, loggingSession, useAuthenticatedUserAsActor, workflowRecord, billingAttribution: upstreamBillingAttribution, - executionType: mode === 'async' ? 'async' : 'sync', + executionType, requestedTimeoutSeconds, }) @@ -670,7 +690,9 @@ export async function executeWorkflowService( executionMode: 'sync', } - const timeoutController = createTimeoutAbortController(preprocessResult.executionTimeout?.sync) + const timeoutController = createTimeoutAbortController( + preprocessResult.executionTimeout?.[executionType] + ) let requestAborted = false const abortFromRequest = () => { requestAborted = true @@ -692,6 +714,7 @@ export async function executeWorkflowService( } const runSynchronousWorkflow = async (): Promise => { + if (detachable) registerManualExecutionAborter(executionId, timeoutController.abort) try { const snapshot = new ExecutionSnapshot( metadata, @@ -898,6 +921,7 @@ export async function executeWorkflowService( } } finally { abortSignal?.removeEventListener('abort', abortFromRequest) + if (detachable) unregisterManualExecutionAborter(executionId, timeoutController.abort) timeoutController.cleanup() /** * The sync result is the run's receipt: callers read its log, state, and cost as @@ -916,6 +940,7 @@ export async function executeWorkflowService( ok: true, executionId, pending: pending.finally(settleExecutionIdClaim), + logged: loggingSession.whenLogStarted(), cancel: abortFromRequest, } } 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 d0d54c207f2..125f57b5667 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 @@ -1,5 +1,5 @@ import { Command } from 'commander' -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { SimApiError } from '../../http/client' import { buildGeneratedCommands } from '../../runtime/build' import { attachWorkflowRunFollow, renderRunStream } from './workflow-run-follow' @@ -142,6 +142,133 @@ describe('renderRunStream', () => { }) }) +describe('sim workflows run --wait-timeout', () => { + const receipt = { + runId: 'run-1', + statusUrl: `https://sim.example/api/v2/workflows/${WORKFLOW_ID}/runs/run-1`, + } + let logged: string[] + let errored: string[] + + beforeEach(() => { + process.exitCode = undefined + logged = [] + errored = [] + vi.spyOn(console, 'log').mockImplementation((line: string) => { + logged.push(line) + }) + vi.spyOn(console, 'error').mockImplementation((line: string) => { + errored.push(line) + }) + }) + + afterEach(() => { + process.exitCode = undefined + }) + + function runRecord(status: string) { + return { + data: { + runId: 'run-1', + workflowId: WORKFLOW_ID, + status, + startedAt: '2026-08-17T00:00:00.000Z', + endedAt: null, + durationMs: null, + cost: null, + error: null, + paused: null, + }, + } + } + + it('asks the server to wait that long and prints a run that settled in time as before', async () => { + requestRaw.mockResolvedValue( + ndjsonResponse( + { type: 'heartbeat' }, + { + type: 'final', + data: { runId: 'run-1', workflowId: WORKFLOW_ID, status: 'completed', output: { ok: 1 } }, + } + ) + ) + + await run(WORKFLOW_ID, '--manual', '--wait-timeout', '240') + + expect(requestRaw.mock.calls[0][1].body).toEqual({ + run: { source: 'manual' }, + waitTimeoutSeconds: 240, + }) + expect(request).not.toHaveBeenCalled() + expect(logged.join('\n')).toContain('"ok": 1') + expect(process.exitCode).toBeUndefined() + }) + + it('prints the run record and exits 4 when the run outlasts the wait', async () => { + requestRaw.mockResolvedValue( + ndjsonResponse({ type: 'heartbeat' }, { type: 'detached', data: receipt }) + ) + request.mockResolvedValue(runRecord('running')) + + await run(WORKFLOW_ID, '--stop-after', 'agent-1', '--wait-timeout', '30') + + expect(request).toHaveBeenCalledTimes(1) + expect(request.mock.calls[0][0]).toBe(`/api/v2/workflows/${WORKFLOW_ID}/runs/run-1`) + expect(JSON.parse(logged.join('\n'))).toMatchObject({ runId: 'run-1', status: 'running' }) + expect(errored.join('\n')).toContain( + `sim workflows runs wait run-1 --workflow ${WORKFLOW_ID} --wait-timeout 30` + ) + expect(process.exitCode).toBe(4) + }) + + it('reports a run that settled between detaching and the read like runs wait does', async () => { + requestRaw.mockResolvedValue(ndjsonResponse({ type: 'detached', data: receipt })) + request.mockResolvedValue(runRecord('failed')) + + await run(WORKFLOW_ID, '--manual', '--wait-timeout', '30') + + expect(JSON.parse(logged.join('\n'))).toMatchObject({ status: 'failed' }) + expect(process.exitCode).toBe(1) + }) + + it('treats a 202 receipt on the JSON transport as a detached run', async () => { + requestRaw.mockResolvedValue( + new Response(JSON.stringify({ data: receipt }), { + status: 202, + headers: { 'content-type': 'application/json' }, + }) + ) + request.mockResolvedValue(runRecord('running')) + + await run(WORKFLOW_ID, '--manual', '--wait-timeout', '30') + + expect(process.exitCode).toBe(4) + }) + + it('rounds a fractional wait up to whole seconds, and 0 waits until the run settles', async () => { + await run(WORKFLOW_ID, '--manual', '--wait-timeout', '1.5') + await run(WORKFLOW_ID, '--manual', '--wait-timeout', '0') + + expect(requestRaw.mock.calls[0][1].body.waitTimeoutSeconds).toBe(2) + expect(requestRaw.mock.calls[1][1].body.waitTimeoutSeconds).toBe(0) + }) + + it('refuses a wait the run cannot honour before sending anything', async () => { + await expect(run(WORKFLOW_ID, '--wait-timeout', '-1')).rejects.toThrow('--wait-timeout') + await expect(run(WORKFLOW_ID, '--wait-timeout', '30', '--async')).rejects.toThrow( + '--wait-timeout' + ) + await expect(run(WORKFLOW_ID, '--wait-timeout', '30', '--follow')).rejects.toThrow( + '--wait-timeout' + ) + expect(requestRaw).not.toHaveBeenCalled() + }) + + it('points a manual --async at --wait-timeout', async () => { + await expect(run(WORKFLOW_ID, '--manual', '--async')).rejects.toThrow('--wait-timeout') + }) +}) + function program(): Command { const root = new Command() root.exitOverride() 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 d51b1423d50..ed43a1fb47f 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts +++ b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts @@ -8,12 +8,18 @@ import { V2_OPERATIONS } from '../../generated/v2-api' import { SimApiError } from '../../http/client' import { readNdjson } from '../../http/ndjson' import { safeOneLine, sanitize } from '../../output/render' -import { connect } from '../../runtime/called-operations' +import { callsOperations, connect } from '../../runtime/called-operations' 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' +import { + parseWaitTimeout, + type RunSnapshot, + WAIT_TIMEOUT_FLAG, + waitForRun, +} from './workflow-run-wait' /** * Declares that this client understands agent-event framing. @@ -75,7 +81,10 @@ export function resolveWorkflowRunSelection( throw new SimApiError('--source-run requires --from-block ', 0) } if ((manual || fromBlock) && flags.async === true) { - throw new SimApiError('Manual execution does not support --async', 0) + throw new SimApiError( + `Manual execution does not support --async; pass ${WAIT_TIMEOUT_FLAG} to let a long run continue in the background`, + 0 + ) } if (useMockPayload && flags.input !== undefined) { throw new SimApiError('--mock-payload cannot be combined with --input', 0) @@ -110,12 +119,29 @@ function stringField(frame: Record, key: string): string | null return typeof value === 'string' ? value : null } +/** A synchronous run's answer: its result, or the id of a run that outlasted `--wait-timeout`. */ +type WorkflowResult = + | { settled: Record } + | { + /** The run continues in the background; follow it on the runs resource. */ + detachedRunId: string + } + +function detachedRunId(receipt: unknown, status: number): string { + if (!isRecordLike(receipt) || typeof receipt.runId !== 'string' || receipt.runId === '') { + throw new SimApiError('Workflow run continued in the background without naming its run', status) + } + return receipt.runId +} + /** * 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. + * ordinary JSON envelope, which is consumed without retrying the run. Either + * transport may instead hand back the receipt of a run that is still going when + * `--wait-timeout` elapsed: a `detached` frame, or a `202`. */ -async function readWorkflowResult(response: Response): Promise> { +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 @@ -130,7 +156,9 @@ async function readWorkflowResult(response: Response): Promise { +/** What a run still going when `--wait-timeout` elapsed says beside its record. */ +function stillRunning(runId: string, workflowId: string, waitedSeconds: number) { + return (snapshot: RunSnapshot) => + `Run ${runId} is still ${snapshot.status} after ${waitedSeconds}s and continues in the background. Keep waiting: sim workflows runs wait ${runId} --workflow ${workflowId} --wait-timeout ${waitedSeconds}. Stop it: sim workflows runs cancel ${runId} --workflow ${workflowId}` +} + +/** + * Runs synchronously while keeping idle-limited HTTP paths active. + * + * With `--wait-timeout`, a run the server stops waiting for keeps going, and this + * command ends as a `workflows runs wait` whose bound just elapsed would: the run + * record on stdout and its exit status (4 while it is still moving). + */ +async function runWithResultStream( + workflowId: string, + command: Command, + waitTimeoutSeconds: number | undefined +): Promise { const flags = command.optsWithGlobals() as Record - const { client, profile } = connect(command, ['executeWorkflow']) + const connection = connect(command, ['executeWorkflow', 'getWorkflowRun']) + const { client, profile } = connection const operation = V2_OPERATIONS.executeWorkflow as OperationSpec const commandSpec = CLI_CONTRACT.executeWorkflow ?? {} try { const request = await buildRequest('executeWorkflow', [workflowId], flags, profile.workspaceId) + const body = withEmbeddedFileReferences(request.body) const response = await client.requestRaw('executeWorkflow', { params: request.params, query: request.query, - body: withEmbeddedFileReferences(request.body), + body: waitTimeoutSeconds === undefined ? body : { ...body, waitTimeoutSeconds }, headers: { ...request.headers, accept: WORKFLOW_RESULT_STREAM_CONTENT_TYPE }, }) - const payload = await readWorkflowResult(response) + const result = await readWorkflowResult(response) + if ('detachedRunId' in result) { + // The server already spent the wait, so the run is read once and reported. + const now = Date.now() + await waitForRun(connection, { + runId: result.detachedRunId, + workflowId, + startedAt: now, + deadline: now, + timedOut: stillRunning(result.detachedRunId, workflowId, waitTimeoutSeconds ?? 0), + }) + return + } + const payload = result.settled renderResult('executeWorkflow', profile.output, payload, commandSpec, { expandedTrace: flags.trace === true, @@ -426,6 +488,22 @@ async function followRun(workflowId: string, command: Command): Promise { function followOrDelegate(previous: ((args: unknown[]) => unknown) | null) { return async (workflowId: string, _options: unknown, command: Command): Promise => { const initialFlags = command.optsWithGlobals() as Record + const waitTimeoutSeconds = + typeof initialFlags.waitTimeout === 'string' + ? Math.ceil(parseWaitTimeout(initialFlags.waitTimeout)) + : undefined + if (waitTimeoutSeconds !== undefined && initialFlags.async === true) { + throw new SimApiError( + '--wait-timeout bounds how long a synchronous run is waited for, and --async never waits; pass one, not both', + 0 + ) + } + if (waitTimeoutSeconds !== undefined && initialFlags.follow === true) { + throw new SimApiError( + '--wait-timeout cannot be combined with --follow, which streams the run until it ends', + 0 + ) + } const selection = resolveWorkflowRunSelection(initialFlags) if (selection) command.setOptionValue('run', selection) const flags = command.optsWithGlobals() as Record @@ -451,7 +529,7 @@ function followOrDelegate(previous: ((args: unknown[]) => unknown) | null) { ) } if (flags.async !== true) { - await runWithResultStream(workflowId, command) + await runWithResultStream(workflowId, command, waitTimeoutSeconds) return } @@ -492,6 +570,8 @@ export function attachWorkflowRunFollow(workflows: Command): void { const held = (run as Command & { _actionHandler?: unknown })._actionHandler const previous = typeof held === 'function' ? (held as (args: unknown[]) => unknown) : null + // A run that outlasts --wait-timeout is read back from the runs resource. + callsOperations(run, ['executeWorkflow', 'getWorkflowRun'], { runs: 'executeWorkflow' }) run .option('--manual', 'Run the current saved workflow state instead of the active deployment') @@ -512,6 +592,10 @@ export function attachWorkflowRunFollow(workflows: Command): void { '--stop-after ', 'Stop the run after this saved block, failing it if the run takes a path that skips the block; with --from-block on the same block, re-runs only that block (implies --manual)' ) + .option( + WAIT_TIMEOUT_FLAG, + 'Wait at most this many seconds for a synchronous run (0 waits until it settles). A run still going then continues in the background under async limits: the command prints its run record and exits 4, and `sim workflows runs wait` keeps waiting. Not with --async or --follow' + ) .option( '--follow', 'Stream the run as it happens; progress on stderr, result on stdout. The stream reports only success and output, so the result omits the run id and timings a non-streaming run returns' diff --git a/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts b/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts index f238cb3b947..23fdcaa36dc 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts +++ b/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts @@ -8,7 +8,7 @@ import type { CommandSpec } from '../../contract/types' import { setSoftExitCode } from '../../embed-context' import { V2_OPERATIONS } from '../../generated/v2-api' import { SimApiError } from '../../http/client' -import { apiCommand } from '../../runtime/called-operations' +import { apiCommand, type Connection } from '../../runtime/called-operations' import { renderResult } from '../../runtime/result' /** @@ -78,9 +78,9 @@ const DEFAULT_WAIT_TIMEOUT_SECONDS = 3600 * many of them. Two knobs both called "timeout" would be read as one, and the * failure that follows is silent — raising the wrong one changes nothing. */ -const WAIT_TIMEOUT_FLAG = '--wait-timeout ' +export const WAIT_TIMEOUT_FLAG = '--wait-timeout ' -interface RunSnapshot { +export interface RunSnapshot { status: string /** `time` resumes itself; `human` and null wait for a person. */ pauseKind: string | null @@ -175,7 +175,7 @@ function waitProgress(): WaitProgress { } } -function parseWaitTimeout(raw: string): number { +export function parseWaitTimeout(raw: string): number { const seconds = Number(raw) if (!Number.isFinite(seconds) || seconds < 0) { throw new SimApiError( @@ -209,6 +209,76 @@ function runSpec(): CommandSpec { return CLI_CONTRACT.getWorkflowRun ?? {} } +interface RunWait { + runId: string + workflowId: string + /** When the wait began, for the progress line. */ + startedAt: number + /** When the wait gives up, in epoch milliseconds; `Infinity` waits indefinitely. */ + deadline: number + /** What to say when the bound elapses with the run still moving. */ + timedOut: (snapshot: RunSnapshot) => string +} + +/** The timeout line `workflows runs wait` prints for its own bound. */ +function waitTimedOut(wait: { runId: string; timeoutSeconds: number }) { + return (snapshot: RunSnapshot) => + `Timed out after ${wait.timeoutSeconds}s waiting for run ${wait.runId} (status: ${snapshot.status}${ + snapshot.resumeAt ? `, resuming at ${snapshot.resumeAt}` : '' + }). Raise ${WAIT_TIMEOUT_FLAG}, or set it to 0 to wait indefinitely.` +} + +/** + * Polls one run until it settles or the deadline passes, prints the run, and sets the + * exit status {@link WAIT_EXIT_CODES} names. A deadline already past still reads the run + * once, so a run that settled meanwhile reports its outcome rather than a timeout. + */ +export async function waitForRun( + { client, profile }: Connection<'getWorkflowRun'>, + wait: RunWait +): Promise { + const params = { workflowId: wait.workflowId, runId: wait.runId } + const progress = waitProgress() + let delayMs = FIRST_POLL_DELAY_MS + + // `finally`, because a request that throws part-way through would + // otherwise leave `running — waiting 12s…` sitting on the line the error + // is then written onto. + try { + while (true) { + const raw = await client.request('getWorkflowRun', { params }) + const snapshot = readRun(raw) + const outcome = classify(snapshot) + + if (outcome) { + progress.finish() + renderResult('getWorkflowRun', profile.output, runData(raw), runSpec()) + const message = explain(outcome, wait.runId, wait.workflowId, snapshot) + if (message) printError(styles().red(message)) + setSoftExitCode(WAIT_EXIT_CODES[outcome]) + return + } + + const remainingMs = wait.deadline - Date.now() + if (remainingMs <= 0) { + progress.finish() + renderResult('getWorkflowRun', profile.output, runData(raw), runSpec()) + printError(styles().red(wait.timedOut(snapshot))) + setSoftExitCode(WAIT_EXIT_CODES.timeout) + return + } + + progress.advance(snapshot.status, Date.now() - wait.startedAt) + // Clamped to the time left so the last sleep of a bounded wait ends + // at the deadline instead of overshooting it by a whole interval. + await sleep(Math.min(delayMs, remainingMs)) + delayMs = Math.min(delayMs * POLL_BACKOFF_FACTOR, MAX_POLL_DELAY_MS) + } + } finally { + progress.finish() + } +} + /** Adds `workflows runs wait` — poll one run until it stops moving. */ export function attachWorkflowRunWait(runs: Command): void { const [wait, connectWait] = apiCommand(runs, 'wait', ['getWorkflowRun']) @@ -235,56 +305,14 @@ export function attachWorkflowRunWait(runs: Command): void { ? DEFAULT_WAIT_TIMEOUT_SECONDS : parseWaitTimeout(options.waitTimeout) - const { client, profile } = connectWait() - const params = { workflowId: options.workflow, runId } - const startedAt = Date.now() - const deadline = - timeoutSeconds === 0 ? Number.POSITIVE_INFINITY : startedAt + timeoutSeconds * 1000 - const progress = waitProgress() - let delayMs = FIRST_POLL_DELAY_MS - - // `finally`, because a request that throws part-way through would - // otherwise leave `running — waiting 12s…` sitting on the line the error - // is then written onto. - try { - while (true) { - const raw = await client.request('getWorkflowRun', { params }) - const snapshot = readRun(raw) - const outcome = classify(snapshot) - - if (outcome) { - progress.finish() - renderResult('getWorkflowRun', profile.output, runData(raw), runSpec()) - const message = explain(outcome, runId, options.workflow, snapshot) - if (message) printError(styles().red(message)) - setSoftExitCode(WAIT_EXIT_CODES[outcome]) - return - } - - const remainingMs = deadline - Date.now() - if (remainingMs <= 0) { - progress.finish() - renderResult('getWorkflowRun', profile.output, runData(raw), runSpec()) - printError( - styles().red( - `Timed out after ${timeoutSeconds}s waiting for run ${runId} (status: ${snapshot.status}${ - snapshot.resumeAt ? `, resuming at ${snapshot.resumeAt}` : '' - }). Raise ${WAIT_TIMEOUT_FLAG}, or set it to 0 to wait indefinitely.` - ) - ) - setSoftExitCode(WAIT_EXIT_CODES.timeout) - return - } - - progress.advance(snapshot.status, Date.now() - startedAt) - // Clamped to the time left so the last sleep of a bounded wait ends - // at the deadline instead of overshooting it by a whole interval. - await sleep(Math.min(delayMs, remainingMs)) - delayMs = Math.min(delayMs * POLL_BACKOFF_FACTOR, MAX_POLL_DELAY_MS) - } - } finally { - progress.finish() - } + await waitForRun(connectWait(), { + runId, + workflowId: options.workflow, + startedAt, + deadline: + timeoutSeconds === 0 ? Number.POSITIVE_INFINITY : startedAt + timeoutSeconds * 1000, + timedOut: waitTimedOut({ runId, timeoutSeconds }), + }) }) } diff --git a/packages/sim-cli/src/contract/commands.ts b/packages/sim-cli/src/contract/commands.ts index 2ec1d809dd8..0223e860639 100644 --- a/packages/sim-cli/src/contract/commands.ts +++ b/packages/sim-cli/src/contract/commands.ts @@ -2139,6 +2139,10 @@ export const CLI_CONTRACT: CliContract = { describe: 'Run a deployed workflow or execute saved state manually', flags: { async: { boolean: true, describe: 'Queue the run and return immediately' }, + executionTimeoutSeconds: { + describe: + "Maximum duration of the run in seconds, capped by the plan's execution timeout. Requires --async or --wait-timeout", + }, input: { json: true, describe: 'Trigger input as JSON' }, run: { hidden: true, @@ -2159,6 +2163,10 @@ export const CLI_CONTRACT: CliContract = { stream: { omit: true }, includeThinking: { omit: true }, includeToolCalls: { omit: true }, + // Spelled `--wait-timeout ` like every other wait, and parsed the + // same way, by `workflow-run-follow.ts`, which also reads the run back + // when the server stops waiting. + waitTimeoutSeconds: { omit: true }, // Its embedded default lives with the synchronous request in // `workflow-run-follow.ts`, because an `--async` run rejects the field. includeFileBase64: { diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts index 09e3a5f7992..af76b92da10 100644 --- a/packages/sim-cli/src/generated/v2-api.ts +++ b/packages/sim-cli/src/generated/v2-api.ts @@ -5547,6 +5547,7 @@ export type ExecuteWorkflowBody = { } async?: boolean executionTimeoutSeconds?: number + waitTimeoutSeconds?: number stream?: boolean selectedOutputs?: Array includeThinking?: boolean @@ -18439,7 +18440,7 @@ export const V2_OPERATIONS = { run: { kind: 'unknown', describe: - 'Workflow state and entry point to execute. Omit for the active deployment. Manual execution requires OAuth or personal-key write access and supports synchronous or streamed runs only.', + 'Workflow state and entry point to execute. Omit for the active deployment. Manual execution requires OAuth or personal-key write access and supports synchronous or streamed runs only; `waitTimeoutSeconds` lets a synchronous one continue in the background.', }, async: { kind: 'boolean', @@ -18450,7 +18451,12 @@ export const V2_OPERATIONS = { executionTimeoutSeconds: { kind: 'integer', describe: - "Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true`; otherwise returns `400`.", + "Maximum duration of an asynchronous run, in seconds, capped by the plan's execution timeout. Requires `async: true` or `waitTimeoutSeconds`; otherwise returns `400`.", + }, + waitTimeoutSeconds: { + kind: 'integer', + describe: + 'Wait at most this many seconds for a synchronous run; `0` waits until it settles. A run still going then continues in the background and the response carries its run receipt instead of its result: `202` on a JSON response, or a closing `detached` frame on an `application/x-ndjson` result stream, which keeps its `200`. The receipt is sent only once the run can be read on the runs resource, where it can also be cancelled. The run is admitted under asynchronous limits, and a dropped connection no longer cancels it. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.', }, stream: { kind: 'boolean', diff --git a/packages/testing/src/mocks/logging-session.mock.ts b/packages/testing/src/mocks/logging-session.mock.ts index 8c4ba248bca..92fc139bad0 100644 --- a/packages/testing/src/mocks/logging-session.mock.ts +++ b/packages/testing/src/mocks/logging-session.mock.ts @@ -5,7 +5,7 @@ import { vi } from 'vitest' * `@/lib/logs/execution/logging-session`. Every instance method is backed by a * shared `vi.fn()` so tests that construct multiple sessions observe identical * mock state. `mockSafeStart` defaults to `true` because callers branch on the - * boolean result. Display projection methods (incl. `projectTraceSpansForLiveDisplay`) return + * boolean result, and `mockWhenLogStarted` resolves at once, as for a log already written. Display projection methods (incl. `projectTraceSpansForLiveDisplay`) return * their input, diagnostic projection fails closed to structural metadata, `hasCompleted` is * `false`, `getPersistedCompletionStatus` and `getDeletedScope` are `null`, and other methods * resolve to `undefined`. @@ -26,6 +26,7 @@ export const loggingSessionMockFns = { mockCompleteWithCancellation: vi.fn(async (..._args: unknown[]): Promise => {}), mockCompleteWithPause: vi.fn(async (..._args: unknown[]): Promise => {}), mockSafeStart: vi.fn(async (..._args: unknown[]): Promise => true), + mockWhenLogStarted: vi.fn(async (): Promise => {}), mockWaitForCompletion: vi.fn(async (..._args: unknown[]): Promise => {}), mockWaitForPostExecution: vi.fn(async (..._args: unknown[]): Promise => {}), mockSetTrustedExecutionCorrelation: vi.fn(), @@ -73,6 +74,7 @@ function buildLoggingSessionInstance() { completeWithCancellation: loggingSessionMockFns.mockCompleteWithCancellation, completeWithPause: loggingSessionMockFns.mockCompleteWithPause, safeStart: loggingSessionMockFns.mockSafeStart, + whenLogStarted: loggingSessionMockFns.mockWhenLogStarted, waitForCompletion: loggingSessionMockFns.mockWaitForCompletion, waitForPostExecution: loggingSessionMockFns.mockWaitForPostExecution, setTrustedExecutionCorrelation: loggingSessionMockFns.mockSetTrustedExecutionCorrelation, diff --git a/scripts/check-unused-exports.baseline.json b/scripts/check-unused-exports.baseline.json index 39101fa20c7..ea30b9e63ca 100644 --- a/scripts/check-unused-exports.baseline.json +++ b/scripts/check-unused-exports.baseline.json @@ -5132,7 +5132,6 @@ "apps/sim/lib/workflows/executor/enqueue-execution.ts#WORKFLOW_EXECUTION_JOB_ID_PREFIX", "apps/sim/lib/workflows/executor/execute-service.ts#ExecuteWorkflowServiceQueued", "apps/sim/lib/workflows/executor/execute-service.ts#ExecuteWorkflowServiceStream", - "apps/sim/lib/workflows/executor/execute-service.ts#ExecuteWorkflowServiceTerminalResult", "apps/sim/lib/workflows/executor/execute-service.ts#resolveOutputIds", "apps/sim/lib/workflows/executor/execution-queries.ts#WorkflowExecutionCursor", "apps/sim/lib/workflows/executor/execution-queries.ts#WorkflowExecutionStatus",