Workflow Engine (Backend)

Support-facing summary of the workflow execution engine as it exists in current backend code.

• Stored workflow types are `agent`, `structured`, and `employee`.
• Organization-owned and teammate-assigned workflows are `agent` or `structured`.
• `employee` workflows are the default workflows attached to an AI teammate and are used by internal chat, reminders, and teammate delegation.
• Default employee workflows are system-managed. Normal workflow update, delete, bulk delete, and restore paths reject them with `403` unless the backend caller uses the internal employee-workflow mutation override.

• A workflow must have a published `flow` map and `starting_step_id`.
• Published runtime metadata also includes:
• `payload_format`
• `state_definition`
• `mode`
• `debounce_seconds`
• `debounce_key`
• `mode` is `parallel` or `restart`.

• Agent workflows:
• `starting_step`
• `agent`
• Structured workflows:
• `starting_step`
• `llm`
• `tool`
• `run_workflow`
• `run_employee`
• `run_code`
• `condition`
• `user_input`
• `send_message`
• `finish`
• Backend type definitions and validators still include `mapping` and `wait_callback`, but `WorkflowProcessor` currently throws `Mapping processing is not implemented` and `Wait callback processing is not implemented`. Treat those as non-runnable.

• Public trigger:
• `POST /api/public/trigger/{public_api_key}/{workflow_id}`
• Private trigger:
• `POST /api/private/trigger/{workflow_id}`
• Manual org-scoped run:
• `POST /api/organizations/{orgId}/workflows/{workflow_id}/runs/`

Public trigger

• Resolves the organization by `public_api_key`.
• Requires the workflow to exist and have published `flow` plus `starting_step_id`.
• Uses the parsed request body as payload.
• Does not validate that payload against the workflow's declared `payload_format`.
• The URL key identifies the organization. It is not proof that the request came from an expected third-party provider.
• The route does not provide generic webhook-signature, timestamp, nonce, or replay verification.
• If the workflow has a `debounce_key`, the route extracts it from the payload using dot-path lookup with flat-key fallback.
• If both `debounce_seconds` and an extracted debounce key are present:
• the route upserts a row in `workflow_debouncer`
• the response returns `{ success: true, data: { debounced: true } }`
• no run is created immediately
• `workflowDebouncerDaemon` later creates the run and enqueues it with `isAsync: true`
• If there is a debounce key and `mode === restart`, but the request is not going through the debouncer path:
• backend looks for an active run on the same workflow and raw debounce key
• matching active run is marked `aborted`
• a new run is created immediately
• Immediate public-trigger runs store `debounce_key` on the run. Debounced runs store the composite key written by the debouncer row.
• Immediate runs use `isAsync: false`, so the HTTP response waits for workflow processing. Debounced requests return after recording the event for later execution.
• Restart marks the matching active run as aborted. It cannot undo completed side effects, and an external call already in progress may still finish.
• Restart and debounce handling is specific to this route. Other workflow entry points do not inherit it automatically.

Private trigger

• Uses `req.organization` context instead of a public API key.
• Requires published `flow` plus `starting_step_id`.
• Creates and enqueues the run immediately.
• Uses `isAsync: false`, so the HTTP response waits for workflow processing.

Manual org-scoped run

• Requires an authenticated org user.
• Validates `payload` against `payload_format`.
• Validates optional `state` against `state_definition`.
• Stores the optional manual state under `runtimeContext.state` before queueing the run.
• Creates the run with no conversation attached and queues execution from the published `starting_step_id`.

• Every execution is persisted as a `WorkflowRun` snapshot.
• A run stores:
• `flow`
• `state`
• `parent_workflow`
• `conversation_id`
• `model`
• `input_tokens`
• `output_tokens`
• `channel`
• `ai_employee_id`
• `trigger_id`
• `trigger_url`
• `debounce_key`
• `workflowRunDaemon` consumes Redis queue messages.
• `WorkflowRunQueue.processMessage()` enriches `runtimeContext` with:
• `isDebug`
• `isAsync`
• `organizationId`
• `employeeId`
• `workflowId`
• The queue layer also copies conversation channel and workflow employee onto top-level run columns when available.
• Before processing a queued step, `WorkflowRunQueue` re-checks the run:
• a run that already reached a terminal status is skipped and logged without writing run state
• the run is aborted instead of processed when the webchat conversation is archived or deleted, when the conversation's automation is disabled because the conversation is assigned to an operator, or when the organization is inactive
• an abort sets status `aborted` and writes the reason into run state `error` as `Workflow run aborted before step {runStepId}: {reason}`, so the run history and logs explain the terminal status instead of showing a silent stop
• `WorkflowProcessor` atomically claims the run before executing it, so terminal runs are not reprocessed. An in-progress run refreshes its heartbeat every 15 seconds without changing `updated_at`. The workflow-run daemon checks for stalled runs every 60 seconds; when an in-progress run's last heartbeat is more than 120 seconds old, the run is marked `failed` with `state.error` set to `WORKFLOW_RUN_INTERRUPTED`. Runs without a recorded heartbeat are not swept. For conversation-backed runs, an unfinished AI reply is failed, or a failed reply is created if none exists, with `Processing was interrupted. Please send your message again.` An already completed AI message is left unchanged.

