WORKFL: Workflow Engine

Purpose: DAG-based workflow execution for deployments, approvals, and custom automation.

Modules

Module: `workflow-designer`

Aspect	Specification
Responsibility	Template creation; DAG graph editor; validation
Dependencies	`step-registry`
Data Entities	`WorkflowTemplate`, `StepNode`, `StepEdge`

Workflow Template Structure:

interface WorkflowTemplate {
  id: UUID;
  tenantId: UUID;
  name: string;
  displayName: string;
  description: string;
  version: number;

  // DAG structure
  nodes: StepNode[];
  edges: StepEdge[];

  // I/O
  inputs: InputDefinition[];
  outputs: OutputDefinition[];

  // Metadata
  tags: string[];
  isBuiltin: boolean;
  createdAt: DateTime;
  createdBy: UUID;
}

Module: `workflow-engine`

Aspect	Specification
Responsibility	DAG execution; state machine; pause/resume
Dependencies	`step-executor`, `step-registry`
Data Entities	`WorkflowRun`, `WorkflowState`
Events Produced	`workflow.started`, `workflow.paused`, `workflow.resumed`, `workflow.completed`, `workflow.failed`

Workflow Execution Algorithm:

class WorkflowEngine:
    def execute(self, workflow_run: WorkflowRun) -> None:
        """Main workflow execution loop."""

        # Initialize
        workflow_run.status = "running"
        workflow_run.started_at = now()
        self.save(workflow_run)

        try:
            while not self.is_terminal(workflow_run):
                # Handle pause state
                if workflow_run.status == "paused":
                    self.wait_for_resume(workflow_run)
                    continue

                # Get nodes ready for execution
                ready_nodes = self.get_ready_nodes(workflow_run)

                if not ready_nodes:
                    # Check if we're waiting on approvals
                    if self.has_pending_approvals(workflow_run):
                        workflow_run.status = "paused"
                        self.save(workflow_run)
                        continue

                    # Check if all nodes are complete
                    if self.all_nodes_complete(workflow_run):
                        break

                    # Deadlock detection
                    raise WorkflowDeadlockError(workflow_run.id)

                # Execute ready nodes in parallel
                futures = []
                for node in ready_nodes:
                    future = self.executor.submit(
                        self.execute_node,
                        workflow_run,
                        node
                    )
                    futures.append((node, future))

                # Wait for at least one to complete
                completed = self.wait_any(futures)

                for node, result in completed:
                    step_run = self.get_step_run(workflow_run, node.id)

                    if result.success:
                        step_run.status = "succeeded"
                        step_run.outputs = result.outputs
                        self.propagate_outputs(workflow_run, node, result.outputs)
                    else:
                        step_run.status = "failed"
                        step_run.error_message = result.error

                        # Handle failure action
                        if node.on_failure == "fail":
                            workflow_run.status = "failed"
                            workflow_run.error_message = f"Step {node.name} failed: {result.error}"
                            self.cancel_pending_steps(workflow_run)
                            return
                        elif node.on_failure == "rollback":
                            self.trigger_rollback(workflow_run, node)
                        elif node.on_failure.startswith("goto:"):
                            target = node.on_failure.split(":")[1]
                            self.add_ready_node(workflow_run, target)
                        # "continue" just continues to next nodes

                    step_run.completed_at = now()
                    self.save(step_run)

            # Workflow completed successfully
            workflow_run.status = "succeeded"
            workflow_run.completed_at = now()
            self.save(workflow_run)

        except WorkflowCancelledError:
            workflow_run.status = "cancelled"
            workflow_run.completed_at = now()
            self.save(workflow_run)
        except Exception as e:
            workflow_run.status = "failed"
            workflow_run.error_message = str(e)
            workflow_run.completed_at = now()
            self.save(workflow_run)

Module: `step-executor`

Aspect	Specification
Responsibility	Step dispatch; retry logic; timeout handling
Dependencies	`step-registry`, `plugin-sandbox`
Data Entities	`StepRun`, `StepResult`
Events Produced	`step.started`, `step.progress`, `step.completed`, `step.failed`, `step.retrying`

Step Node Structure:

interface StepNode {
  id: string;                    // Unique within template (e.g., "deploy-api")
  type: string;                  // Step type from registry
  name: string;                  // Display name
  config: Record<string, any>;   // Step-specific configuration
  inputs: InputBinding[];        // Input value bindings
  outputs: OutputBinding[];      // Output declarations
  position: { x: number; y: number };  // UI position

  // Execution settings
  timeout: number;               // Seconds (default from step type)
  retryPolicy: RetryPolicy;
  onFailure: FailureAction;
  condition?: string;            // JS expression for conditional execution

