From 323071f35d481ef68fae189acc29b7ef4715d110 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 20:20:02 -0700 Subject: [PATCH 1/3] feat(workflows): let a synchronous run continue in the background after a wait timeout A synchronous run, including a manual draft run (--manual, --trigger, --from-block, --stop-after), could only be waited on until it settled or hit the sync execution limit. Manual runs refuse async, so a caller had no way to get the run id back from a long or stuck draft run and decide to keep waiting, watch it, or cancel it. The v2 execute body takes waitTimeoutSeconds. Past it the response answers 202 with the run receipt (NDJSON: a `detached` frame) and the run keeps going in this process exactly as it started: same captured draft state, entry point and pause persistence. Such a run is admitted like an async one (async rate bucket, reservation and timeout), takes no request abort signal so a dropped connection never cancels it, and registers an in-process aborter so runs cancel reaches it without Redis. Without the field nothing changes. The CLI exposes it as `workflows run --wait-timeout `. A run that outlasts it is read back once and reported the way `workflows runs wait` reports an elapsed bound: the run record on stdout and exit 4 while it is still moving, so `workflows runs wait ` continues it. --- apps/docs/content/docs/cli/reference.mdx | 3 +- apps/docs/content/docs/cli/workflows.mdx | 3 +- apps/docs/openapi-v2-workflows.json | 31 ++- .../[workflowId]/execute/route.test.ts | 211 ++++++++++++++++++ .../workflows/[workflowId]/execute/route.ts | 172 +++++++++----- apps/sim/lib/api/contracts/v2/workflows.ts | 33 ++- .../application/execute-manual-workflow.ts | 5 +- .../workflows/application/execute-workflow.ts | 3 + .../lib/workflows/executor/execute-service.ts | 28 ++- .../protocol/workflow-run-follow.test.ts | 129 ++++++++++- .../commands/protocol/workflow-run-follow.ts | 108 ++++++++- .../commands/protocol/workflow-run-wait.ts | 136 ++++++----- packages/sim-cli/src/contract/commands.ts | 8 + packages/sim-cli/src/generated/v2-api.ts | 10 +- scripts/check-unused-exports.baseline.json | 1 - 15 files changed, 742 insertions(+), 139 deletions(-) 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..619d81d42f5 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, then answer `202` with the run receipt while it continues in the background; `0` waits until it settles. The run is admitted under asynchronous limits, a dropped connection no longer cancels it, and it can be watched or cancelled through the runs resource. 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..8b7fa36c71c 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 @@ -35,8 +35,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, @@ -1120,4 +1123,212 @@ 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('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..1e5dbdb73ee 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,106 @@ 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 pending run's result, or null when `afterMs` passes first; the run is never cancelled here. */ +async function settledWithin( + pending: Promise, + afterMs: number | null +): Promise { + if (afterMs === null) return pending + let timer: ReturnType | undefined + const deadline = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), afterMs) + }) + try { + return await Promise.race([pending, deadline]) + } finally { + clearTimeout(timer) + } +} + /** * 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 - } + let detachId: ReturnType | undefined + const stopTimers = () => { + clearInterval(heartbeatId) + clearTimeout(detachId) + heartbeatId = undefined + detachId = 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?.afterMs != null) { + const { receipt } = detachment + detachId = setTimeout(() => { + send({ type: 'detached', data: receipt }) + finish() + }, detachment.afterMs) + } } void pendingRun.pending @@ -206,15 +264,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 +432,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 +483,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 +502,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 +516,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 +529,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 +542,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 +604,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 settledWithin(result.pending, 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 +637,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..882ae14a69a 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, then answer `202` with the run receipt while it continues in the background; `0` waits until it settles. The run is admitted under asynchronous limits, a dropped connection no longer cancels it, and it can be watched or cancelled through the runs resource. 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/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..c9f2f92b364 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 @@ -284,6 +297,7 @@ export async function executeWorkflowService( triggerBlockId, runFromBlock, stopAfterBlockId, + detachable = false, } = params let reqLogger = logger.withMetadata({ requestId, workflowId, userId }) @@ -307,6 +321,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 +414,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 +688,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 +712,7 @@ export async function executeWorkflowService( } const runSynchronousWorkflow = async (): Promise => { + if (detachable) registerManualExecutionAborter(executionId, timeoutController.abort) try { const snapshot = new ExecutionSnapshot( metadata, @@ -898,6 +919,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 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..e8c54fa8834 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` + ) + 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..3dd8a590907 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_FLAG}. 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..a9f20478f04 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 ?? {} } +export 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..d7e66074826 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, then answer `202` with the run receipt while it continues in the background; `0` waits until it settles. The run is admitted under asynchronous limits, a dropped connection no longer cancels it, and it can be watched or cancelled through the runs resource. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.', }, stream: { kind: 'boolean', 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", From 6dbae532a29b8c54771cf396d45a3c7511f3d356 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 20:32:23 -0700 Subject: [PATCH 2/3] fix(workflows): hand out a detached run's receipt only once the run can be read A detach could fire before the run's log row existed, so the runs resource answered 404 for the receipt just handed out. The response now waits for the logging session to start before detaching, and the wait timeout docs say how each transport carries the receipt. --- apps/docs/openapi-v2-workflows.json | 2 +- .../[workflowId]/execute/route.test.ts | 31 ++++++++++++++++ .../workflows/[workflowId]/execute/route.ts | 35 +++++++++++-------- apps/sim/lib/api/contracts/v2/workflows.ts | 2 +- .../sim/lib/logs/execution/logging-session.ts | 18 ++++++++++ .../lib/workflows/executor/execute-service.ts | 3 ++ .../protocol/workflow-run-follow.test.ts | 2 +- .../commands/protocol/workflow-run-follow.ts | 2 +- .../commands/protocol/workflow-run-wait.ts | 2 +- packages/sim-cli/src/generated/v2-api.ts | 2 +- .../testing/src/mocks/logging-session.mock.ts | 4 ++- 11 files changed, 82 insertions(+), 21 deletions(-) diff --git a/apps/docs/openapi-v2-workflows.json b/apps/docs/openapi-v2-workflows.json index 619d81d42f5..386196d06a8 100644 --- a/apps/docs/openapi-v2-workflows.json +++ b/apps/docs/openapi-v2-workflows.json @@ -14384,7 +14384,7 @@ "maximum": 604800 }, "waitTimeoutSeconds": { - "description": "Wait at most this many seconds for a synchronous run, then answer `202` with the run receipt while it continues in the background; `0` waits until it settles. The run is admitted under asynchronous limits, a dropped connection no longer cancels it, and it can be watched or cancelled through the runs resource. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.", + "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 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 8b7fa36c71c..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, @@ -417,6 +418,7 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => { ok: true, executionId: 'execution-123', pending, + logged: Promise.resolve(), cancel: vi.fn(), }) @@ -1229,6 +1231,35 @@ describe('POST /api/v2/workflows/[workflowId]/execute', () => { } }) + 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 { 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 1e5dbdb73ee..3b8a7d30dbe 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts @@ -156,21 +156,27 @@ function terminalRunResponse(result: ExecuteWorkflowServiceTerminalResult): Resp }) } -/** The pending run's result, or null when `afterMs` passes first; the run is never cancelled here. */ -async function settledWithin( - pending: Promise, +/** + * 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 pending + if (afterMs === null) return run.pending let timer: ReturnType | undefined const deadline = new Promise((resolve) => { timer = setTimeout(() => resolve(null), afterMs) }) try { - return await Promise.race([pending, deadline]) + const settled = await Promise.race([run.pending, deadline]) + if (settled) return settled } finally { clearTimeout(timer) } + return Promise.race([run.pending, run.logged.then(() => null)]) } /** @@ -189,12 +195,9 @@ function streamPendingRun( ): Response { let closed = false let heartbeatId: ReturnType | undefined - let detachId: ReturnType | undefined const stopTimers = () => { clearInterval(heartbeatId) - clearTimeout(detachId) heartbeatId = undefined - detachId = undefined } const abandon = () => { if (closed) return @@ -226,12 +229,16 @@ function streamPendingRun( heartbeatId = setInterval(() => { send({ type: 'heartbeat', timestamp: new Date().toISOString() }) }, WORKFLOW_RESULT_HEARTBEAT_INTERVAL_MS) - if (detachment?.afterMs != null) { + if (detachment) { const { receipt } = detachment - detachId = setTimeout(() => { - send({ type: 'detached', data: receipt }) - finish() - }, detachment.afterMs) + void settledBeforeDetach(pendingRun, detachment.afterMs).then( + (settled) => { + if (settled) return + send({ type: 'detached', data: receipt }) + finish() + }, + () => {} + ) } } @@ -618,7 +625,7 @@ export const POST = withRouteHandler( /** The ticket holds this server's capacity for as long as the run, not the response. */ ticketTransferred = true void result.pending.then(ticket.release, ticket.release) - const settled = await settledWithin(result.pending, detachment.afterMs) + const settled = await settledBeforeDetach(result, detachment.afterMs) if (settled) return terminalRunResponse(settled) return v2Data(detachment.receipt, { status: 202, diff --git a/apps/sim/lib/api/contracts/v2/workflows.ts b/apps/sim/lib/api/contracts/v2/workflows.ts index 882ae14a69a..a441518b467 100644 --- a/apps/sim/lib/api/contracts/v2/workflows.ts +++ b/apps/sim/lib/api/contracts/v2/workflows.ts @@ -1484,7 +1484,7 @@ export const v2ExecuteWorkflowBodySchema = z .max(MAX_WORKFLOW_EXECUTION_TIMEOUT_SECONDS) .optional() .describe( - 'Wait at most this many seconds for a synchronous run, then answer `202` with the run receipt while it continues in the background; `0` waits until it settles. The run is admitted under asynchronous limits, a dropped connection no longer cancels it, and it can be watched or cancelled through the runs resource. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.' + '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() 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/workflows/executor/execute-service.ts b/apps/sim/lib/workflows/executor/execute-service.ts index c9f2f92b364..5438f6a0776 100644 --- a/apps/sim/lib/workflows/executor/execute-service.ts +++ b/apps/sim/lib/workflows/executor/execute-service.ts @@ -195,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 } @@ -938,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 e8c54fa8834..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 @@ -216,7 +216,7 @@ describe('sim workflows run --wait-timeout', () => { 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` + `sim workflows runs wait run-1 --workflow ${WORKFLOW_ID} --wait-timeout 30` ) expect(process.exitCode).toBe(4) }) 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 3dd8a590907..ed43a1fb47f 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts +++ b/packages/sim-cli/src/commands/protocol/workflow-run-follow.ts @@ -200,7 +200,7 @@ function withEmbeddedFileReferences( /** 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_FLAG}. Stop it: sim workflows runs cancel ${runId} --workflow ${workflowId}` + `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}` } /** 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 a9f20478f04..23fdcaa36dc 100644 --- a/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts +++ b/packages/sim-cli/src/commands/protocol/workflow-run-wait.ts @@ -209,7 +209,7 @@ function runSpec(): CommandSpec { return CLI_CONTRACT.getWorkflowRun ?? {} } -export interface RunWait { +interface RunWait { runId: string workflowId: string /** When the wait began, for the progress line. */ diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts index d7e66074826..af76b92da10 100644 --- a/packages/sim-cli/src/generated/v2-api.ts +++ b/packages/sim-cli/src/generated/v2-api.ts @@ -18456,7 +18456,7 @@ export const V2_OPERATIONS = { waitTimeoutSeconds: { kind: 'integer', describe: - 'Wait at most this many seconds for a synchronous run, then answer `202` with the run receipt while it continues in the background; `0` waits until it settles. The run is admitted under asynchronous limits, a dropped connection no longer cancels it, and it can be watched or cancelled through the runs resource. Requires an API key or OAuth access token. Cannot be combined with `async` or `stream`.', + '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, From 83690adf0ce5e8066441183c95ad051ffb6b14f8 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 29 Oct 2026 21:07:47 -0700 Subject: [PATCH 3/3] feat(mothership): declare that workflows run outlasts --wait-timeout to the worker Sim now sends the workflow-run-wait-timeout client capability on chat turns, so the worker waits on a synchronous workflows run in slices only on a Sim whose CLI takes --wait-timeout there; an older Sim keeps its held run. The protocol copy is synced from the mothership contract that defines the token. --- apps/sim/lib/mothership/chat/payload.test.ts | 22 +++++++++++++++---- apps/sim/lib/mothership/chat/payload.ts | 7 ++++-- apps/sim/lib/mothership/generated/protocol.ts | 11 ++++++++-- 3 files changed, 32 insertions(+), 8 deletions(-) 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; /**