• `created`
• `in_progress`
• `suspended`
• `completed`
• `failed`
• `aborted`

• `agent`: runs Alloy AI with tools/streaming behavior. The agent step appends workspace-scoped tools based on the AI teammate's capabilities: the `storage.*` filesystem tools when the teammate has shared-folder grants (system employees always), the `mastra_workspace.sandbox_*` tools when sandbox mode is enabled, and conversation-scoped tools when the workflow run has a conversation. It validates the primary model configuration before building the agent, so a missing provider key is reported directly; configured fallback models without usable credentials are skipped with a warning rather than preventing the primary model from running.
• Each internal agent step is persisted as its own ordered row while running and when completed. Rows are keyed by fallback attempt and step number, so when a fallback model replays the agent from step zero, the failed attempt remains available instead of being overwritten. Run detail APIs and workflow-run tools hydrate those rows back into the existing flat `additional_data.steps[]` response shape in oldest-attempt, then step-number order; historical logs that already contain the array remain readable. Persisted result rows omit duplicated provider requests and accumulated message history; step-log attachment payloads are removed or redacted while metadata is retained. These reductions affect stored logs, not the live AI step input.
• AI-agent calls through `agent.call_teammate` and dynamically generated workflow tools share an operator-configured nesting guard. The limit defaults to `5`; exceeding it rejects the tool call before another child workflow run is started.
• `llm`: single prompt/model step without tool execution.
• `tool`: executes one backend tool after parameter substitution. Current node-tool options include `http.http_request` and `stargate.execute_task`.
• `run_workflow`: creates a child workflow run, waits for completion, and resumes the parent with the child result.
• `run_employee`: displayed in the builder as `Assign AI`; executes an AI Teammate workflow with an assignment, supports `attachments_json`, and can shape plain-text child output into a declared AI response schema.
• When schema shaping is needed, the formatting model is tried up to three times until it returns an object.
• If every attempt returns no object, the step preserves the child text as `{ message: textOutput }`; an empty text response becomes `{ message: "Empty response" }`.
• `run_code`: executes Python or JavaScript code in an isolated sandbox. When both `MICRO_SANDBOX_API_URL` and `MICRO_SANDBOX_API_KEY` are configured, the server uses the self-hosted microsandbox panel; otherwise it uses E2B. A partially configured panel falls back to E2B. When E2B is selected and no explicit template was supplied, `E2B_RUNCODE_DEFAULT_TEMPLATE` can select the operator's default code-execution template; otherwise E2B uses its standard default image.
• An optional `sandboxTemplate` value creates the sandbox from an approved template ID of the selected provider. The operator-configured allowlist is `MICRO_SANDBOX_RUN_CODE_TEMPLATE_IDS` for the panel or `E2B_RUN_CODE_TEMPLATE_IDS` for E2B; an empty list permits only the selected provider's default template.
• A blank explicit template value is rejected, and an unapproved template ID fails validation and step execution.
• Step data can declare storage transfers: `importFromStorage` entries copy files or folders from organization storage into the sandbox before the code runs, and `exportToStorage` entries copy sandbox files or folders back after the sandbox execution returns. An error reported inside the sandbox execution result does not itself stop the export; a thrown sandbox or transfer failure does. A path may name a file or a folder; folders are copied recursively with their structure, and transfer paths support the same `{{...}}` placeholder syntax as the code.
• Storage transfers check readable source paths for imports and writable destination paths for exports before the sandbox starts. When Storage Access Group enforcement is enabled, the grants reflect the initiating user and, when applicable, the AI teammate's shared folders; organization-credential runs use organization-wide access. Directory transfers expand into per-file entries with a cap of `100` expanded files per call and a sandbox listing depth of `10`. Each transferred file is capped at `100 MB` and the files in one call at `300 MB` in total.
• The run-code limit is `180` seconds (an execution timeout on the microsandbox panel; an E2B request timeout on E2B).
• When the sandbox returns an execution result, `saveToState` reads its code-interpreter fields via `{{output.results[0].json.fieldName}}`, `{{output.results[0].text}}`, `{{output.logs.stdout}}`, or `{{output.error.value}}`. The next step receives that same execution result without a `result` wrapper, via the corresponding `{{input.*}}` paths. An exception inside the code can populate `error` with `name`, `value`, and `traceback` while the workflow continues.
• Sandbox creation, import/export, or other thrown failures instead produce a step-log output with `success: false`, an error message, and `result: null`; the next step receives that null result. `saveToState` runs only on the successful processor path. The persisted step-log output also carries `success`, `message`, and `language`, with the code-interpreter result under `result` on the successful path; those wrapper fields are not next-step input.
• Secret organization-variable placeholders are resolved only for sandbox execution. Persisted step input keeps the placeholders, and returned output, tracebacks, and errors are scrubbed before they enter step logs or run state.
• `condition`: evaluates branches and continues to the matching next step or default branch.
• `user_input`: sends its prompt only when the run has a conversation, then suspends and waits for resume payload. Without a conversation, it suspends without sending the prompt. There is no public API route to resume it.
• `send_message`: sends a message only when the run has a conversation, then continues to the next step. Without a conversation, it continues with step output but sends nothing.
• `finish`: finalizes the run.

