Pranor Flow — DAG Workflow & Saga Orchestrator

Version: 1.0.0
Module Path: github.com/vyuvaraj/pranor/flow
Default Port: 8096
License: AGPL-3.0 (OSS) / Enterprise License (EE with BFT Raft & Visual Designer)

Overview

Pranor Flow is a stateful, DAG-based workflow orchestrator and Saga compensation engine for the Pranor ecosystem. It supports durable execution with checkpointing, WASM step functions, sub-workflow composition, per-execution tracing, a Dead Letter Workflow Queue with manual retry, and automatic reverse compensation on failure.

Pranor Flow can run as:

• A standalone binary with file-based checkpoint persistence
• An integrated module within the Pranor ecosystem with OTel tracing, Pranor Lock leader election, and Console visual designer

Key Features

Architecture

graph TD

    subgraph API ["🌐 Workflow Control Interface"]
        Define["REST DAG Definition API"]
        Exec["Execution Manager API"]
    end

    subgraph Core ["⚡ Core Distributed Saga Orchestrator"]
        Topo["Topological Sort and Dependency Evaluator"]
        HTTPExec["HTTP / REST Step Executor"]
        WASMExec["WASM Sandbox Step Executor"]
        SagaComp["Saga Reverse Compensation Handler"]
    end

    subgraph Storage ["💾 Durable State Persistence"]
        WALStore["WAL Checkpoint Store"]
        DLWQ["Dead-Letter Workflow Queue"]
        BFTConsensus["BFT Raft State Consensus"]
    end

    Define --> Topo
    Exec --> Topo
    Topo --> HTTPExec
    Topo --> WASMExec
    HTTPExec --> WALStore
    WASMExec --> WALStore
    WALStore -.->|On Failure| SagaComp
    SagaComp -.->|Max Retries Exhausted| DLWQ
    WALStore -.-> BFTConsensus

Saga Execution & Compensation Sequence Flow

sequenceDiagram
    autonumber
    participant Client as Client Application
    participant Flow as Pranor Flow Orchestrator
    participant Inventory as Inventory Service
    participant Payment as Payment Gateway
    participant Shipping as Shipping Service
    participant WAL as WAL Checkpoint Store

    Client->>Flow: Execute Workflow (Order Checkout DAG)
    Flow->>Inventory: Step 1: POST /inventory/reserve
    Inventory-->>Flow: 200 OK (Reserved)
    Flow->>WAL: Checkpoint Step 1 Completed
    Flow->>Payment: Step 2: POST /payment/charge
    Payment-->>Flow: 500 Internal Error (Payment Failed)
    Flow->>WAL: Log Step 2 Execution Failure
    Note over Flow,Inventory: Trigger Reverse Compensation Rollback
    Flow->>Inventory: Compensate Step 1: POST /inventory/release
    Inventory-->>Flow: 200 OK (Inventory Unreserved)
    Flow->>WAL: Saga Rollback Completed
    Flow-->>Client: Workflow Execution Failed (Compensated)

Ecosystem Cross-Module Integration

Pranor Flow acts as the primary saga orchestrator across the Pranor ecosystem:

• Pranor Pulse: Dispatches asynchronous event triggers and listens to topic completions during long-running saga steps.
• Pranor Trace: Annotates every workflow execution and individual step with W3C traceparent headers, tracking LLM token costs and latency flamegraphs.
• Pranor Lock: Acquires distributed fencing token leases to ensure saga execution steps are evaluated by a single leader node during failover.
• Pranor Console: Provides a visual DAG designer, live workflow step progress tracking, and 1-click DLQ retry controls.

Installation & Deployment

Binary

cd pranor/flow
go build -o pranor-flow .
./pranor-flow --port 8096

Docker

docker run -p 8096:8096 \
  -v flow-data:/data \
  ghcr.io/vyuvaraj/pranor-flow:latest

With Checkpoint Persistence

./pranor-flow --port 8096 --checkpoint-dir /data/checkpoints

As Part of Pranor Ecosystem

When running under the Pranor platform, Flow integrates automatically with Lock (leader election), Trace (OTel spans), Console (visual designer), and Pulse (event triggers).

Configuration

Environment Variables

YAML Config (flow.yaml)

port: "8096"
checkpoint_dir: "/data/checkpoints"
otel_endpoint: "http://pranor-trace:8090"
wasm_modules_dir: "./wasm"
dlq_max_size: 1000

CLI Flags

API Reference

Base URL: http://localhost:8096

POST /api/workflows/define

Define a new DAG workflow.

Request:

{
  "name": "order-fulfillment",
  "steps": [
    {
      "id": "reserve-inventory",
      "type": "http",
      "url": "http://inventory/reserve",
      "depends_on": [],
      "compensate_url": "http://inventory/release"
    },
    {
      "id": "charge-payment",
      "type": "http",
      "url": "http://payments/charge",
      "depends_on": ["reserve-inventory"],
      "compensate_url": "http://payments/refund"
    },
    {
      "id": "notify-customer",
      "type": "http",
      "url": "http://notifications/send",
      "depends_on": ["charge-payment"]
    }
  ]
}

Response (201):

{
  "id": "wf-def-001",
  "name": "order-fulfillment",
  "step_count": 3,
  "status": "registered"
}