  // Documentation
  description?: string;
  documentation?: string;
}

type FailureAction = "fail" | "continue" | "rollback" | "goto:{nodeId}";

interface InputBinding {
  name: string;                  // Input parameter name
  source: InputSource;
}

type InputSource =
  | { type: "literal"; value: any }
  | { type: "context"; path: string }        // e.g., "release.name"
  | { type: "output"; nodeId: string; outputName: string }
  | { type: "secret"; secretName: string }
  | { type: "expression"; expression: string };  // JS expression

interface StepEdge {
  id: string;
  from: string;           // Source node ID
  to: string;             // Target node ID
  condition?: string;     // Optional condition expression
  label?: string;         // Display label for conditional edges
}

interface RetryPolicy {
  maxRetries: number;
  backoffType: "fixed" | "exponential";
  backoffSeconds: number;
  retryableErrors: string[];
}

Module: `step-registry`

Aspect	Specification
Responsibility	Built-in + plugin-provided step types
Dependencies	`plugin-registry`
Data Entities	`StepType`, `StepSchema`

Built-in Step Types:

Step Type	Category	Description
`approval`	Control	Wait for human approval
`security-gate`	Gate	Evaluate security policy
`custom-gate`	Gate	Custom OPA policy evaluation
`deploy-docker`	Deploy	Deploy single container
`deploy-compose`	Deploy	Deploy Docker Compose stack
`deploy-ecs`	Deploy	Deploy to AWS ECS
`deploy-nomad`	Deploy	Deploy to HashiCorp Nomad
`health-check`	Verify	HTTP/TCP health check
`smoke-test`	Verify	Run smoke test suite
`execute-script`	Custom	Run C#/Bash script
`webhook`	Integration	Call external webhook
`trigger-ci`	Integration	Trigger CI pipeline
`wait-ci`	Integration	Wait for CI pipeline
`notify`	Notification	Send notification
`rollback`	Recovery	Rollback deployment
`traffic-shift`	Progressive	Shift traffic percentage

Step Type Definition:

interface StepType {
  type: string;                  // "deploy-compose"
  displayName: string;           // "Deploy Compose Stack"
  description: string;
  category: StepCategory;
  icon: string;

  // Schema
  configSchema: JSONSchema;      // Step configuration schema
  inputSchema: JSONSchema;       // Required inputs schema
  outputSchema: JSONSchema;      // Produced outputs schema

  // Execution
  executor: "builtin" | UUID;    // builtin or plugin ID
  defaultTimeout: number;
  safeToRetry: boolean;
  retryableErrors: string[];

  // Documentation
  documentation: string;
  examples: StepExample[];
}

Workflow Run State Machine

┌─────────────────────────────────────────────────────────────────────────────┐
│                    WORKFLOW RUN STATE MACHINE                               │
│                                                                             │
│                              ┌──────────┐                                   │
│                              │ CREATED  │                                   │
│                              └────┬─────┘                                   │
│                                   │ start()                                 │
│                                   ▼                                         │
│                    ┌─────────────────────────────┐                          │
│                    │                             │                          │
│         pause() ┌──┴──────────┐                  │                          │
│       ┌────────►│   PAUSED    │◄─────────┐      │                          │
│       │         └──────┬──────┘          │      │                          │
│       │                │ resume()        │      │                          │
│       │                ▼                 │      │                          │
│       │         ┌─────────────┐          │      │                          │
│       └─────────│   RUNNING   │──────────┘      │                          │
│                 └──────┬──────┘ (waiting for    │                          │
│                        │         approval)       │                          │
│           ┌────────────┼────────────┐           │                          │
│           │            │            │           │                          │
│           ▼            ▼            ▼           │                          │
│    ┌───────────┐ ┌───────────┐ ┌───────────┐   │                          │
│    │ SUCCEEDED │ │  FAILED   │ │ CANCELLED │   │                          │
│    └───────────┘ └───────────┘ └───────────┘   │                          │
│                                                                             │
│    Transitions:                                                             │
│    - CREATED → RUNNING: start()                                            │
│    - RUNNING → PAUSED: pause(), waiting approval                           │
│    - PAUSED → RUNNING: resume(), approval granted                          │
│    - RUNNING → SUCCEEDED: all nodes complete                               │
│    - RUNNING → FAILED: node fails with fail action                         │
│    - RUNNING → CANCELLED: cancel()                                         │
│    - PAUSED → CANCELLED: cancel()                                          │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

Step Run State Machine

