Workflow Engine

The Workflow Engine is the core runtime that executes BPMN process definitions. It manages execution lifecycle, module dispatching, state transitions, and audit trails.

Architecture Overview

Key Components

• WorkflowService — Core execution loop. Resolves BPMN tasks, dispatches to modules, manages state transitions.
• ProjectRepository — Loads and caches workflow data (process definitions, module processors, environment config).
• IWorkflowStorage — Persistence layer for waiting executions. Size-tiered: light sessions (< EngineOptions.HeavySessionThresholdBytes, default 1 MB) land in Redis (10-min sliding TTL); heavy sessions skip Redis and live only in the Azure audit blob, which is uploaded synchronously for them so the next inbound can resume immediately. Resume reads Redis → Azure execution.msgpack → SessionNotFound (no bare-audit fallback).
• IAuditService — Two parallel sinks per segment-end. Blob side: serializes the full Execution to MessagePack via AuditBlobSerializer (same format Redis uses) — async via BlobUploadQueue for the common case, sync direct upload for heavy waiting saves. Bus side: enqueues a slim ExecutionAudit snapshot (identity + lifecycle + delta logs, no dynamic payloads) for the external DB consumer.
• BlobUploadWorker — Background drainer for the blob queue. Batches up to 200 items, dedups by blob name (so repeated overwrites of execution.msgpack collapse), uploads with bounded parallelism (20).
• ExecutionAuditQueueWorker — Background drainer for the audit queue. Serializes each slim snapshot to JSON, ships to the execution-audit Service Bus queue.
• External audit consumer — Lives in a separate repo. Subscribes to the bus queue, deserializes the JSON snapshot, UPDATEs the ExecutionAudits MySQL row and bulk-inserts new logs. Never touches blob storage (the engine already wrote it).

Execution Model

An Execution represents a single run of a BPMN process. Each iteration of the execution loop produces an ExecutionStep snapshot.

Execution Lifecycle

1. Start — A new Execution is created via WorkflowService.Start() with TokenState.Running.
2. Loop — The engine iterates through BPMN tasks, resolving the next task based on outgoing connections and task output.
3. Module Dispatch — For each task, the engine resolves a module processor and dispatches execution to the appropriate module via HTTP transport.
4. Completion — When an EndEvent is reached, the state becomes Stopped | Completed.
5. Waiting — If a module sets WaitingTaskCallback, the execution is saved to Redis and paused until a callback resumes it.
6. Error — Exceptions set Exception | Stopped and halt execution.

TokenState (Bitwise Flags)

The execution state is tracked using a single sbyte field (State) with bitwise flags. This allows multiple states to be active simultaneously.

[Flags]
public enum TokenState : sbyte
{
    None                = 0b_0000_0000,  // 0   — Default/uninitialized
    Running             = 0b_0000_0001,  // 1   — Execution is actively running
    WaitingTaskCallback = 0b_0000_0010,  // 2   — Paused, waiting for external task callback
    WaitingTask         = 0b_0000_0100,  // 4   — Dispatched to task/module, awaiting response
    Stopped             = 0b_0000_1000,  // 8   — Execution has ended (normal or error)
    Exception           = 0b_0001_0000,  // 16  — An error occurred during execution
    Completed           = 0b_0010_0000,  // 32  — Execution finished (EndEvent reached)
    TaskCompleted       = 0b_0100_0000,  // 64  — Current step's task executed successfully
}

State Combinations

State	Bit Value	Meaning
Running	1	Actively executing tasks in the loop
Running | WaitingTask	5	Task dispatched to module, awaiting response
Running | WaitingTaskCallback	3	Paused, saved to Redis, waiting for external callback
Running | TaskCompleted	65	Current step's task completed successfully
Stopped | Completed	40	Normal completion — reached an EndEvent
Stopped | Exception	24	Error termination — an exception stopped execution
Running | Exception	17	Error detected but execution continued

Status Derivation (UI)

The GetPrimaryStatus() extension method derives a human-readable status string from TokenState:

Priority	Condition	Display Status
1	Exception flag set	Error
2	Stopped + Completed	Completed
3	Stopped only	Stopped
4	WaitingTaskCallback or WaitingTask	Waiting
5	Running	Running
6	None of the above	Unknown

Execution Loop Detail

The core loop in WorkflowService.Run():

