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`.
• Regular org workflows and skills 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.

• 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 raw request body as payload.
• 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.

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.

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.
• `WorkflowProcessor` atomically claims the run before executing it, so terminal runs are not reprocessed.

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

• `agent`: runs Alloy AI with tools/streaming behavior.
• `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.
• `run_code`: executes Python or JavaScript code.
• `condition`: evaluates branches and continues to the matching next step or default branch.
• `user_input`: suspends the run and waits for resume payload.
• `send_message`: sends a message and continues to the next step.
• `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`.
• `{{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`
• `run_id`
• `workflow_id`
• `conversation_id`
• `channel`
• `ai_employee_id`
• `model`
• `trigger_id`
• Runtime-state filters supported today:
• `state[{key}]={value}`
• Range filters supported today:
• `started`
• `ended`
• `duration`
• `tokens`
• 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.

• `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)`.

• 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`