┌─────────────────────────────────────────────────────────────────────────────┐
│                      STEP RUN STATE MACHINE                                 │
│                                                                             │
│    ┌──────────┐                                                             │
│    │ PENDING  │ ◄──── Initial state; dependencies not met                  │
│    └────┬─────┘                                                             │
│         │ dependencies met + condition true                                 │
│         ▼                                                                   │
│    ┌──────────┐                                                             │
│    │ RUNNING  │ ◄──── Step is executing                                    │
│    └────┬─────┘                                                             │
│         │                                                                   │
│    ┌────┴────────────────┬─────────────────┐                               │
│    │                     │                 │                               │
│    ▼                     ▼                 ▼                               │
│ ┌───────────┐      ┌───────────┐     ┌───────────┐                        │
│ │ SUCCEEDED │      │  FAILED   │     │  SKIPPED  │                        │
│ └───────────┘      └─────┬─────┘     └───────────┘                        │
│                          │            ▲                                    │
│                          │            │ condition false                    │
│                          ▼            │                                    │
│                    ┌───────────┐      │                                    │
│                    │ RETRYING  │──────┘ (max retries exceeded)             │
│                    └─────┬─────┘                                           │
│                          │                                                  │
│                          │ retry attempt                                    │
│                          └──────────────────┐                              │
│                                             │                              │
│                                             ▼                              │
│                                        ┌──────────┐                        │
│                                        │ RUNNING  │ (retry)                │
│                                        └──────────┘                        │
│                                                                             │
│    Additional transitions:                                                  │
│    - Any state → CANCELLED: workflow cancelled                             │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

Database Schema

-- Workflow Templates
CREATE TABLE release.workflow_templates (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID REFERENCES tenants(id) ON DELETE CASCADE,
    name VARCHAR(255) NOT NULL,
    display_name VARCHAR(255) NOT NULL,
    description TEXT,
    version INTEGER NOT NULL DEFAULT 1,
    nodes JSONB NOT NULL,
    edges JSONB NOT NULL,
    inputs JSONB NOT NULL DEFAULT '[]',
    outputs JSONB NOT NULL DEFAULT '[]',
    tags JSONB NOT NULL DEFAULT '[]',
    is_builtin BOOLEAN NOT NULL DEFAULT FALSE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    created_by UUID REFERENCES users(id)
);

CREATE INDEX idx_workflow_templates_tenant ON release.workflow_templates(tenant_id);
CREATE INDEX idx_workflow_templates_name ON release.workflow_templates(name);

-- Workflow Runs
CREATE TABLE release.workflow_runs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    template_id UUID NOT NULL REFERENCES release.workflow_templates(id),
    template_version INTEGER NOT NULL,
    status VARCHAR(50) NOT NULL DEFAULT 'created',
    context JSONB NOT NULL,
    inputs JSONB NOT NULL DEFAULT '{}',
    outputs JSONB NOT NULL DEFAULT '{}',
    error_message TEXT,
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    created_by UUID REFERENCES users(id)
);

CREATE INDEX idx_workflow_runs_tenant ON release.workflow_runs(tenant_id);
CREATE INDEX idx_workflow_runs_template ON release.workflow_runs(template_id);
CREATE INDEX idx_workflow_runs_status ON release.workflow_runs(status);

-- Step Runs
CREATE TABLE release.step_runs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workflow_run_id UUID NOT NULL REFERENCES release.workflow_runs(id) ON DELETE CASCADE,
    node_id VARCHAR(255) NOT NULL,
    status VARCHAR(50) NOT NULL DEFAULT 'pending',
    inputs JSONB NOT NULL DEFAULT '{}',
    outputs JSONB NOT NULL DEFAULT '{}',
    error_message TEXT,
    logs TEXT,
    attempt_number INTEGER NOT NULL DEFAULT 1,
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    UNIQUE (workflow_run_id, node_id)
);

CREATE INDEX idx_step_runs_workflow ON release.step_runs(workflow_run_id);
CREATE INDEX idx_step_runs_status ON release.step_runs(status);