POST /api/workflows/execute

Execute a workflow instance.

Request:

{
  "workflow": "order-fulfillment",
  "input": { "order_id": "ord-123", "amount": 99.99 }
}

Response (200):

{
  "instance_id": "wf-abc-001",
  "status": "running",
  "started_at": "2026-08-01T10:00:00Z"
}

GET /api/workflows/instances/

Get execution status and step logs.

Response (200):

{
  "instance_id": "wf-abc-001",
  "workflow": "order-fulfillment",
  "status": "completed",
  "steps": [
    { "id": "reserve-inventory", "status": "success", "duration_ms": 45 },
    { "id": "charge-payment", "status": "success", "duration_ms": 230 },
    { "id": "notify-customer", "status": "success", "duration_ms": 12 }
  ]
}

POST /api/workflows/resume

Resume a workflow from its last checkpoint.

Request:

{
  "instance_id": "wf-abc-001"
}

Response (200):

{
  "status": "resumed",
  "resumed_from_step": "charge-payment"
}

GET /api/workflows/dlq

Browse Dead Letter Workflow Queue.

Response (200):

{
  "workflows": [
    {
      "instance_id": "wf-xyz-002",
      "workflow": "order-fulfillment",
      "failed_step": "charge-payment",
      "error": "connection timeout",
      "failed_at": "2026-08-01T09:30:00Z"
    }
  ]
}

POST /api/workflows/dlq/{id}/retry

Retry a DLQ workflow.

Response (200):

{
  "status": "retrying",
  "instance_id": "wf-xyz-002"
}

GET /healthz

Liveness probe.

{"status":"UP","service":"pranor-flow","version":"1.0.0"}

Security

Standalone Mode

In standalone mode, Flow runs without authentication. Workflow callbacks are dispatched without auth headers.

Ecosystem Mode (Full Auth Stack)

When running within the Pranor ecosystem:

1. OTel Tracing — every request and step execution gets a span
2. Rate Limiting — per-client request throttling
3. JWT Auth — validates Bearer tokens against Pranor Auth
4. Tenant Isolation — workflows scoped per tenant namespace
5. Callback Auth — configurable bearer token forwarded to step URLs

WASM Sandbox Security

WASM steps execute in a sandboxed environment:

• No filesystem access beyond stdin/stdout
• Per-step execution timeout prevents runaway processes
• Memory limits enforced per WASM module

Observability

Prometheus Metrics

OpenTelemetry Tracing

Every workflow and step generates OTel spans:

• flow.workflow.execute — root workflow span
• flow.step.http — HTTP step execution
• flow.step.wasm — WASM step execution
• flow.saga.compensate — compensation rollback
• flow.dlq.enqueue — DLQ enqueue event

Logging

Structured JSON logs with fields: level, timestamp, trace_id, instance_id, step_id, status, duration_ms.

Enterprise Edition

Operational Runbook

Workflow stuck in "running" state

1. Check /api/workflows/instances/{id} for step-level status
2. Identify which step is blocking — check its callback URL health
3. If step timed out, the workflow may be waiting for checkpoint write
4. Resume from checkpoint: POST /api/workflows/resume
5. If stuck permanently, check disk space for checkpoint directory

Saga compensation failing

1. Check compensate_url endpoints are reachable and returning 200
2. Review logs for compensation step errors
3. Compensations are best-effort — if they fail, manual intervention required
4. Check pranor_flow_compensations_total metric for failure patterns

DLQ growing unbounded

1. Monitor pranor_flow_dlq_depth gauge
2. Review failed workflows in DLQ for common error patterns
3. Fix root cause (downstream service, timeout, etc.)
4. Retry workflows: POST /api/workflows/dlq/{id}/retry
5. Adjust PRANOR_FLOW_DLQ_MAX_SIZE to prevent memory issues

WASM steps timing out

1. Check PRANOR_FLOW_WASM_MODULES_DIR for module availability
2. Review step timeout configuration in workflow definition
3. Check WASM module for infinite loops or excessive memory allocation
4. Monitor pranor_flow_step_duration_ms histogram for WASM steps

v2.0 AgentStep & Saga Engine

In v2.0, Pranor Flow extends with a Saga runner for governed AI agent step execution.

AgentStep Interface

type AgentStep interface {
    Execute(ctx context.Context, input StepInput) (StepOutput, error)
    Compensate(ctx context.Context, input StepInput) error
    Name() string
}

SagaConfig Defaults

Limit Policies

• LimitPolicyCompensate: Automatically unwinds completed steps in reverse order
• LimitPolicyPauseForHITL: Pauses and routes to HITL Approval Queue (EE: Slack/Teams/Email)

Compensation Contract

On step failure, Saga calls Compensate on all previously completed steps in reverse order. Partial compensation failures are recorded in SagaResult.CompensatedSteps but do not prevent the result from being returned.

HITL Approval Queue

The flow/pkg/hitl package provides an in-memory approval queue:

• Submit(req ApprovalRequest) (string, error) — enqueue for review
• Approve(id string, note string) error — mark approved
• Reject(id string, reason string) error — mark rejected with reason
• ListPending() []ApprovalRequest — list outstanding approvals

EE extends with Slack, Microsoft Teams, and Email integrations with SLA timer escalation.