for each iteration:
  1. Clear TaskCompleted flag (each step starts fresh)
  2. Determine next BPMN task:
     - If no TaskId: find StartEvent
     - If has TaskId: resolve next task via outgoing connections + TaskOutput
  3. Increment Sequence, assign new ExecutionStepId
  4. If task has no type: set Stopped | Exception (TasksHasNoType)
  5. Resolve module processor for task type + channel
  6. Dispatch to module (sets WaitingTask, then clears on response)
  7. On module success: set TaskCompleted flag
  8. On EndEvent: set Stopped | Completed
  9. On loop limit (200): set Stopped | Exception
  10. Enqueue audit snapshot

Loop continues while: Running AND NOT Stopped AND NOT WaitingTaskCallback

Callback Flow

When a module needs an external callback (e.g., user input, webhook):

1. Module sets WaitingTaskCallback on the execution.
2. Engine exits the loop. EnqueueExecutionCompletedAsync runs first — it serializes the Execution once via MessagePack, sets PayloadSizeBytes, and either enqueues the blob (light/terminal) or sync-uploads it (heavy waiting).
3. SaveWaitingExecutionAsync reads PayloadSizeBytes and picks the tier:
   • Light (< EngineOptions.HeavySessionThresholdBytes, default 1 MB): write the full Execution to Redis (session:{id}, 10-min sliding TTL).
   • Heavy (>= 1 MB): skip Redis. The blob upload from step 2 is the durable copy.
   • Reverse index (channel_session:{cs_id} -> executionId) goes to Redis either way — it's ~50 bytes and lets channel-session inbounds resolve to the right execution without a DB hit.
4. External system calls back via API (by execution ID or channel-session ID).
5. Engine loads execution: Redis hit (light, hot) → return; miss → DB lookup for WorkspaceId → Azure execution.msgpack → normalize payloads to ExpandoObject for module callbacks. If neither has it: SessionNotFound (no bare-audit fallback — silently resuming against an empty Execution is what the original "context erased after chat reply" bug looked like).
6. ExecuteCallback runs, clears WaitingTaskCallback, loop resumes.

Split into two diagrams — pause (first call) and resume (callback). They share the same participants but each fits cleanly on screen.

Phase 1 — Execute hits async task and pauses

Phase 2 — User responds, engine resumes

Module Resolution

Modules are resolved using a two-level key: (taskType, channel).

1. Specific match — Try (taskType, channelName) first
2. Generic fallback — Try (taskType, null) if no channel-specific handler exists

Each module processor defines which BPMN task types it handles and optionally which channels.

Audit Pipeline

Every loop iteration produces an audit event:

1. Per-step blob — EnqueueStepBlob (called once per loop iteration) snapshots the full Execution to MessagePack and enqueues {stepId}.msgpack for the BlobUploadWorker. Path: {workspaceId}/{executionId}/{stepId}.msgpack. Immutable (stepId is per-step).
2. Segment-end blob — EnqueueExecutionCompletedAsync (called once per segment-end — terminal OR waiting) serializes the full Execution to MessagePack via AuditBlobSerializer. The same blob path (execution.msgpack) is overwritten each segment. For a heavy waiting save the upload runs synchronously on the engine thread (no queue) so the next inbound can resume without racing the async worker; for everything else (light waiting, terminal) it goes through BlobUploadQueue.
3. Bus message — Same call also enqueues a slim ExecutionAudit (identity columns + lifecycle + delta logs, no dynamic payloads) onto ExecutionAuditQueue. The ExecutionAuditQueueWorker drains it, serializes to JSON via System.Text.Json, sends to the execution-audit Service Bus queue. LastPersistedLogIndex tracks the log watermark so the same entries never ship twice across segments.
4. External consumer — Lives in a separate repo. Subscribes to the bus queue, deserializes the JSON snapshot, UPDATEs the ExecutionAudits MySQL row (or INSERTs on first sight), bulk-inserts the new logs. Never touches blob storage — the engine wrote it directly.

Persistence and Serialization

Two storage tiers with one shared serialization format for the heavy artifact:

Concern	Where	Format	Writer	Reader	Sync/Async
Hot session cache (light only)	Redis session:{id}	MessagePack via AuditBlobSerializer	SaveWaitingExecutionAsync	GetExecutionAsync (Redis hit)	sync; 10-min sliding TTL
Reverse index	Redis channel_session:{cs_id}	UTF-8 GUID string	SaveWaitingExecutionAsync	GetExecutionByChannelSessionAsync	sync; same 10-min TTL
Per-step audit	Azure Blob {ws}/{exec}/{stepId}.msgpack	MessagePack via AuditBlobSerializer	EnqueueStepBlob	Web UI step replay	async via BlobUploadQueue
Segment-end audit + resume durable copy	Azure Blob {ws}/{exec}/execution.msgpack	MessagePack via AuditBlobSerializer	EnqueueExecutionCompletedAsync	Web UI session detail; engine cold-resume	sync if heavy waiting, otherwise async via queue
Lifecycle event	Service Bus execution-audit	JSON via System.Text.Json	ExecutionAuditQueueWorker	External DB consumer (separate repo)	async via ExecutionAuditQueue
Audit row	MySQL ExecutionAudits	DB columns	External consumer (after bus message)	WorkflowStorage for WorkspaceId on cold resume	external

Why MessagePack for the blob (and now for Redis)

The blob carries the full Execution: five dynamic payloads (Request/Response/Context/Module/Execution), ConversationHistories, ExecutionMap, Logs, and metadata. Two reasons it's MessagePack:

• Performance — A measured micro-benchmark (SerializationBenchmarks) shows MessagePack is 2–5× faster end-to-end and allocates 3–14× less garbage than the previous Newtonsoft BSON path, across ContextEntries={1, 10, 50, 200}. Serialize-only is ~7× faster.
• Correctness — The previous Redis path used Newtonsoft BSON. The *RawPayload byte fields it relied on were marked [JsonIgnore] (intended to keep them out of "blob JSON"), but [JsonIgnore] also applies to BSON. The bytes never landed in Redis, then DeserializeFromPersistence read them back as null and clobbered every dynamic payload with an empty ExpandoObject on every round-trip. The unified MessagePack path drops the raw-byte indirection entirely; the dynamic payloads ride along inline, and a single DeserializeFromPersistence() normalize step promotes them back to ExpandoObject so module callbacks' is IDictionary<string, object?> type-checks keep matching.

Why JSON for the bus message

The bus message is slim (~1–5 KB) and the cost is dominated by the Service Bus network round-trip (~20–50 ms). Serialization is less than 1% of that, and JSON gives free human-readability in the Azure portal, Service Bus Explorer, and dead-letter queues. The external consumer also expects application/json today; changing it would require a coordinated cross-repo deploy with negligible perf win.

Save and Resume Behavior (per-segment)

Three focused diagrams. The arrows in (1) Save and (3) Resume are on the engine thread; (2) Drain runs in the background.

(1) Save at end-of-segment — serialize once, choose sync vs async upload

(2) Background drain — blob worker + audit worker + external consumer

(3) Resume on next inbound — Redis hit vs cold blob fallback

Failure modes worth knowing

Scenario	Result	Why
User pauses under 10 min, Redis warm	Hot path: 1 Redis GET, ~1 ms	sliding TTL refreshes on every hit
User pauses over 10 min on a light session	Cold path: 1 DB SELECT + 1 Azure GET, ~50–150 ms	Redis expired; execution.msgpack is the durable copy (queue drains in seconds, so by the time TTL expires it's already there)
Heavy session at any time	Cold path always	Redis is skipped by design — blob is the only copy
Engine crashes between sync blob upload and SaveWaitingExecutionAsync returning	Resumable	The blob is already in Azure; next inbound finds it via DB → Azure
Blob upload fails (sync, heavy)	SaveWaitingExecutionAsync returns an error result	Execution is not resumable; caller logs CRITICAL
Bus consumer is down / lagging	New executions still run, the audit DB row just lags behind	Engine's resume path uses the audit row only for WorkspaceId; for a never-completed execution the row may not exist yet — cold resume returns SessionNotFound until the first segment's __started message has been consumed
Legacy BSON bytes still in Redis after deploy	Treated as a miss → cold path → Azure blob → re-warmed in MessagePack	MessagePackSerializationException is caught; a single 10-min TTL window bridges the deploy

Error Handling

Scenario	TokenState Set	Behavior
Module not found	Exception added	Warning logged, execution continues
Module returns error	Exception | Stopped	Execution halts
Loop limit (200 steps)	Exception | Stopped	Execution forcefully halted
Redis save failure	Log error	Execution returned but not resumable
Start task not found	Exception | Stopped	Execution halts immediately
Process/pool not found	Exception | Stopped	Execution halts immediately
Task has no type	Exception | Stopped	Execution halts — task node has no module type assigned