-- Step Registry
CREATE TABLE release.step_types (
    type VARCHAR(255) PRIMARY KEY,
    display_name VARCHAR(255) NOT NULL,
    description TEXT,
    category VARCHAR(100) NOT NULL,
    icon VARCHAR(255),
    config_schema JSONB NOT NULL,
    input_schema JSONB NOT NULL,
    output_schema JSONB NOT NULL,
    executor VARCHAR(255) NOT NULL DEFAULT 'builtin',
    default_timeout INTEGER NOT NULL DEFAULT 300,
    safe_to_retry BOOLEAN NOT NULL DEFAULT FALSE,
    retryable_errors JSONB NOT NULL DEFAULT '[]',
    documentation TEXT,
    examples JSONB NOT NULL DEFAULT '[]',
    plugin_id UUID REFERENCES release.plugins(id),
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_step_types_category ON release.step_types(category);
CREATE INDEX idx_step_types_plugin ON release.step_types(plugin_id);

Workflow Template Example: Standard Deployment

{
  "id": "template-standard-deploy",
  "name": "standard-deploy",
  "displayName": "Standard Deployment",
  "version": 1,
  "inputs": [
    { "name": "releaseId", "type": "uuid", "required": true },
    { "name": "environmentId", "type": "uuid", "required": true },
    { "name": "promotionId", "type": "uuid", "required": true }
  ],
  "nodes": [
    {
      "id": "approval",
      "type": "approval",
      "name": "Approval Gate",
      "config": {},
      "inputs": [
        { "name": "promotionId", "source": { "type": "context", "path": "promotionId" } }
      ],
      "position": { "x": 100, "y": 100 }
    },
    {
      "id": "security-gate",
      "type": "security-gate",
      "name": "Security Verification",
      "config": {
        "blockOnCritical": true,
        "blockOnHigh": true
      },
      "inputs": [
        { "name": "releaseId", "source": { "type": "context", "path": "releaseId" } }
      ],
      "position": { "x": 100, "y": 200 }
    },
    {
      "id": "deploy-targets",
      "type": "deploy-compose",
      "name": "Deploy to Targets",
      "config": {
        "strategy": "rolling",
        "parallelism": 2
      },
      "inputs": [
        { "name": "releaseId", "source": { "type": "context", "path": "releaseId" } },
        { "name": "environmentId", "source": { "type": "context", "path": "environmentId" } }
      ],
      "timeout": 600,
      "retryPolicy": {
        "maxRetries": 2,
        "backoffType": "exponential",
        "backoffSeconds": 30
      },
      "onFailure": "rollback",
      "position": { "x": 100, "y": 400 }
    },
    {
      "id": "health-check",
      "type": "health-check",
      "name": "Health Verification",
      "config": {
        "type": "http",
        "path": "/health",
        "expectedStatus": 200,
        "timeout": 30,
        "retries": 5
      },
      "inputs": [
        { "name": "targets", "source": { "type": "output", "nodeId": "deploy-targets", "outputName": "deployedTargets" } }
      ],
      "onFailure": "rollback",
      "position": { "x": 100, "y": 500 }
    },
    {
      "id": "notify-success",
      "type": "notify",
      "name": "Success Notification",
      "config": {
        "channel": "slack",
        "template": "deployment-success"
      },
      "onFailure": "continue",
      "position": { "x": 100, "y": 700 }
    },
    {
      "id": "rollback-handler",
      "type": "rollback",
      "name": "Rollback Handler",
      "config": {
        "strategy": "to-previous"
      },
      "inputs": [
        { "name": "deploymentJobId", "source": { "type": "output", "nodeId": "deploy-targets", "outputName": "jobId" } }
      ],
      "position": { "x": 300, "y": 450 }
    }
  ],
  "edges": [
    { "id": "e1", "from": "approval", "to": "security-gate" },
    { "id": "e2", "from": "security-gate", "to": "deploy-targets" },
    { "id": "e3", "from": "deploy-targets", "to": "health-check" },
    { "id": "e4", "from": "health-check", "to": "notify-success" },
    { "id": "e5", "from": "deploy-targets", "to": "rollback-handler", "condition": "status === 'failed'" },
    { "id": "e6", "from": "health-check", "to": "rollback-handler", "condition": "status === 'failed'" }
  ]
}

API Endpoints

See API Documentation for full specification.

# Workflow Templates
POST   /api/v1/workflow-templates
GET    /api/v1/workflow-templates
GET    /api/v1/workflow-templates/{id}
PUT    /api/v1/workflow-templates/{id}
DELETE /api/v1/workflow-templates/{id}
POST   /api/v1/workflow-templates/{id}/validate

# Step Registry
GET    /api/v1/step-types
GET    /api/v1/step-types/{type}

# Workflow Runs
POST   /api/v1/workflow-runs
GET    /api/v1/workflow-runs
GET    /api/v1/workflow-runs/{id}
POST   /api/v1/workflow-runs/{id}/pause
POST   /api/v1/workflow-runs/{id}/resume
POST   /api/v1/workflow-runs/{id}/cancel
GET    /api/v1/workflow-runs/{id}/steps
GET    /api/v1/workflow-runs/{id}/steps/{nodeId}
GET    /api/v1/workflow-runs/{id}/steps/{nodeId}/logs
GET    /api/v1/workflow-runs/{id}/steps/{nodeId}/artifacts