• `payload_format` currently validates `string`, `number`, `boolean`, `date`, and `datetime` field types.
• `date` expects a calendar-date string in `YYYY-MM-DD` format.
• `datetime` expects an ISO 8601 datetime string.
• `state_definition` currently validates `string`, `number`, and `boolean` field types.
• Declared runtime state is exposed to templates through `{{state.key}}`.
• `saveToState` writes values into `runtimeContext.state`.
• Tool steps leave secret organization-variable placeholders unresolved for the tool to process. Tools that resolve them for an outbound request must return and persist a redacted representation rather than the plaintext value.
• `{{input.*}}` resolves only from direct predecessor output.
• If a node has multiple direct predecessors, the referenced input path must exist in every direct predecessor schema.
• Starting-step payload keys can use dotted field names.
• Current canonical payload-format storage is an object keyed by field name where each field stores `type` plus optional `description`. Legacy read shapes with direct type strings are still accepted for backward compatibility.

• Org-wide runs:
• `GET /api/organizations/{orgId}/workflows/runs/`
• Workflow-scoped runs:
• `GET /api/organizations/{orgId}/workflows/{workflow_id}/runs/`
• Run detail:
• `GET /api/organizations/{orgId}/workflows/{workflow_id}/runs/{run_id}/`
• Exact-match filters supported today:
• `status`
• `statuses`
• `run_id`
• `workflow_id`
• `conversation_id`
• `channel`
• `ai_employee_id`
• `initiator_user_id`
• `contact_id`
• `model`
• `trigger_id`
• Runtime-state filters supported today:
• `state[{key}]={value}`
• Range filters supported today:
• `started`
• `ended`
• `duration`
• `tokens`
• `cost`
• Express query parsing is `extended`, so range filters use bracket syntax such as `duration[gte]=5`.
• Runtime-state filters are also parsed with bracket syntax, for example `state[customerId]=123`.
• Multiple `state[...]` filters are ANDed together and compare against `state.runtimeContext.state` after string coercion.

• `status` and `statuses` are mutually exclusive in query validation.
• `initiator_user_id` is matched from `state.runtimeContext.userId`.
• `contact_id` is resolved to the contact's active channel ids first, then matched against both direct conversation links and contact-channel participant links.
• `started` maps to `created_at`.
• `ended` maps to `completed_at` only. Failed or suspended timestamps are not part of the `ended` filter.
• `duration` is computed as `completed_at - created_at`, so non-completed runs do not match duration filters meaningfully.
• `tokens` is `COALESCE(input_tokens, 0) + COALESCE(output_tokens, 0)`.
• `cost` is the summed `organization_token_history.cost_usd` for `type = 'decrease'` rows linked to the workflow run.

• Query validation allows `exclude=flow`, `exclude=state`, or `exclude=flow,state`.
• In `GetWorkflowRunsAction`, any truthy `exclude` value switches the serializer to `listData()`.
• In that mode the DB query excludes only the `flow` column.
• `state` is still returned, but in compressed form:
• `currentStepId`
• top-level `error`
• `runtimeContext.userId`
• `runtimeContext.isAsync`
• trimmed step summaries with reduced input/output payloads
• Top-level list fields still include:
• `conversation_channel`
• `conversation_title`
• `model`
• `input_tokens`
• `output_tokens`
• `channel`
• `ai_employee_id`
• `trigger_id`
• `trigger_url`
• `debounce_key`
• `costs`
• Each `costs[]` entry is aggregated per token `key_type` and currently includes:
• `key_type`
• `input_tokens`
• `output_tokens`
• `cached_input_tokens`
• `usd_cost`