diff --git a/apps/docs/content/docs/workflows/deployment/api.mdx b/apps/docs/content/docs/workflows/deployment/api.mdx index 412b1e76c63..481ac6a47af 100644 --- a/apps/docs/content/docs/workflows/deployment/api.mdx +++ b/apps/docs/content/docs/workflows/deployment/api.mdx @@ -14,6 +14,17 @@ You need workspace **Admin** access and the applicable deployment permission. Te Ordinary API calls use the active snapshot. Editing the draft does not change that version. Deploy an update before expecting a normal API call to use your changes. +### Run an earlier version [#run-a-version] + +To run a specific deployment version instead of the active one, name its number in `run.version`. The workflow must still be deployed, and the caller needs workspace **Write** access with an API key or OAuth access token; anonymous public-API callers cannot pin a version. An unknown version returns `404`. The run's status reports the version it ran in `version`. + +```bash +curl -X POST 'https://www.sim.ai/api/v2/workflows/{workflow-id}/execute' \ + -H "X-API-Key: $SIM_API_KEY" \ + -H 'Content-Type: application/json' \ + -d '{"input":{"message":"Summarize this article"},"run":{"source":"deployment","version":3}}' +``` + ## Authentication Send a valid Sim API key in `X-API-Key`, or use a supported OAuth access token. Deployed workflow execution requires access to the workflow's workspace; OAuth callers also need the operation's `api:write` scope. See [API authentication](/api-reference/authentication). @@ -249,6 +260,8 @@ A completed response includes the result when requested: Stop polling on `completed`, `failed`, or `cancelled`. A paused Human run requires submitted input; a time pause can resume automatically. See [Human](/workflows/blocks/human-in-the-loop) and [Wait](/workflows/blocks/wait). Resuming an originally streaming run through the v2 resume endpoint returns an async receipt, not an SSE continuation. +A paused run lists every point it waits at in `paused.points`: the block, whether it is a `human` or `time` pause, its resume state, any parallel branch or loop iteration, its place in the resume queue, and the `inputFields` a Human pause asks for. With `includeOutput=true`, each point also carries the reviewer `display` content with resolved secrets redacted; organization permission groups that hide execution data withhold it, along with `inputFields`, whose labels and descriptions are resolved from run data. Resume a point by sending its `contextId` to `POST /api/v2/workflows/{workflow-id}/runs/{runId}/resume`. + ## Select outputs and handle large values [#oversized-outputs] `selectedOutputs` can project smaller fields. For a synchronous v2 call, selected values appear in `data.blockOutputs`, separately from `data.output`. diff --git a/apps/docs/openapi-v2-workflows.json b/apps/docs/openapi-v2-workflows.json index 58a725eeae1..8d7fc7cb356 100644 --- a/apps/docs/openapi-v2-workflows.json +++ b/apps/docs/openapi-v2-workflows.json @@ -3958,7 +3958,7 @@ "post": { "operationId": "executeWorkflowV2", "summary": "Execute workflow", - "description": "Run the active deployment, or set `run.source: \"manual\"` to run the saved draft. Deployed execution requires workspace read access and OAuth `api:write` scope or an API key; public deployments also allow anonymous synchronous or streamed runs. Manual runs require personal-key or OAuth write access and cannot be async.\n\nUse `async: true` for a queue receipt, or `stream: true` for SSE. For a synchronous run with both flags false, send `Accept: application/x-ndjson` for heartbeat frames and a final frame containing `data`. With `waitTimeoutSeconds`, a synchronous run still going when the wait elapses continues in the background and answers with its run receipt instead: `202` on a JSON response, or a closing `detached` frame on an NDJSON stream, which keeps its `200`. Timeouts return `200` with `status: \"failed\"` and `error.code: \"TIMEOUT\"`; inspect run status. See [Streaming](/workflows/deployment/api#streaming) for event formats. Input descriptions specify compatible modes; invalid combinations return `400`.\n\nSupply `X-Run-Id` to prevent duplicate execution. Reuse returns `409`, without replaying results. A `503` with `error.details.code: \"ASYNC_ENQUEUE_AMBIGUOUS\"` omits `Retry-After` because the run may already have started. Reconcile using the returned run ID; do not start another run with a new or omitted ID.\n\nOAuth scope: `api:write`.", + "description": "Run the active deployment, set `run: { source: \"deployment\", version }` to run an earlier deployment version, or set `run.source: \"manual\"` to run the saved draft. Deployed execution requires workspace read access and OAuth `api:write` scope or an API key; public deployments also allow anonymous synchronous or streamed runs. Running a chosen version requires workspace write access and an API key or OAuth access token; an unknown version returns `404`. Manual runs require personal-key or OAuth write access and cannot be async.\n\nUse `async: true` for a queue receipt, or `stream: true` for SSE. For a synchronous run with both flags false, send `Accept: application/x-ndjson` for heartbeat frames and a final frame containing `data`. With `waitTimeoutSeconds`, a synchronous run still going when the wait elapses continues in the background and answers with its run receipt instead: `202` on a JSON response, or a closing `detached` frame on an NDJSON stream, which keeps its `200`. Timeouts return `200` with `status: \"failed\"` and `error.code: \"TIMEOUT\"`; inspect run status. See [Streaming](/workflows/deployment/api#streaming) for event formats. Input descriptions specify compatible modes; invalid combinations return `400`.\n\nSupply `X-Run-Id` to prevent duplicate execution. Reuse returns `409`, without replaying results. A `503` with `error.details.code: \"ASYNC_ENQUEUE_AMBIGUOUS\"` omits `Retry-After` because the run may already have started. Reconcile using the returned run ID; do not start another run with a new or omitted ID.\n\nOAuth scope: `api:write`.", "x-sim-operation": "workflows.execute", "x-oauth-scope": "api:write", "tags": ["Workflows"], @@ -4405,7 +4405,7 @@ "get": { "operationId": "getWorkflowRunV2", "summary": "Get workflow run", - "description": "Read current run state with workspace read access. Request final output and files with `includeOutput`, or block outputs with `selectedOutputs`. Organization permission groups can hide output, files, or cost.\n\n`includeFileBase64` inlines file bytes. A file or the total above 16 MiB returns `413` naming a download path; download large files individually. `HEAD` checks access with the same authorization as `GET` but skips side effects, returning an empty `200` without payload headers on success.\n\nOAuth scope: `api:read`.", + "description": "Read current run state with workspace read access. Request final output and files with `includeOutput`, or block outputs with `selectedOutputs`. Organization permission groups can hide output, files, or cost. A paused run lists every pause point in `paused.points`, with the fields a human pause asks for and its content with resolved secrets redacted; resume one with Resume workflow run.\n\n`includeFileBase64` inlines file bytes. A file or the total above 16 MiB returns `413` naming a download path; download large files individually. `HEAD` checks access with the same authorization as `GET` but skips side effects, returning an empty `200` without payload headers on success.\n\nOAuth scope: `api:read`.", "x-sim-operation": "workflows.runs.read", "x-oauth-scope": "api:read", "tags": ["Workflow Runs"], @@ -4517,6 +4517,7 @@ "startedAt": "2026-08-09T18:04:10.000Z", "endedAt": "2026-08-09T18:04:11.000Z", "durationMs": 1000, + "version": 3, "paused": null, "cost": { "total": 12 @@ -4691,7 +4692,7 @@ "post": { "operationId": "resumeWorkflowRunV2", "summary": "Resume workflow run", - "description": "Resume one human-in-the-loop pause with workspace write access. The resumed attempt receives a new run ID and returns a synchronous result or queue receipt. A run that originally streamed resumes asynchronously through v2; this endpoint returns no SSE stream.\n\nOAuth scope: `api:write`.", + "description": "Resume one human-in-the-loop pause with workspace write access, naming a `contextId` from the run's `paused.points`. The resumed attempt receives a new run ID and returns a synchronous result or queue receipt. A run that originally streamed resumes asynchronously through v2; this endpoint returns no SSE stream.\n\nOAuth scope: `api:write`.", "x-sim-operation": "workflows.runs.resume", "x-oauth-scope": "api:write", "tags": ["Workflow Runs"], @@ -14916,7 +14917,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; `waitTimeoutSeconds` lets a synchronous one continue in the background.", + "description": "Workflow state and entry point to execute. Omit for the active deployment; `source: \"deployment\"` with `version` runs that deployment version. 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", @@ -14924,7 +14925,13 @@ "source": { "type": "string", "const": "deployment", - "description": "Execute the active deployed workflow state." + "description": "Execute a deployed workflow version." + }, + "version": { + "description": "Deployment version to execute instead of the active one. Requires an API key or OAuth access token with workspace write access, and the workflow must be deployed. An unknown version returns `404`.", + "type": "integer", + "minimum": 1, + "maximum": 2147483647 } }, "required": ["source"], @@ -15244,6 +15251,222 @@ } ] }, + "WorkflowRunPausePoint": { + "type": "object", + "properties": { + "contextId": { + "type": "string", + "description": "Pause context to name in `contextId` when resuming this point." + }, + "blockId": { + "type": "string", + "description": "Workflow block that paused." + }, + "blockName": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Name of the block in the deployment version the run executed, or null for a manual run or when the version no longer exists." + }, + "kind": { + "type": "string", + "enum": ["human", "time"], + "description": "`human` waits for a resume request; `time` resumes automatically at `resumeAt`." + }, + "status": { + "type": "string", + "enum": ["paused", "queued", "resuming", "resumed", "failed"], + "description": "Resume state of this point: waiting, queued behind another resume, resuming, resumed, or a failed resume attempt." + }, + "registeredAt": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "ISO 8601 timestamp when the run paused at this point, or null for a point recorded without one.", + "format": "date-time" + }, + "resumeAt": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "ISO 8601 automatic-resume timestamp for a time pause, otherwise null.", + "format": "date-time" + }, + "parallel": { + "anyOf": [ + { + "type": "object", + "properties": { + "parallelId": { + "type": "string", + "description": "Parallel block containing the pause." + }, + "branchIndex": { + "type": "integer", + "minimum": -9007199254740991, + "maximum": 9007199254740991, + "description": "Zero-based branch that paused." + }, + "branchTotal": { + "anyOf": [ + { + "type": "integer", + "minimum": -9007199254740991, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ], + "description": "Number of branches, or null when unrecorded." + } + }, + "required": ["parallelId", "branchIndex", "branchTotal"], + "additionalProperties": false + }, + { + "type": "null" + } + ], + "description": "Parallel branch of the pause, or null outside a parallel." + }, + "loop": { + "anyOf": [ + { + "type": "object", + "properties": { + "loopId": { + "type": "string", + "description": "Loop block containing the pause." + }, + "iteration": { + "type": "integer", + "minimum": -9007199254740991, + "maximum": 9007199254740991, + "description": "Zero-based iteration that paused." + } + }, + "required": ["loopId", "iteration"], + "additionalProperties": false + }, + { + "type": "null" + } + ], + "description": "Loop iteration of the pause, or null outside a loop." + }, + "inputFields": { + "anyOf": [ + { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowRunPauseInputField" + } + }, + { + "type": "null" + } + ], + "description": "Fields a human pause asks for, without default values. Labels and descriptions are resolved, so they are withheld when organization permission groups hide execution data. Null then, for a time pause, a pause with no declared form, fields over 64 KiB or not provably free of secrets, or one recorded before pause details were captured." + }, + "display": { + "anyOf": [ + { + "description": "Reviewer content value." + }, + { + "type": "null" + } + ], + "description": "Content the pause presents to its reviewer, with the run's resolved secrets redacted. Null unless `includeOutput=true`, when hidden by organization permission groups, for a time pause, for content over 64 KiB or not provably free of secrets, or for a pause recorded before pause details were captured." + }, + "queuePosition": { + "anyOf": [ + { + "type": "integer", + "minimum": -9007199254740991, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ], + "description": "Position among pending resume requests for this run, or null when none waits." + } + }, + "required": [ + "contextId", + "blockId", + "blockName", + "kind", + "status", + "registeredAt", + "resumeAt", + "parallel", + "loop", + "inputFields", + "display", + "queuePosition" + ], + "additionalProperties": false, + "title": "Workflow run pause point", + "description": "One point at which a paused run waits." + }, + "WorkflowRunPauseInputField": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Key to supply this field under in the resume `input`." + }, + "label": { + "type": "string", + "description": "Display label for the field." + }, + "type": { + "type": "string", + "description": "Declared value type, such as `string`, `number`, or `boolean`." + }, + "description": { + "description": "Field help text, when configured.", + "type": "string" + }, + "placeholder": { + "description": "Placeholder text, when configured.", + "type": "string" + }, + "required": { + "type": "boolean", + "description": "Whether the field is marked required." + }, + "options": { + "description": "Selectable options, when the field offers a fixed set.", + "type": "array", + "items": { + "description": "One selectable option." + } + } + }, + "required": ["name", "label", "type", "required"], + "additionalProperties": false, + "title": "Pause input field", + "description": "One field a human pause asks the resumer to fill." + }, "V2RunFile": { "type": "object", "properties": { @@ -15361,6 +15584,19 @@ ], "description": "Run duration in milliseconds, or null while active." }, + "version": { + "anyOf": [ + { + "type": "integer", + "minimum": -9007199254740991, + "maximum": 9007199254740991 + }, + { + "type": "null" + } + ], + "description": "Deployment version the run executed. Null for a manual run, a queued run not yet started, or a run whose version no longer exists." + }, "paused": { "anyOf": [ { @@ -15437,6 +15673,13 @@ "resumedCount": { "type": "number", "description": "Number of pause points that have resumed." + }, + "points": { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowRunPausePoint" + }, + "description": "Every point the run is paused at, earliest first, at most 100. Resume one with Resume workflow run, naming its `contextId` and supplying `input`." } }, "required": [ @@ -15447,7 +15690,8 @@ "blockedOnBlockId", "automaticResumeWaitingReason", "pausePointCount", - "resumedCount" + "resumedCount", + "points" ], "additionalProperties": false }, @@ -15538,6 +15782,7 @@ "startedAt", "endedAt", "durationMs", + "version", "paused", "cost", "error", @@ -15571,6 +15816,7 @@ "startedAt": "2026-08-09T18:04:10.000Z", "endedAt": "2026-08-09T18:04:11.000Z", "durationMs": 1000, + "version": 3, "paused": null, "cost": { "total": 12 diff --git a/apps/docs/scripts/fixtures/content-baseline.json b/apps/docs/scripts/fixtures/content-baseline.json index 94df7334dd3..88a1d78e393 100644 --- a/apps/docs/scripts/fixtures/content-baseline.json +++ b/apps/docs/scripts/fixtures/content-baseline.json @@ -181727,6 +181727,13 @@ ], "expressions": [], "codeFences": [ + { + "lang": "bash", + "meta": null, + "sha256": "32c20ca5dbc0941290361d3ddba31d3585a80493e65c486960e2660fae4c03d0", + "bytes": 243, + "line": 21 + }, { "lang": "bash", "meta": null, 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 24fbbcd3417..a97fe7c4d0a 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/execute/route.ts @@ -40,7 +40,10 @@ import { executeManualWorkflowFromBlockOperation, executeManualWorkflowOperation, } from '@/lib/workflows/application/execute-manual-workflow' -import { executeWorkflowOperation } from '@/lib/workflows/application/execute-workflow' +import { + executeWorkflowOperation, + executeWorkflowVersionOperation, +} from '@/lib/workflows/application/execute-workflow' import { workflowOperations } from '@/lib/workflows/application/operations' import { type ExecuteWorkflowServiceFailure, @@ -316,6 +319,8 @@ type V2ExecuteWorkflowRouteContext = { * * - Auth: `X-API-Key` (personal/workspace) or the anonymous public-API path for * workflows deployed with `isPublicApi` (actor = owner; sync/stream only). + * - `run: { source: 'deployment', version }` runs that deployment version under + * the write-role `workflows.execute_version` operation; keyed callers only. * - `async: true` (body flag — v2 has no mode headers) → 202 * `{ data: { runId, statusUrl } }`; poll the v2 runs resource. * - `stream: true` → SSE passthrough (no `{data}` envelope on event frames). @@ -422,6 +427,7 @@ export const POST = withRouteHandler( if (!parsed.success) return parsed.response const body = parsed.data.body const manualRun = body.run?.source === 'manual' ? body.run : undefined + const pinnedVersion = body.run?.source === 'deployment' ? body.run.version : undefined if (manualRun && !apiKeyPrincipal) { return v2Error( @@ -439,6 +445,9 @@ export const POST = withRouteHandler( ) { return v2Error('BAD_REQUEST', 'input and run.entry.useMockPayload cannot be combined') } + if (pinnedVersion !== undefined && isPublicApiAccess) { + return v2Error('BAD_REQUEST', 'run.version requires an OAuth access token or API key') + } const detachable = body.waitTimeoutSeconds !== undefined if (detachable && (body.async || body.stream)) { @@ -544,6 +553,17 @@ export const POST = withRouteHandler( }, request: req, }) + } else if (pinnedVersion !== undefined) { + result = await executeWorkflowVersionOperation.execute({ + principal: apiKeyPrincipal, + input: { + ...commonInput, + input: body.input ?? {}, + mode: body.async ? 'async' : body.stream ? 'stream' : syncMode, + version: pinnedVersion, + }, + request: req, + }) } else { result = await executeWorkflowOperation.execute({ principal: apiKeyPrincipal, diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.test.ts b/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.test.ts index 12c3f8c0249..54c0fc63784 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.test.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.test.ts @@ -59,6 +59,7 @@ const baseStatus = { startedAt: '2026-07-31T00:00:00.000Z', endedAt: '2026-07-31T00:00:05.000Z', totalDurationMs: 5000, + deploymentVersion: null, paused: null, cost: { total: 0.02 }, error: 'Send Email: Invalid credentials', diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.ts b/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.ts index dab2588457e..9bbf5c353a6 100644 --- a/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.ts +++ b/apps/sim/app/api/v2/workflows/[workflowId]/runs/[runId]/route.ts @@ -45,7 +45,13 @@ export const GET = defineV2JsonRoute({ startedAt: status.startedAt, endedAt: status.endedAt, durationMs: status.totalDurationMs, - paused: status.paused ? v2WorkflowRunStatusSchema.shape.paused.parse(status.paused) : null, + version: status.deploymentVersion ?? null, + paused: status.paused + ? v2WorkflowRunStatusSchema.shape.paused.parse({ + ...status.paused, + points: status.pausePoints ?? [], + }) + : null, cost: status.cost, error: status.error ? classifyExecutionError(new Error(status.error)) : null, output: status.finalOutput, diff --git a/apps/sim/app/api/workflows/[id]/executions/[executionId]/route.test.ts b/apps/sim/app/api/workflows/[id]/executions/[executionId]/route.test.ts index f6984733410..1a5ffa73f25 100644 --- a/apps/sim/app/api/workflows/[id]/executions/[executionId]/route.test.ts +++ b/apps/sim/app/api/workflows/[id]/executions/[executionId]/route.test.ts @@ -53,6 +53,7 @@ describe('internal execution status route projection subject', () => { startedAt: '2026-08-05T12:00:00.000Z', endedAt: null, totalDurationMs: null, + deploymentVersion: null, paused: null, cost: null, error: null, diff --git a/apps/sim/background/deployed-run-variables.integration.ts b/apps/sim/background/deployed-run-variables.integration.ts new file mode 100644 index 00000000000..1875397aee3 --- /dev/null +++ b/apps/sim/background/deployed-run-variables.integration.ts @@ -0,0 +1,287 @@ +/** + * Scheduled and webhook runs against real PostgreSQL: each executes its deployed version with + * the variables that version was deployed with, not the editor's saved ones, except for a + * version deployed before deployments snapshotted variables, which keeps the saved ones. + */ +import { serializePrincipal } from '@sim/auth/principal' +import { db } from '@sim/db' +import { + permissions, + user, + webhook, + workflow, + workflowDeploymentVersion, + workflowExecutionLogs, + workflowExecutionSnapshots, + workflowSchedule, + workspace, +} from '@sim/db/schema' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { generateId } from '@sim/utils/id' +import { eq, inArray } from 'drizzle-orm' +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' +import { resolveBillingAttribution } from '@/lib/billing/core/attribution' +import { createWebhookExecutionPrincipal } from '@/lib/webhooks/execution-principal' +import { getWorkflowExecutionStatus } from '@/lib/workflows/executor/execution-status' +import { executeScheduleJob } from '@/background/schedule-execution' +import { executeWebhookJob } from '@/background/webhook-execution' + +const ownerId = generateId() +const workspaceId = generateId() +const workflowIds: string[] = [] + +function variables(workflowId: string, greeting: string) { + return { + greeting: { id: 'greeting', workflowId, name: 'greeting', type: 'plain', value: greeting }, + } +} + +const SCHEDULE_TRIGGER = { + id: 'trigger', + type: 'schedule', + name: 'Schedule', + enabled: true, + position: { x: 0, y: 0 }, + outputs: {}, + subBlocks: { + scheduleType: { id: 'scheduleType', type: 'dropdown', value: 'custom' }, + cronExpression: { id: 'cronExpression', type: 'short-input', value: '0 0 * * *' }, + }, +} + +const WEBHOOK_TRIGGER = { + id: 'trigger', + type: 'generic_webhook', + name: 'Webhook', + enabled: true, + position: { x: 0, y: 0 }, + outputs: {}, + subBlocks: {}, + triggerMode: true, +} + +/** Trigger into a Response echoing the `greeting` variable; `variables` omitted for a legacy snapshot. */ +function deployedState( + trigger: typeof SCHEDULE_TRIGGER | typeof WEBHOOK_TRIGGER, + snapshotVariables: Record | undefined +) { + return { + blocks: { + trigger, + respond: { + id: 'respond', + type: 'response', + name: 'Respond', + enabled: true, + position: { x: 300, y: 0 }, + outputs: {}, + subBlocks: { + dataMode: { id: 'dataMode', type: 'dropdown', value: 'json' }, + data: { id: 'data', type: 'code', value: '{"greeting": ""}' }, + status: { id: 'status', type: 'short-input', value: '200' }, + }, + }, + }, + edges: [{ id: 'trigger-respond', source: 'trigger', target: 'respond' }], + loops: {}, + parallels: {}, + ...(snapshotVariables ? { variables: snapshotVariables } : {}), + } +} + +/** + * A workflow whose saved variables differ from the deployed ones. With `superseded`, the + * returned version is no longer active: a later one with its own variables is. + */ +async function deployWorkflow( + trigger: typeof SCHEDULE_TRIGGER | typeof WEBHOOK_TRIGGER, + { snapshotsVariables, superseded = false }: { snapshotsVariables: boolean; superseded?: boolean } +) { + const workflowId = generateId() + const versionId = generateId() + workflowIds.push(workflowId) + const now = new Date() + await db.insert(workflow).values({ + id: workflowId, + userId: ownerId, + workspaceId, + name: `Deployed variables fixture ${workflowId}`, + isDeployed: true, + deployedAt: now, + variables: variables(workflowId, 'saved'), + lastSynced: now, + createdAt: now, + updatedAt: now, + }) + await db.insert(workflowDeploymentVersion).values({ + id: versionId, + workflowId, + version: 1, + state: deployedState( + trigger, + snapshotsVariables ? variables(workflowId, 'deployed') : undefined + ), + isActive: !superseded, + }) + if (superseded) { + await db.insert(workflowDeploymentVersion).values({ + id: generateId(), + workflowId, + version: 2, + state: deployedState(trigger, variables(workflowId, 'active')), + isActive: true, + }) + } + return { workflowId, versionId } +} + +async function settledOutput(workflowId: string, executionId: string) { + return vi.waitFor( + async () => { + const status = await getWorkflowExecutionStatus({ + workflowId, + executionId, + includeOutput: true, + selectedOutputs: [], + workspaceId, + viewerUserId: ownerId, + }) + if (status?.status !== 'completed') { + throw new Error(`Run ${executionId} has not completed: ${status?.status} ${status?.error}`) + } + return status.finalOutput + }, + { timeout: 20_000, interval: 250 } + ) +} + +async function runSchedule(workflowId: string, versionId: string) { + const scheduleId = generateId() + const executionId = generateId() + const now = new Date() + await db.insert(workflowSchedule).values({ + id: scheduleId, + workflowId, + deploymentVersionId: versionId, + blockId: 'trigger', + cronExpression: '0 0 * * *', + triggerType: 'schedule', + lastQueuedAt: now, + createdAt: now, + updatedAt: now, + }) + await executeScheduleJob({ + scheduleId, + workflowId, + workspaceId, + billingAttribution: await resolveBillingAttribution({ actorUserId: ownerId, workspaceId }), + executionId, + blockId: 'trigger', + deploymentVersionId: versionId, + cronExpression: '0 0 * * *', + now: now.toISOString(), + }) + return settledOutput(workflowId, executionId) +} + +async function runWebhook(workflowId: string, versionId: string) { + const webhookId = generateId() + const executionId = generateId() + await db.insert(webhook).values({ + id: webhookId, + workflowId, + deploymentVersionId: versionId, + blockId: 'trigger', + path: `deployed-variables-${webhookId}`, + provider: 'generic', + providerConfig: {}, + }) + await executeWebhookJob({ + webhookId, + workflowId, + principal: serializePrincipal( + createWebhookExecutionPrincipal({ webhookId, workflowId, workspaceId, provider: 'generic' }) + ), + userId: ownerId, + billingAttribution: await resolveBillingAttribution({ actorUserId: ownerId, workspaceId }), + executionId, + provider: 'generic', + body: { event: 'ping' }, + headers: {}, + method: 'POST', + path: `deployed-variables-${webhookId}`, + blockId: 'trigger', + deploymentVersionId: versionId, + workspaceId, + }) + return settledOutput(workflowId, executionId) +} + +beforeAll(async () => { + const now = new Date() + await db.insert(user).values({ + id: ownerId, + name: 'Deployed variables fixture', + email: `${ownerId}@deployed-variables.test`, + emailVerified: true, + createdAt: now, + updatedAt: now, + }) + await insertWorkspaceFixture(db, { + id: workspaceId, + name: 'Deployed variables fixture', + ownerId, + billedAccountUserId: ownerId, + }) + await db.insert(permissions).values({ + id: generateId(), + userId: ownerId, + entityType: 'workspace', + entityId: workspaceId, + permissionType: 'admin', + }) +}) + +afterAll(async () => { + if (workflowIds.length > 0) { + await db + .delete(workflowExecutionLogs) + .where(inArray(workflowExecutionLogs.workflowId, workflowIds)) + await db + .delete(workflowExecutionSnapshots) + .where(inArray(workflowExecutionSnapshots.workflowId, workflowIds)) + } + await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId)) + await db.delete(user).where(eq(user.id, ownerId)) + await db.$client.end() +}) + +describe('variables a queued deployed run executes with', () => { + it('runs a schedule with its deployed variables, not the saved ones', async () => { + const { workflowId, versionId } = await deployWorkflow(SCHEDULE_TRIGGER, { + snapshotsVariables: true, + }) + await expect(runSchedule(workflowId, versionId)).resolves.toMatchObject({ + data: { greeting: 'deployed' }, + }) + }) + + it('runs a webhook delivery admitted on an earlier version with that version variables', async () => { + const { workflowId, versionId } = await deployWorkflow(WEBHOOK_TRIGGER, { + snapshotsVariables: true, + superseded: true, + }) + await expect(runWebhook(workflowId, versionId)).resolves.toMatchObject({ + data: { greeting: 'deployed' }, + }) + }) + + it('keeps the saved variables for a version deployed before variables were snapshotted', async () => { + const { workflowId, versionId } = await deployWorkflow(SCHEDULE_TRIGGER, { + snapshotsVariables: false, + }) + await expect(runSchedule(workflowId, versionId)).resolves.toMatchObject({ + data: { greeting: 'saved' }, + }) + }) +}) diff --git a/apps/sim/background/schedule-execution.ts b/apps/sim/background/schedule-execution.ts index 533d5eb0063..eeabc810762 100644 --- a/apps/sim/background/schedule-execution.ts +++ b/apps/sim/background/schedule-execution.ts @@ -37,7 +37,10 @@ import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base6 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core' import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence' import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure' -import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils' +import { + loadDeployedWorkflowState, + resolveDeployedRunVariables, +} from '@/lib/workflows/persistence/utils' import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications' import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons' import { @@ -622,7 +625,7 @@ async function runWorkflowExecution({ metadata, workflowRecord, input, - workflowRecord.variables || {}, + resolveDeployedRunVariables(deployedData, workflowRecord.variables), [] ) diff --git a/apps/sim/background/webhook-execution.ts b/apps/sim/background/webhook-execution.ts index 660840b0cf2..bd4387132b7 100644 --- a/apps/sim/background/webhook-execution.ts +++ b/apps/sim/background/webhook-execution.ts @@ -84,6 +84,7 @@ import { import { loadDeployedWorkflowState, loadWorkflowDeploymentVersionState, + resolveDeployedRunVariables, } from '@/lib/workflows/persistence/utils' import { WEBHOOK_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits' import { getBlock } from '@/blocks' @@ -867,8 +868,6 @@ async function executeWebhookJobInternal( throw new Error(`Workflow ${payload.workflowId} has no associated workspace`) } - const workflowVariables = (workflowRecord.variables as Record) || {} - let deploymentVersionId: string | undefined /** * Flipped immediately before `executeWorkflowCore` is invoked. While false, @@ -917,6 +916,7 @@ async function executeWebhookJobInternal( } const { blocks, edges, loops, parallels } = workflowData + const workflowVariables = resolveDeployedRunVariables(workflowData, workflowRecord.variables) deploymentVersionId = 'deploymentVersionId' in workflowData ? (workflowData.deploymentVersionId as string) diff --git a/apps/sim/background/workflow-execution.ts b/apps/sim/background/workflow-execution.ts index 0aa2c684e8a..809b9493ec2 100644 --- a/apps/sim/background/workflow-execution.ts +++ b/apps/sim/background/workflow-execution.ts @@ -32,6 +32,7 @@ import { RESERVATION_TTL_BUFFER_MS, } from '@/lib/core/execution-limits' import type { RequestAttribution } from '@/lib/core/utils/request-attribution' +import { withDatabaseReadRetry } from '@/lib/db/read-retry' import { isWorkflowStateRejection, type PreprocessExecutionError, @@ -47,6 +48,11 @@ import { isRecordedWorkflowFailure, resolveDeletedExecutionScope, } from '@/lib/workflows/executor/recorded-failure' +import { + loadDeployedWorkflowState, + loadWorkflowDeploymentVersionState, + resolveDeployedRunVariables, +} from '@/lib/workflows/persistence/utils' import { WORKFLOW_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits' import { ExecutionSnapshot } from '@/executor/execution/snapshot' import type { ExecutionMetadata } from '@/executor/execution/types' @@ -106,6 +112,11 @@ export type WorkflowExecutionPayload = { */ enforceCredentialAccess?: boolean isPublicApiAccess?: boolean + /** + * Deployment version validated by the enqueuing use case. Only the id rides the + * payload; jobs queued without it run the active deployment. + */ + pinnedDeploymentVersionId?: string } type LegacyWorkflowExecutionPayload = Omit & { @@ -272,6 +283,23 @@ export async function executeWorkflowJob( logger.info(`Preprocessing passed. Using actor: ${actorUserId}`, { requestId }) const workflow = preprocessResult.workflowRecord! + /** + * The version this run executes, loaded once: the core runs this same version's + * blocks, so a deploy landing mid-setup cannot pair one version's variables with + * another's graph. + */ + const pinnedDeploymentVersionId = payload.pinnedDeploymentVersionId + const deployedState = await withDatabaseReadRetry( + () => + pinnedDeploymentVersionId + ? loadWorkflowDeploymentVersionState( + workflowId, + pinnedDeploymentVersionId, + workspaceId + ) + : loadDeployedWorkflowState(workflowId, workspaceId), + { label: 'loadDeployedRunState' } + ) const metadata: ExecutionMetadata = { requestId, @@ -286,6 +314,7 @@ export async function executeWorkflowJob( triggerType: payload.triggerType || 'api', triggerBlockId: payload.triggerBlockId, useDraftState: false, + pinnedDeploymentVersionId: deployedState.deploymentVersionId, startTime: new Date().toISOString(), isClientSession: false, enforceCredentialAccess: payload.enforceCredentialAccess ?? false, @@ -299,7 +328,7 @@ export async function executeWorkflowJob( metadata, workflow, payload.input, - workflow.variables || {}, + resolveDeployedRunVariables(deployedState, workflow.variables), [] ) diff --git a/apps/sim/executor/execution/engine.ts b/apps/sim/executor/execution/engine.ts index ffb20f432f9..092c830e2a6 100644 --- a/apps/sim/executor/execution/engine.ts +++ b/apps/sim/executor/execution/engine.ts @@ -11,6 +11,7 @@ import { serializePauseSnapshot, } from '@/executor/execution/snapshot-serializer' import type { SerializableExecutionState } from '@/executor/execution/types' +import { capturePauseDisplay } from '@/executor/human-in-the-loop/utils' import type { NodeExecutionOrchestrator } from '@/executor/orchestrators/node' import type { ExecutionContext, @@ -651,7 +652,9 @@ export class ExecutionEngine { this.context.metadata.status = 'paused' const snapshotSeed = serializePauseSnapshot(this.context, [], this.dag, this.edgeManager) + const registry = this.context.resolvedSecretTraceRegistry const pausePoints: PausePoint[] = Array.from(this.pausedBlocks.values()).map((pause) => ({ + ...capturePauseDisplay(pause, registry), contextId: pause.contextId, blockId: pause.blockId, response: pause.response, diff --git a/apps/sim/executor/execution/types.ts b/apps/sim/executor/execution/types.ts index 905dc3ac6ec..36265be745e 100644 --- a/apps/sim/executor/execution/types.ts +++ b/apps/sim/executor/execution/types.ts @@ -75,6 +75,12 @@ export interface ExecutionMetadata { variables?: Record deploymentVersionId?: string } + /** + * Deployment version a deployed run executes instead of the active one, chosen + * and validated by the application use case. A resumed run ignores it: resume + * reloads the version its durable log admitted. + */ + pinnedDeploymentVersionId?: string largeValueExecutionIds?: string[] largeValueKeys?: string[] fileKeys?: string[] diff --git a/apps/sim/executor/human-in-the-loop/utils.ts b/apps/sim/executor/human-in-the-loop/utils.ts index ef9c8501c7a..2c000374b77 100644 --- a/apps/sim/executor/human-in-the-loop/utils.ts +++ b/apps/sim/executor/human-in-the-loop/utils.ts @@ -1,6 +1,22 @@ -import type { ExecutionContext, LoopPauseScope, ParallelPauseScope } from '@/executor/types' +import { isRecordLike } from '@sim/utils/object' +import type { + ExecutionContext, + LoopPauseScope, + ParallelPauseScope, + PauseInputField, + PauseMetadata, + PausePoint, +} from '@/executor/types' +import { projectResolvedSecretModelJsonContent } from '@/executor/utils/resolved-secret-content-projection' +import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' import { buildBranchNodeId } from '@/executor/utils/subflow-node-id-codec' +/** Per-part ceiling on captured display content and fields; larger parts are withheld, not truncated. */ +const MAX_PAUSE_DISPLAY_BYTES = 64 * 1024 + +/** Response keys that drive resume and rendering rather than describe what the reviewer sees. */ +const PAUSE_RESPONSE_CONTROL_KEYS = new Set(['_resume', 'operation', 'inputFormat', 'submission']) + interface NodeMetadataLike { nodeId: string loopId?: string @@ -59,3 +75,62 @@ export function mapNodeMetadataToPauseScopes( loopScope, } } + +function toPauseInputField(field: unknown): PauseInputField | null { + if (!isRecordLike(field) || typeof field.name !== 'string' || !field.name.trim()) return null + return { + name: field.name, + label: typeof field.label === 'string' ? field.label : field.name, + type: typeof field.type === 'string' ? field.type : 'string', + ...(typeof field.description === 'string' ? { description: field.description } : {}), + ...(typeof field.placeholder === 'string' ? { placeholder: field.placeholder } : {}), + required: field.required === true, + ...(Array.isArray(field.options) ? { options: field.options } : {}), + } +} + +/** + * The reviewer-facing view of a pause, scrubbed of every secret the run resolved. + * + * A pause's `response` is persisted as the block produced it, so a reference such as + * `{{API_KEY}}` in its configuration is stored resolved. This projects the parts a caller + * reading the run may see — the content without resume links, control keys, or API-mode status + * and headers, and the input fields without their default values — through the run's committed + * secret matches. Content and fields are proven separately: a part whose projection is + * unprovable (no registry, an incomplete one, or over {@link MAX_PAUSE_DISPLAY_BYTES}) is null + * while the other may survive. Time pauses carry no reviewer content. + */ +export function capturePauseDisplay( + pause: Pick, + registry: ResolvedSecretTraceRegistry | undefined +): Pick { + const withheld = { displayScrubbed: true, display: null, inputFields: null } + if (pause.pauseKind !== 'human') return withheld + + const data: unknown = isRecordLike(pause.response) ? pause.response.data : undefined + const content = isRecordLike(data) + ? Object.fromEntries( + Object.entries(data).filter(([key]) => !PAUSE_RESPONSE_CONTROL_KEYS.has(key)) + ) + : (data ?? null) + const inputFields = + isRecordLike(data) && Array.isArray(data.inputFormat) + ? data.inputFormat.flatMap((field) => toPauseInputField(field) ?? []) + : null + + /** Each part is projected on its own, so content too large to prove withholds only itself. */ + const project = (value: unknown): unknown => { + const projection = projectResolvedSecretModelJsonContent( + value, + registry, + MAX_PAUSE_DISPLAY_BYTES + ) + return projection.safe ? projection.value : null + } + const projectedFields = inputFields ? project(inputFields) : null + return { + displayScrubbed: true, + display: project(content) ?? null, + inputFields: Array.isArray(projectedFields) ? (projectedFields as PauseInputField[]) : null, + } +} diff --git a/apps/sim/executor/types.ts b/apps/sim/executor/types.ts index 50e4b108773..6c384a38191 100644 --- a/apps/sim/executor/types.ts +++ b/apps/sim/executor/types.ts @@ -97,6 +97,26 @@ export interface PausePoint { } pauseKind: PauseKind resumeAt?: string + /** + * Set when {@link display} and {@link inputFields} were scrubbed of the run's resolved + * secrets as the pause was captured. Points persisted before scrubbing existed lack it, and + * readers must expose neither field for them: their `response` was stored unscrubbed. + */ + displayScrubbed?: boolean + /** Reviewer-facing pause content without resume links or API-mode status and headers. */ + display?: unknown + /** Fields a human pause asks the resumer to fill, without their default values. */ + inputFields?: PauseInputField[] | null +} + +export interface PauseInputField { + name: string + label: string + type: string + description?: string + placeholder?: string + required: boolean + options?: unknown[] } export interface SerializedSnapshot { diff --git a/apps/sim/lib/api/contracts/v2/openapi/workflows.ts b/apps/sim/lib/api/contracts/v2/openapi/workflows.ts index 0eb21e359f1..929402de51e 100644 --- a/apps/sim/lib/api/contracts/v2/openapi/workflows.ts +++ b/apps/sim/lib/api/contracts/v2/openapi/workflows.ts @@ -1282,7 +1282,7 @@ const declaredRoutes = [ applicationOperation: workflowOperations.execute, operationId: 'executeWorkflowV2', summary: 'Execute workflow', - description: `Run the active deployment, or set \`run.source: "manual"\` to run the saved draft. Deployed execution requires workspace read access and OAuth \`api:write\` scope or an API key; public deployments also allow anonymous synchronous or streamed runs. Manual runs require personal-key or OAuth write access and cannot be async.\n\nUse \`async: true\` for a queue receipt, or \`stream: true\` for SSE. For a synchronous run with both flags false, send \`Accept: application/x-ndjson\` for heartbeat frames and a final frame containing \`data\`. With \`waitTimeoutSeconds\`, a synchronous run still going when the wait elapses continues in the background and answers with its run receipt instead: \`202\` on a JSON response, or a closing \`detached\` frame on an NDJSON stream, which keeps its \`200\`. Timeouts return \`200\` with \`status: "failed"\` and \`error.code: "TIMEOUT"\`; inspect run status. See [Streaming](/workflows/deployment/api#streaming) for event formats. ${EXECUTE_OPTION_CONSTRAINTS}\n\nSupply \`X-Run-Id\` to prevent duplicate execution. Reuse returns \`409\`, without replaying results. A \`503\` with \`error.details.code: "ASYNC_ENQUEUE_AMBIGUOUS"\` omits \`Retry-After\` because the run may already have started. Reconcile using the returned run ID; do not start another run with a new or omitted ID.`, + description: `Run the active deployment, set \`run: { source: "deployment", version }\` to run an earlier deployment version, or set \`run.source: "manual"\` to run the saved draft. Deployed execution requires workspace read access and OAuth \`api:write\` scope or an API key; public deployments also allow anonymous synchronous or streamed runs. Running a chosen version requires workspace write access and an API key or OAuth access token; an unknown version returns \`404\`. Manual runs require personal-key or OAuth write access and cannot be async.\n\nUse \`async: true\` for a queue receipt, or \`stream: true\` for SSE. For a synchronous run with both flags false, send \`Accept: application/x-ndjson\` for heartbeat frames and a final frame containing \`data\`. With \`waitTimeoutSeconds\`, a synchronous run still going when the wait elapses continues in the background and answers with its run receipt instead: \`202\` on a JSON response, or a closing \`detached\` frame on an NDJSON stream, which keeps its \`200\`. Timeouts return \`200\` with \`status: "failed"\` and \`error.code: "TIMEOUT"\`; inspect run status. See [Streaming](/workflows/deployment/api#streaming) for event formats. ${EXECUTE_OPTION_CONSTRAINTS}\n\nSupply \`X-Run-Id\` to prevent duplicate execution. Reuse returns \`409\`, without replaying results. A \`503\` with \`error.details.code: "ASYNC_ENQUEUE_AMBIGUOUS"\` omits \`Retry-After\` because the run may already have started. Reconcile using the returned run ID; do not start another run with a new or omitted ID.`, errors: [ 'BadRequest', 'Unauthorized', @@ -1366,7 +1366,7 @@ const declaredRoutes = [ applicationOperation: workflowOperations.readRun, operationId: 'getWorkflowRunV2', summary: 'Get workflow run', - description: `Read current run state with workspace read access. Request final output and files with \`includeOutput\`, or block outputs with \`selectedOutputs\`. Organization permission groups can hide output, files, or cost.\n\n\`includeFileBase64\` inlines file bytes. A file or the total above 16 MiB returns \`413\` naming a download path; download large files individually. ${HEAD_MIRRORS_GET}`, + description: `Read current run state with workspace read access. Request final output and files with \`includeOutput\`, or block outputs with \`selectedOutputs\`. Organization permission groups can hide output, files, or cost. A paused run lists every pause point in \`paused.points\`, with the fields a human pause asks for and its content with resolved secrets redacted; resume one with Resume workflow run.\n\n\`includeFileBase64\` inlines file bytes. A file or the total above 16 MiB returns \`413\` naming a download path; download large files individually. ${HEAD_MIRRORS_GET}`, errors: [...RESOURCE_CONFLICT_ERRORS, 'PayloadTooLarge'], success: jsonSuccess('The workflow run status.'), }), @@ -1388,6 +1388,7 @@ const declaredRoutes = [ startedAt: '2026-08-09T18:04:10.000Z', endedAt: '2026-08-09T18:04:11.000Z', durationMs: 1_000, + version: 3, paused: null, cost: { total: 12 }, error: null, @@ -1435,7 +1436,7 @@ const declaredRoutes = [ operationId: 'resumeWorkflowRunV2', summary: 'Resume workflow run', description: - 'Resume one human-in-the-loop pause with workspace write access. The resumed attempt receives a new run ID and returns a synchronous result or queue receipt. A run that originally streamed resumes asynchronously through v2; this endpoint returns no SSE stream.', + "Resume one human-in-the-loop pause with workspace write access, naming a `contextId` from the run's `paused.points`. The resumed attempt receives a new run ID and returns a synchronous result or queue receipt. A run that originally streamed resumes asynchronously through v2; this endpoint returns no SSE stream.", errors: [...RESOURCE_ERRORS, 'UsageLimitExceeded', 'Conflict', 'PayloadTooLarge'], success: { byStatus: { diff --git a/apps/sim/lib/api/contracts/v2/workflows.ts b/apps/sim/lib/api/contracts/v2/workflows.ts index d608b59c7a6..8f51805f36f 100644 --- a/apps/sim/lib/api/contracts/v2/workflows.ts +++ b/apps/sim/lib/api/contracts/v2/workflows.ts @@ -1504,7 +1504,12 @@ export const EXECUTE_OPTION_CONSTRAINTS = export const v2WorkflowRunSelectionSchema = z.discriminatedUnion('source', [ z .object({ - source: z.literal('deployment').describe('Execute the active deployed workflow state.'), + source: z.literal('deployment').describe('Execute a deployed workflow version.'), + version: versionNumberSchema + .optional() + .describe( + 'Deployment version to execute instead of the active one. Requires an API key or OAuth access token with workspace write access, and the workflow must be deployed. An unknown version returns `404`.' + ), }) .strict(), z @@ -1569,9 +1574,10 @@ export type V2WorkflowRunSelection = z.input +const v2PauseInputFieldSchema = z + .object({ + name: z.string().describe('Key to supply this field under in the resume `input`.'), + label: z.string().describe('Display label for the field.'), + type: z.string().describe('Declared value type, such as `string`, `number`, or `boolean`.'), + description: z.string().optional().describe('Field help text, when configured.'), + placeholder: z.string().optional().describe('Placeholder text, when configured.'), + required: z.boolean().describe('Whether the field is marked required.'), + options: z + .array(z.unknown().describe('One selectable option.')) + .optional() + .describe('Selectable options, when the field offers a fixed set.'), + }) + .meta({ + id: 'WorkflowRunPauseInputField', + title: 'Pause input field', + description: 'One field a human pause asks the resumer to fill.', + }) + +const v2WorkflowRunPausePointSchema = z + .object({ + contextId: z + .string() + .describe('Pause context to name in `contextId` when resuming this point.'), + blockId: z.string().describe('Workflow block that paused.'), + blockName: z + .string() + .nullable() + .describe( + 'Name of the block in the deployment version the run executed, or null for a manual run or when the version no longer exists.' + ), + kind: z + .enum(['human', 'time']) + .describe('`human` waits for a resume request; `time` resumes automatically at `resumeAt`.'), + status: z + .enum(['paused', 'queued', 'resuming', 'resumed', 'failed']) + .describe( + 'Resume state of this point: waiting, queued behind another resume, resuming, resumed, or a failed resume attempt.' + ), + registeredAt: z + .string() + .nullable() + .describe( + 'ISO 8601 timestamp when the run paused at this point, or null for a point recorded without one.' + ) + .meta({ format: 'date-time' }), + resumeAt: z + .string() + .nullable() + .describe('ISO 8601 automatic-resume timestamp for a time pause, otherwise null.') + .meta({ format: 'date-time' }), + parallel: z + .object({ + parallelId: z.string().describe('Parallel block containing the pause.'), + branchIndex: z.number().int().describe('Zero-based branch that paused.'), + branchTotal: z + .number() + .int() + .nullable() + .describe('Number of branches, or null when unrecorded.'), + }) + .nullable() + .describe('Parallel branch of the pause, or null outside a parallel.'), + loop: z + .object({ + loopId: z.string().describe('Loop block containing the pause.'), + iteration: z.number().int().describe('Zero-based iteration that paused.'), + }) + .nullable() + .describe('Loop iteration of the pause, or null outside a loop.'), + inputFields: z + .array(v2PauseInputFieldSchema) + .nullable() + .describe( + 'Fields a human pause asks for, without default values. Labels and descriptions are resolved, so they are withheld when organization permission groups hide execution data. Null then, for a time pause, a pause with no declared form, fields over 64 KiB or not provably free of secrets, or one recorded before pause details were captured.' + ), + display: z + .unknown() + .describe('Reviewer content value.') + .nullable() + .describe( + "Content the pause presents to its reviewer, with the run's resolved secrets redacted. Null unless `includeOutput=true`, when hidden by organization permission groups, for a time pause, for content over 64 KiB or not provably free of secrets, or for a pause recorded before pause details were captured." + ), + queuePosition: z + .number() + .int() + .nullable() + .describe('Position among pending resume requests for this run, or null when none waits.'), + }) + .meta({ + id: 'WorkflowRunPausePoint', + title: 'Workflow run pause point', + description: 'One point at which a paused run waits.', + }) + /** * The polled run resource. `queued` is backfilled from the async job * queue before the worker writes the durable log row — v1's jobs endpoint 404 @@ -2059,8 +2160,22 @@ export const v2WorkflowRunStatusSchema = z .number() .nullable() .describe('Run duration in milliseconds, or null while active.'), + version: z + .number() + .int() + .nullable() + .describe( + 'Deployment version the run executed. Null for a manual run, a queued run not yet started, or a run whose version no longer exists.' + ), paused: workflowExecutionPausedDetailSchema .omit({ pausedExecutionId: true }) + .extend({ + points: z + .array(v2WorkflowRunPausePointSchema) + .describe( + 'Every point the run is paused at, earliest first, at most 100. Resume one with Resume workflow run, naming its `contextId` and supplying `input`.' + ), + }) .nullable() .describe('Current pause details, or null when the run is not paused.'), cost: z diff --git a/apps/sim/lib/api/contracts/workflows.ts b/apps/sim/lib/api/contracts/workflows.ts index e9aa60fca3c..b5eabc856fa 100644 --- a/apps/sim/lib/api/contracts/workflows.ts +++ b/apps/sim/lib/api/contracts/workflows.ts @@ -612,6 +612,11 @@ const workflowExecutionStatusResponseSchema = z.object({ startedAt: z.string(), endedAt: z.string().nullable(), totalDurationMs: z.number().nullable(), + /** + * Deployment version number the run executed; null for draft or not-yet-recorded runs. + * Optional so a client deployed ahead of the server still parses its response. + */ + deploymentVersion: z.number().int().nullable().optional(), paused: workflowExecutionPausedDetailSchema.nullable(), cost: z.object({ total: z.number() }).nullable(), error: z.string().nullable(), diff --git a/apps/sim/lib/api/mcp/generated/v2-operations.ts b/apps/sim/lib/api/mcp/generated/v2-operations.ts index 11edb9d0bcd..0b7a66f8915 100644 --- a/apps/sim/lib/api/mcp/generated/v2-operations.ts +++ b/apps/sim/lib/api/mcp/generated/v2-operations.ts @@ -1252,7 +1252,7 @@ export const V2_MCP_OPERATIONS = { contract: v2ExecuteWorkflowContract, summary: 'Execute workflow', description: - 'Run the active deployment, or set `run.source: "manual"` to run the saved draft. Deployed execution requires workspace read access and OAuth `api:write` scope or an API key; public deployments also allow anonymous synchronous or streamed runs. Manual runs require personal-key or OAuth write access and cannot be async.\n\nUse `async: true` for a queue receipt, or `stream: true` for SSE. For a synchronous run with both flags false, send `Accept: application/x-ndjson` for heartbeat frames and a final frame containing `data`. With `waitTimeoutSeconds`, a synchronous run still going when the wait elapses continues in the background and answers with its run receipt instead: `202` on a JSON response, or a closing `detached` frame on an NDJSON stream, which keeps its `200`. Timeouts return `200` with `status: "failed"` and `error.code: "TIMEOUT"`; inspect run status. See [Streaming](/workflows/deployment/api#streaming) for event formats. Input descriptions specify compatible modes; invalid combinations return `400`.\n\nSupply `X-Run-Id` to prevent duplicate execution. Reuse returns `409`, without replaying results. A `503` with `error.details.code: "ASYNC_ENQUEUE_AMBIGUOUS"` omits `Retry-After` because the run may already have started. Reconcile using the returned run ID; do not start another run with a new or omitted ID.\n\nOAuth scope: `api:write`.', + 'Run the active deployment, set `run: { source: "deployment", version }` to run an earlier deployment version, or set `run.source: "manual"` to run the saved draft. Deployed execution requires workspace read access and OAuth `api:write` scope or an API key; public deployments also allow anonymous synchronous or streamed runs. Running a chosen version requires workspace write access and an API key or OAuth access token; an unknown version returns `404`. Manual runs require personal-key or OAuth write access and cannot be async.\n\nUse `async: true` for a queue receipt, or `stream: true` for SSE. For a synchronous run with both flags false, send `Accept: application/x-ndjson` for heartbeat frames and a final frame containing `data`. With `waitTimeoutSeconds`, a synchronous run still going when the wait elapses continues in the background and answers with its run receipt instead: `202` on a JSON response, or a closing `detached` frame on an NDJSON stream, which keeps its `200`. Timeouts return `200` with `status: "failed"` and `error.code: "TIMEOUT"`; inspect run status. See [Streaming](/workflows/deployment/api#streaming) for event formats. Input descriptions specify compatible modes; invalid combinations return `400`.\n\nSupply `X-Run-Id` to prevent duplicate execution. Reuse returns `409`, without replaying results. A `503` with `error.details.code: "ASYNC_ENQUEUE_AMBIGUOUS"` omits `Retry-After` because the run may already have started. Reconcile using the returned run ID; do not start another run with a new or omitted ID.\n\nOAuth scope: `api:write`.', handler: () => import('@/app/api/v2/workflows/[workflowId]/execute/route').then((route) => route.POST), }, @@ -1653,7 +1653,7 @@ export const V2_MCP_OPERATIONS = { contract: v2GetWorkflowRunContract, summary: 'Get workflow run', description: - 'Read current run state with workspace read access. Request final output and files with `includeOutput`, or block outputs with `selectedOutputs`. Organization permission groups can hide output, files, or cost.\n\n`includeFileBase64` inlines file bytes. A file or the total above 16 MiB returns `413` naming a download path; download large files individually. `HEAD` checks access with the same authorization as `GET` but skips side effects, returning an empty `200` without payload headers on success.\n\nOAuth scope: `api:read`.', + 'Read current run state with workspace read access. Request final output and files with `includeOutput`, or block outputs with `selectedOutputs`. Organization permission groups can hide output, files, or cost. A paused run lists every pause point in `paused.points`, with the fields a human pause asks for and its content with resolved secrets redacted; resume one with Resume workflow run.\n\n`includeFileBase64` inlines file bytes. A file or the total above 16 MiB returns `413` naming a download path; download large files individually. `HEAD` checks access with the same authorization as `GET` but skips side effects, returning an empty `200` without payload headers on success.\n\nOAuth scope: `api:read`.', handler: () => import('@/app/api/v2/workflows/[workflowId]/runs/[runId]/route').then((route) => route.GET), }, @@ -2576,7 +2576,7 @@ export const V2_MCP_OPERATIONS = { contract: v2ResumeWorkflowContract, summary: 'Resume workflow run', description: - 'Resume one human-in-the-loop pause with workspace write access. The resumed attempt receives a new run ID and returns a synchronous result or queue receipt. A run that originally streamed resumes asynchronously through v2; this endpoint returns no SSE stream.\n\nOAuth scope: `api:write`.', + "Resume one human-in-the-loop pause with workspace write access, naming a `contextId` from the run's `paused.points`. The resumed attempt receives a new run ID and returns a synchronous result or queue receipt. A run that originally streamed resumes asynchronously through v2; this endpoint returns no SSE stream.\n\nOAuth scope: `api:write`.", handler: () => import('@/app/api/v2/workflows/[workflowId]/runs/[runId]/resume/route').then( (route) => route.POST diff --git a/apps/sim/lib/mothership/__integration__/content-delivery-compat.baseline.json b/apps/sim/lib/mothership/__integration__/content-delivery-compat.baseline.json index 2f85e7c1452..a9dcd23f9db 100644 --- a/apps/sim/lib/mothership/__integration__/content-delivery-compat.baseline.json +++ b/apps/sim/lib/mothership/__integration__/content-delivery-compat.baseline.json @@ -853,6 +853,7 @@ "startedAt": "