Enterprise Deployment Guide¶

Deploy Helix AI agents and pipelines to production with enterprise-grade reliability. This guide covers graceful shutdown, observability, security, multi-tenancy, clustering, and containerization — everything you need to run intelligent agents at scale.

Table of Contents¶

• Production Deployment
• Building for Production
• Docker Deployment
• Kubernetes Deployment
• Graceful Shutdown
• Observability
• Prometheus Metrics
• OpenTelemetry Tracing
• Structured Logging
• Security
• JWT Authentication
• OAuth2 Token Validation
• Basic Authentication
• API Key Authentication
• TLS Configuration
• Rate Limiting
• Multi-Tenancy
• Audit Logging
• Clustering
• Cluster Manager
• Member States
• Distributed Locks
• Gossip Protocol
• Health Checks and Readiness
• Health Endpoint
• Readiness Endpoint
• Custom Health Checks
• Management API
• Key Endpoints
• Workflow Lifecycle Operations
• Production Checklist
• Event System & Lifecycle Listeners
• Event Types
• EventNotifier Interface
• Event Struct
• Built-in Notifiers
• Implementing a Custom Event Listener
• Event Delivery Semantics
• Advanced Clustering
• Redis Backend Configuration
• Cluster Join/Leave Events
• Health Check Integration with Cluster State
• Distributed Lock Patterns
• Leader Election Details and Failover
• Workflow Distribution
• Custom Interceptors
• Interceptor Interface
• Building a Custom Interceptor
• Chaining Interceptors
• Built-in Interceptors
• Interceptor Ordering
• Reconnection Patterns (Advanced)
• Reconnector Configuration
• Reconnector API
• Backoff Strategy
• Usage Pattern
• Per-Component Reconnection Behavior
• Performance Tuning
• Exchange Hot Path Optimizations
• Exchange Pool
• Zero-Alloc Iterators
• Exchange ID Generation
• SEDA Backpressure Policies
• Script Caching for Starlang Expressions
• Strict YAML Validation
• What is Validated
• Monaco Editor Integration
• OCR Document Processing
• Supported Backends
• Session Management
• Configuration
• Protocol-Specific Behavior

Production Deployment¶

Building for Production¶

Build a statically linked binary:

CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o helix ./cmd/helix

Docker Deployment¶

Generate a multi-stage Dockerfile from your flow:

helix export flow.yaml --output=Dockerfile

This produces:

# Build stage
FROM golang:1.25-alpine AS builder
RUN apk add --no-cache git
WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /helix ./cmd/helix

# Runtime stage
FROM alpine:latest
RUN apk add --no-cache ca-certificates tzdata
COPY --from=builder /helix /helix
COPY flow.yaml /flow.yaml
EXPOSE 8080 8081 9090
CMD ["/helix", "run", "flow.yaml"]

Build and run:

docker build -t my-helix-app .
docker run -p 8080:8080 -p 8081:8081 -p 9090:9090 \
  -e OPENAI_API_KEY="${OPENAI_API_KEY}" \
  -e DATABASE_URL="${DATABASE_URL}" \
  my-helix-app

Kubernetes Deployment¶

apiVersion: apps/v1
kind: Deployment
metadata:
  name: helix-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: helix-app
  template:
    metadata:
      labels:
        app: helix-app
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9100"
    spec:
      containers:
        - name: helix
          image: my-helix-app:latest
          ports:
            - containerPort: 8080
              name: http
            - containerPort: 8081
              name: health
            - containerPort: 9090
              name: mgmt
            - containerPort: 9100
              name: metrics
          livenessProbe:
            httpGet:
              path: /health
              port: 8081
            initialDelaySeconds: 10
            periodSeconds: 15
          readinessProbe:
            httpGet:
              path: /ready
              port: 8081
            initialDelaySeconds: 5
            periodSeconds: 10
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: app-secrets
                  key: database-url
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: "1"
              memory: 512Mi

Graceful Shutdown¶

Helix supports graceful shutdown with in-flight exchange draining. When the process receives SIGINT or SIGTERM:

1. The engine transitions to the Stopping state
2. Workflow consumers stop accepting new messages
3. The InflightRepository tracks active exchanges
4. The engine waits up to drain_timeout for all in-flight exchanges to complete
5. If the timeout elapses, remaining exchanges are abandoned
6. All components are stopped
7. The process exits

Configure the drain timeout:

helix:
  shutdown:
    drain_timeout: "60s"  # default: 30s

The inflight tracker uses atomic counters and sync.WaitGroup for zero-allocation tracking during normal operation.

Observability¶

Prometheus Metrics¶

Enable the metrics endpoint:

helix:
  metrics:
    prometheus:
      enabled: true
      port: 9100

The /metrics endpoint serves Prometheus text exposition format. Key metrics:

The metrics collector (pkg/metrics/) implements Counter, Gauge, and Histogram types using atomic operations -- zero external dependencies.

Grafana Dashboard¶

Create a Grafana dashboard with these queries:

# Exchange throughput (per second)
rate(helix_exchanges_total[5m])

# Error rate
rate(helix_exchanges_failed_total[5m]) / rate(helix_exchanges_total[5m])

# P99 latency
histogram_quantile(0.99, rate(helix_exchange_duration_seconds_bucket[5m]))

# In-flight exchanges
helix_exchanges_inflight

OpenTelemetry Tracing¶

Enable distributed tracing:

helix:
  tracing:
    opentelemetry:
      enabled: true
      endpoint: "http://otel-collector:4318"

Features: - W3C Trace Context propagation (128-bit trace IDs, 64-bit span IDs) - Span attributes: workflow ID, trigger URI, exchange pattern, component scheme - Span events for errors - Sampling: configurable sampling strategy (default: always sample)

When no endpoint is configured, spans are exported to structured logs (useful for development).

The tracing implementation (pkg/tracing/) is fully custom with no OTel SDK dependency. It generates random trace/span IDs, manages span lifecycle, and exports via HTTP to a collector endpoint.

Span Structure¶

Each exchange creates a span hierarchy:

helix.route (workflow ID)
  |-- helix.processor (step 1)
  |-- helix.processor (step 2)
  |-- helix.producer (to: kafka://orders)

Jaeger / Tempo Integration¶

Export to Jaeger or Grafana Tempo via the OTel collector:

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      http:
        endpoint: "0.0.0.0:4318"

exporters:
  jaeger:
    endpoint: "jaeger:14250"

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [jaeger]

Structured Logging¶

Configure JSON logging for log aggregation (ELK, Loki):

helix:
  logging:
    level: "info"
    format: "json"
    components:
      kafka: "debug"
      postgres: "warn"

Log output in JSON format:

{"time":"2026-03-04T10:00:00Z","level":"INFO","msg":"Exchange processed","component":"kafka","workflow":"order-intake","duration":"45ms"}

Per-component overrides allow verbose logging for specific subsystems without increasing global noise.

Security¶

JWT Authentication¶

Validate JWT tokens on incoming exchanges using stdlib crypto (no external JWT libraries).

Supported algorithms: - HS256 -- HMAC-SHA256 with shared secret - RS256 -- RSA-PKCS1v15-SHA256 with public key

YAML DSL:

workflows:
  - id: jwt-protected
    trigger:
      uri: http-server://0.0.0.0:8080/api
    auth:
      type: jwt
      jwt:
        secret: "${env:JWT_SECRET}"
        issuer: auth.myapp.com
        audience: my-api
    steps:
      - target: direct://process

Go API:

security.NewAuthProcessor(&security.AuthConfig{
    Type: "jwt",
    JWT: &security.JWTConfig{
        Secret:   "${JWT_SECRET}",       // for HS256
        PublicKey: "/path/to/public.pem", // for RS256
        Issuer:   "auth.myapp.com",
        Audience: "my-api",
        Claims: map[string]string{
            "role": "admin",
        },
    },
})

Validation checks: 1. Bearer token extraction from Authorization header 2. JWT structure validation (header.payload.signature) 3. Algorithm detection from JWT header 4. Signature verification (HMAC or RSA) 5. Expiration (exp) claim 6. Issuer (iss) claim 7. Audience (aud) claim (string or array) 8. Custom required claims

OAuth2 Token Validation¶

Validate OAuth2 tokens via RFC 7662 token introspection.

YAML DSL:

workflows:
  - id: oauth2-protected
    trigger:
      uri: http-server://0.0.0.0:8080/api
    auth:
      type: oauth2
      oauth2:
        token-url: https://auth.example.com/oauth2/introspect
        client-id: "${env:OAUTH2_CLIENT_ID}"
        client-secret: "${env:OAUTH2_CLIENT_SECRET}"
        scopes: [read, write]
    steps:
      - target: direct://process

Go API:

security.NewAuthProcessor(&security.AuthConfig{
    Type: "oauth2",
    OAuth2: &security.OAuth2Config{
        TokenURL:     "https://auth.example.com/oauth2/introspect",
        ClientID:     "${OAUTH2_CLIENT_ID}",
        ClientSecret: "${OAUTH2_CLIENT_SECRET}",
        Scopes:       []string{"read", "write"},
    },
})

Basic Authentication¶

HTTP Basic auth with bcrypt-hashed passwords.

YAML DSL:

workflows:
  - id: basic-auth-protected
    trigger:
      uri: http-server://0.0.0.0:8080/api
    auth:
      type: basic
      basic:
        users:
          admin: "${env:ADMIN_BCRYPT_HASH}"
          reader: "${env:READER_BCRYPT_HASH}"
    steps:
      - target: direct://process

Go API:

security.NewAuthProcessor(&security.AuthConfig{
    Type: "basic",
    Basic: &security.BasicAuthConfig{
        Users: map[string]string{
            "admin": "$2a$10$...",  // bcrypt hash
            "reader": "$2a$10$...",
        },
    },
})

API Key Authentication¶

Simple header-based API key validation.

YAML DSL:

workflows:
  - id: apikey-protected
    trigger:
      uri: http-server://0.0.0.0:8080/api
    auth:
      type: apikey
      apikey:
        header: X-API-Key
        keys: ["${env:API_KEY_1}", "${env:API_KEY_2}"]
    steps:
      - target: direct://process

Go API:

security.NewAuthProcessor(&security.AuthConfig{
    Type: "apikey",
    APIKey: &security.APIKeyConfig{
        Header: "X-API-Key",  // default: X-API-Key
        Keys:   []string{"key1", "key2"},
    },
})

TLS Configuration¶

Configure TLS for components.

YAML DSL:

workflows:
  - id: tls-secured
    trigger:
      uri: http-server://0.0.0.0:8443/api
    tls:
      cert-file: /certs/server.crt
      key-file: /certs/server.key
      ca-file: /certs/ca.crt
      min-version: "1.2"
      client-auth: require
    steps:
      - target: direct://process

Go API:

tlsCfg := &security.TLSConfig{
    Enabled:    true,
    CertFile:   "/certs/server.crt",
    KeyFile:    "/certs/server.key",
    CAFile:     "/certs/ca.crt",
    MinVersion: "1.2",           // "1.2" or "1.3"
    ClientAuth: "require",       // "none", "request", "require"
}
goTLSConfig, err := tlsCfg.Build()

Features: - X.509 certificate key pairs - Custom CA certificate pools - Minimum TLS version enforcement (1.2 or 1.3) - Mutual TLS (mTLS) with client certificate verification

Rate Limiting¶

Use connector middleware for per-connector rate limiting:

connector.WithRateLimit(100) // 100 requests per second

For workflow-level throttling, use the Throttle orchestration pattern:

steps:
  - throttle:
      rate: 100
      period: "1s"
      steps:
        - target: http://api.example.com/endpoint

Multi-Tenancy¶

Enable tenant isolation across all workflows:

helix:
  tenant:
    enabled: true
    header_name: "X-Tenant-ID"
    required: true
    allowed_tenants:
      - tenant-a
      - tenant-b

The tenant interceptor: 1. Extracts the tenant ID from the specified message header 2. Rejects exchanges with missing tenant headers (if required: true) 3. Validates against the allowlist (if allowed_tenants is non-empty) 4. Sets the helix.tenant exchange property

Downstream processors can use this property for: - Tenant-specific routing (choice with starlang predicate checking properties["helix.tenant"]) - Scoped database queries - Metrics labeling - Audit trail isolation

In AI workloads, multi-tenancy isolates agent sessions, memory stores, and tool access so that one tenant's agent context never leaks into another's.

Audit Logging¶

Enable audit logging to track every exchange:

helix:
  audit:
    enabled: true
    store: "log"           # "memory" or "log"
    max_entries: 50000
    body_preview: 512

In AI workloads, audit trails track agent decisions, tool calls, and LLM interactions, providing a full chain-of-reasoning record for compliance and debugging.

Each audit entry records:

Storage backends: - memory -- in-memory ring buffer (fast, lost on restart) - log -- writes to structured log AND keeps a ring buffer

Query audit entries via the management API:

curl http://localhost:9090/api/audit

Clustering¶

Helix supports clustering via a gossip-based protocol for multi-node deployments.

Cluster Manager¶

The cluster manager (pkg/cluster/) provides: - Membership tracking -- member join, suspect, dead, leave states - Gossip protocol -- UDP-based peer-to-peer communication - Leader election -- deterministic leader selection based on member IDs - Event system -- callbacks for join, leave, failure, leader changes

mgr := cluster.NewManager(&cluster.Config{
    NodeID:            "node-1",
    BindAddress:       "0.0.0.0",
    BindPort:          7946,
    Seeds:             []string{"node-2:7946", "node-3:7946"},
    HeartbeatInterval: 1 * time.Second,
    SuspectTimeout:    5 * time.Second,
    DeadTimeout:       30 * time.Second,
})

mgr.OnEvent(func(event cluster.Event) {
    switch event.Type {
    case cluster.EventMemberJoined:
        log.Info("member joined", "id", event.Member.ID)
    case cluster.EventLeaderElected:
        log.Info("new leader", "id", event.Member.ID)
    }
})

mgr.Start(ctx)
defer mgr.Stop()

Member States¶

Alive --> Suspect --> Dead
  |                    ^
  +--> Left -----------+

Distributed Locks¶

Redis-backed distributed locks (pkg/cluster/redis.go) for coordinating shared resources across nodes.

Gossip Protocol¶

Nodes communicate via UDP gossip messages:

When a node starts, it contacts seed nodes to join the cluster. Each node periodically pings random peers to detect failures. When a member is not seen within the configured timeouts, it transitions through Suspect to Dead.

Health Checks and Readiness¶

Health Endpoint¶

The /health endpoint reports per-component health:

curl http://localhost:8081/health

{
  "status": "UP",
  "checks": {
    "helix": {"status": "UP"},
    "kafka": {"status": "UP"},
    "postgres": {"status": "DOWN", "details": {"error": "connection refused"}}
  }
}

• Returns 200 when all checks are UP
• Returns 503 when any check is DOWN

Readiness Endpoint¶

The /ready endpoint is a simple boolean check:

curl http://localhost:8081/ready

• Returns 200 with {"status":"READY"} when all checks pass
• Returns 503 with {"status":"NOT_READY"} when any check fails

Health checks have a 5-second timeout per check to prevent slow probes from blocking the endpoint.

Custom Health Checks¶

Register custom checks programmatically:

healthMgr.Register("external-api", func(ctx context.Context) *health.Result {
    resp, err := http.Get("https://api.example.com/health")
    if err != nil {
        return &health.Result{
            Status:  health.StatusDown,
            Details: map[string]any{"error": err.Error()},
        }
    }
    defer resp.Body.Close()
    return &health.Result{Status: health.StatusUp}
})

Management API¶

The management API provides runtime control of workflows and observation of the engine state.

Key Endpoints¶

# Instance info
curl http://localhost:9090/api/info

# List all workflows
curl http://localhost:9090/api/workflows

# Workflow detail
curl http://localhost:9090/api/workflows/my-workflow

# Control workflows
curl -X POST http://localhost:9090/api/workflows/my-workflow/stop
curl -X POST http://localhost:9090/api/workflows/my-workflow/start
curl -X POST http://localhost:9090/api/workflows/my-workflow/suspend
curl -X POST http://localhost:9090/api/workflows/my-workflow/resume

# Metrics snapshot
curl http://localhost:9090/api/metrics

# Audit log
curl http://localhost:9090/api/audit

# Enable backlog tracer
curl -X POST http://localhost:9090/api/tracer/enable
curl http://localhost:9090/api/tracer

Workflow Lifecycle Operations¶

Production Checklist¶

• [ ] Enable structured JSON logging for log aggregation
• [ ] Enable Prometheus metrics and configure Grafana dashboards
• [ ] Enable OpenTelemetry tracing and configure Jaeger/Tempo
• [ ] Configure graceful shutdown drain timeout appropriate for your workload
• [ ] Enable health checks and configure Kubernetes liveness/readiness probes
• [ ] Enable audit logging for compliance requirements
• [ ] Configure TLS for all external-facing components
• [ ] Use environment variables for all secrets (API keys, passwords, tokens)
• [ ] Enable multi-tenancy if serving multiple customers
• [ ] Set per-component log levels to reduce noise in production
• [ ] Configure connector lifecycle management with reconnect for critical integrations
• [ ] Use circuit breakers and rate limiters on external API connectors
• [ ] Deploy multiple replicas with clustering for high availability
• [ ] Monitor in-flight exchange counts to right-size drain timeout

Event System & Lifecycle Listeners¶

Helix provides a comprehensive event system for observing engine, workflow, and exchange lifecycle transitions. Events are dispatched through EventNotifier implementations registered on the HelixContext.

Event Types¶

All event types are defined in pkg/engine/events.go:

EventNotifier Interface¶

// EventNotifier receives events from the engine.
type EventNotifier interface {
    Notify(ctx context.Context, event *Event) error
    IsEnabled(eventType EventType) bool
}

The IsEnabled method acts as a filter gate -- the engine calls it before Notify and skips delivery when it returns false. This avoids constructing event payloads for notifiers that do not care about a particular event type.

Event Struct¶

Each event carries contextual metadata:

type Event struct {
    Type      EventType
    Timestamp time.Time
    Source    string              // workflow ID, component name, or context name
    Exchange  *exchange.Exchange  // non-nil for exchange events
    Endpoint  Endpoint            // non-nil for sending events
    Route     *Route              // non-nil for workflow lifecycle events
    Error     error               // non-nil for failure events
    Metadata  map[string]any      // arbitrary key-value pairs
}

For context lifecycle events, Metadata includes operational details such as the number of components and workflows started, or the elapsed startup time. For example, EventContextStarted carries:

Metadata: map[string]any{
    "components": len(components),
    "workflows": len(workflows),
    "elapsed":    time.Since(hc.startTime).String(),
}

Built-in Notifiers¶

Helix ships with three notifier implementations:

LoggingEventNotifier -- logs every enabled event via slog:

notifier := engine.NewLoggingEventNotifier(slog.Default())
hc.AddEventNotifier(notifier)

FilteringEventNotifier -- wraps another notifier and only forwards specific event types:

inner := engine.NewLoggingEventNotifier(nil)
filtered := engine.NewFilteringEventNotifier(inner,
    engine.EventExchangeFailed,
    engine.EventRouteStopped,
)
hc.AddEventNotifier(filtered)

ChannelEventNotifier -- sends events to a Go channel for async processing:

notifier := engine.NewChannelEventNotifier(1000) // 1000-event buffer
hc.AddEventNotifier(notifier)

go func() {
    for event := range notifier.Ch {
        // Process events asynchronously (e.g., forward to external system)
        fmt.Printf("Event: %s from %s\n", event.Type.String(), event.Source)
    }
}()

If the channel buffer is full, events are dropped rather than blocking the engine pipeline.

Implementing a Custom Event Listener¶

Embed EventNotifierSupport to get event type filtering for free, then implement Notify:

type MetricsEventNotifier struct {
    engine.EventNotifierSupport
    exchangesProcessed int64
    exchangesFailed    int64
}

func NewMetricsEventNotifier() *MetricsEventNotifier {
    n := &MetricsEventNotifier{
        EventNotifierSupport: *engine.NewEventNotifierSupport(),
    }
    // Only subscribe to exchange completion events
    n.EnableEvent(engine.EventExchangeCompleted)
    n.EnableEvent(engine.EventExchangeFailed)
    return n
}

func (n *MetricsEventNotifier) Notify(ctx context.Context, event *engine.Event) error {
    switch event.Type {
    case engine.EventExchangeCompleted:
        atomic.AddInt64(&n.exchangesProcessed, 1)
    case engine.EventExchangeFailed:
        atomic.AddInt64(&n.exchangesFailed, 1)
    }
    return nil
}

Register the notifier before starting the context:

hc := engine.NewContext("my-app")
hc.AddEventNotifier(NewMetricsEventNotifier())
hc.Start(ctx)

When no events are explicitly enabled via EnableEvent, EventNotifierSupport.IsEnabled returns true for all event types (opt-in filtering). Use DisableEvent to selectively suppress specific types after enabling others.

Event Delivery Semantics¶

• Events are delivered synchronously within NotifyEvent. Slow notifiers delay the engine, so keep Notify implementations fast.
• Errors returned by Notify are logged but do not halt event delivery to other notifiers.
• Notifiers are called in registration order.
• For async processing, use ChannelEventNotifier or launch a goroutine inside your Notify implementation.

Advanced Clustering¶

Helix provides two clustering backends: a UDP gossip protocol (pkg/cluster/cluster.go) for peer-to-peer discovery, and a Redis-backed coordinator (pkg/cluster/redis.go) for environments where UDP multicast is not available (e.g., Kubernetes, cloud VMs).

Redis Backend Configuration¶

The RedisClusterManager uses Redis for heartbeats, leader election, and distributed locks:

cfg := &cluster.RedisClusterConfig{
    Addr:         "redis:6379",
    Password:     os.Getenv("REDIS_PASSWORD"),
    DB:           0,
    MemberID:     "node-1",           // unique per node
    KeyPrefix:    "helix:",            // namespace for all Redis keys
    HeartbeatTTL: 10 * time.Second,   // member heartbeat expiry
    LeaderTTL:    15 * time.Second,   // leader lock expiry
    LockTTL:      30 * time.Second,   // default distributed lock expiry
}
mgr := cluster.NewRedisClusterManager(cfg)
mgr.Start(ctx)
defer mgr.Stop(ctx)

YAML equivalent:

helix:
  cluster:
    backend: redis
    redis:
      addr: "redis:6379"
      password: "${REDIS_PASSWORD}"
      db: 0
      key_prefix: "helix:"
      heartbeat_ttl: "10s"
      leader_ttl: "15s"
      lock_ttl: "30s"

Redis keys created by the manager:

Heartbeats are published every heartbeat_ttl / 3. Members whose heartbeat key expires are considered dead. The Members(ctx) method discovers live members by scanning helix:member:* keys.

Cluster Join/Leave Events¶

Gossip backend -- the Manager emits five event types via OnEvent:

mgr.OnEvent(func(event cluster.Event) {
    switch event.Type {
    case cluster.EventMemberJoined:
        slog.Info("node joined", "id", event.Member.ID,
            "addr", fmt.Sprintf("%s:%d", event.Member.Address, event.Member.Port))
    case cluster.EventMemberLeft:
        slog.Info("node left gracefully", "id", event.Member.ID)
    case cluster.EventMemberFailed:
        slog.Warn("node failed", "id", event.Member.ID,
            "last_seen", event.Member.LastSeen)
    case cluster.EventLeaderElected:
        slog.Info("new leader", "id", event.Member.ID)
    case cluster.EventLeaderLost:
        slog.Warn("lost leadership", "id", event.Member.ID)
    }
})

Redis backend -- use OnElected and OnRevoked callbacks:

mgr.OnElected(func() {
    slog.Info("this node is now the leader -- starting singleton tasks")
})
mgr.OnRevoked(func() {
    slog.Warn("leadership revoked -- stopping singleton tasks")
})

Health Check Integration with Cluster State¶

When clustering is active, the engine integrates cluster state into workflow lifecycle via HelixContext.SetCluster:

hc := engine.NewContext("my-app")
hc.SetCluster(clusterMgr)

This wires membership change events to automatic singleton workflow rebalancing. When a member joins or leaves:

1. Leader election runs (LeaderElection.Elect())
2. Workflow distribution rebalances (WorkflowDistributor.Rebalance()) using FNV-1a consistent hashing
3. Singleton workflows are started or suspended on each node via reconcileSingletonRoutes()

Routes marked Singleton: true or using singleton-by-nature schemes (cron://, timer://) are automatically distributed -- only the assigned node runs each singleton workflow. Non-singleton workflows run on all nodes.

Distributed Lock Patterns¶

In-memory locks (gossip backend) via LockManager:

lockMgr := cluster.NewLockManager("node-1")

// Acquire with TTL (auto-expires if holder crashes)
acquired, err := lockMgr.Acquire(ctx, "batch-processor", 30*time.Second)
if err != nil {
    return err
}
if !acquired {
    slog.Info("lock held by another node, skipping")
    return nil
}
defer lockMgr.Release("batch-processor")

// ... do exclusive work ...

The in-memory lock manager supports re-entrant acquisition (same node can extend TTL) and automatic expiry. Use IsHeld(name) to check lock status and Holder(name) to identify the current owner.

Redis-backed distributed locks (production-grade):

redisMgr := cluster.NewRedisClusterManager(cfg)
lock := redisMgr.CreateLock("batch-processor")

// Blocking acquire (exponential backoff 50ms -> 2s)
if err := lock.Lock(ctx); err != nil {
    return fmt.Errorf("failed to acquire lock: %w", err)
}
defer lock.Unlock(ctx)

// Extend TTL for long-running operations
if err := lock.Extend(ctx, 60*time.Second); err != nil {
    slog.Warn("failed to extend lock", "error", err)
}

Redis locks use SET NX PX for acquisition and a Lua script for atomic check-and-delete on release, following the single-instance Redlock pattern. The Extend method atomically extends the TTL only if the caller still owns the lock.

For non-blocking attempts, use TryLock:

acquired, err := lock.TryLock(ctx)
if err != nil {
    return err
}
if !acquired {
    return nil // skip, another node holds the lock
}
defer lock.Unlock(ctx)

The RedisLockFactory creates locks under a shared key prefix:

factory := cluster.NewRedisLockFactory(redisClient, "helix:", 30*time.Second)
lock := factory.CreateLock("my-resource") // key: helix:lock:my-resource

Leader Election Details and Failover¶

Gossip-based election -- deterministic: all alive member IDs are sorted alphabetically, and the first one is elected leader. Every membership change triggers re-election. Callbacks fire on transitions:

le := cluster.NewLeaderElection(mgr)
le.OnElected(func() {
    // Start leader-only work (e.g., schedule batch jobs)
})
le.OnDemoted(func() {
    // Stop leader-only work
})
le.Elect() // called automatically on membership changes

Query leadership state:

leaderID := le.Leader()      // returns the elected leader's ID
isLeader := le.IsLeader()    // true if this node is the leader

Redis-based election -- uses SETNX with a TTL. The leader must renew every ttl/3. If the leader crashes, the lock expires and another member claims it within one renewal interval:

le := cluster.NewRedisLeaderElection(redisClient, "helix:leader", "node-1", 15*time.Second)
le.OnElected(func() { /* ... */ })
le.OnRevoked(func() { /* ... */ })
le.Start(ctx)
defer le.Stop(ctx)

Failover timeline with default LeaderTTL: 15s: - Leader crashes at T=0 - Lock expires at T=15s - Surviving nodes attempt SETNX every 5s (ttl/3) - New leader elected between T=15s and T=20s (worst case)

To reduce failover time, lower LeaderTTL (minimum practical: 5s with 500ms renewal). On graceful shutdown, Stop(ctx) releases the leader lock immediately via the Lua unlock script, allowing instant failover without waiting for TTL expiry.

Workflow Distribution¶

The RouteDistributor assigns singleton workflows to cluster members using FNV-1a hashing over sorted alive member IDs:

rd := cluster.NewRouteDistributor(mgr)
rd.Assign("my-cron-workflow")        // returns the assigned member ID
rd.IsLocal("my-cron-workflow")       // true if assigned to this node
assignments := rd.Assignments()    // map[routeID]memberID

When called automatically via SetCluster, the distributor rebalances on every membership change, and the engine starts/suspends singleton workflows accordingly.

Custom Interceptors¶

Interceptors wrap every exchange passing through the processor pipeline. They execute Before logic before any processor runs and After logic after all processors complete (or fail).

Interceptor Interface¶

type Interceptor interface {
    Before(ctx context.Context, exchange *exchange.Exchange) error
    After(ctx context.Context, exchange *exchange.Exchange) error
}

Both methods receive the full exchange with access to headers, properties, body, and exception state. Returning an error from Before short-circuits the entire pipeline. Returning an error from After is captured, but only if the pipeline itself did not already produce an error.

Building a Custom Interceptor¶

Example: a timing interceptor that measures processing duration and sets a response header:

type TimingInterceptor struct{}

func (t *TimingInterceptor) Before(ctx context.Context, ex *exchange.Exchange) error {
    ex.SetProperty("timing.start", time.Now())
    return nil
}

func (t *TimingInterceptor) After(ctx context.Context, ex *exchange.Exchange) error {
    if start, ok := ex.Property("timing.start").(time.Time); ok {
        duration := time.Since(start)
        ex.In().SetHeader("X-Processing-Time", duration.String())
        slog.Info("exchange processed",
            "id", ex.ID(),
            "workflow", ex.WorkflowID(),
            "duration", duration,
        )
    }
    return nil
}

Register it on the context:

hc := engine.NewContext("my-app")
hc.AddInterceptor(&TimingInterceptor{})

Chaining Interceptors¶

Multiple interceptors execute in registration order for Before and the same order for After:

hc.AddInterceptor(&TenantInterceptor{})   // 1st
hc.AddInterceptor(&AuditInterceptor{})    // 2nd
hc.AddInterceptor(&TimingInterceptor{})   // 3rd

The pipeline execution order is:

TenantInterceptor.Before()
AuditInterceptor.Before()
TimingInterceptor.Before()
  [Processor 1] -> [Processor 2] -> ... -> [Processor N]
TimingInterceptor.After()
AuditInterceptor.After()
TenantInterceptor.After()

Note: After methods are called in the same forward order as Before (not reversed). This matches the implementation in BuildPipeline (pkg/engine/context.go).

Built-in Interceptors¶

Example: creating an AuditInterceptor (from pkg/engine/audit.go):

store := engine.NewSlogAuditStore(slog.Default(), 50000)
auditInterceptor := engine.NewAuditInterceptor(store, 512) // 512-byte body preview
hc.AddInterceptor(auditInterceptor)

Interceptor Ordering¶

Interceptors added via AddInterceptor are appended to an ordered slice. The order matters:

• Security interceptors (tenant validation, auth) should come first so unauthorized exchanges are rejected before audit or metrics recording.
• Audit interceptors should come early to capture all exchanges, including those rejected by later interceptors.
• Metrics/SLA interceptors should come after security so they only measure authorized traffic.
• Tracing interceptors (BacklogTracer) are typically added last via SetTracer, which appends to the interceptor list.

The Interceptors() method returns a copy of the current interceptor slice, and BuildPipeline receives this snapshot when a workflow starts. Adding interceptors after route startup does not affect already-running routes.

Reconnection Patterns (Advanced)¶

The Reconnector utility (pkg/engine/reconnect.go) provides reusable exponential backoff with jitter for any component that needs automatic reconnection.

Reconnector Configuration¶

type ReconnectConfig struct {
    InitialBackoff time.Duration // first retry delay (default: 500ms)
    MaxBackoff     time.Duration // backoff ceiling (default: 30s)
    MaxAttempts    int           // 0 = unlimited retries
    JitterFraction float64       // 0.0-1.0, randomness fraction (default: 0.2)
}

Use DefaultReconnectConfig() for sensible defaults:

rc := engine.NewReconnector("my-component", engine.DefaultReconnectConfig())
// Returns: InitialBackoff=500ms, MaxBackoff=30s, MaxAttempts=0, JitterFraction=0.2

Or configure custom values:

rc := engine.NewReconnector("kafka-producer", engine.ReconnectConfig{
    InitialBackoff: 1 * time.Second,
    MaxBackoff:     60 * time.Second,
    MaxAttempts:    10,
    JitterFraction: 0.25,
})

Reconnector API¶

Backoff Strategy¶

The backoff doubles after each failed attempt:

Attempt 1: 500ms  +/- jitter
Attempt 2: 1s     +/- jitter
Attempt 3: 2s     +/- jitter
Attempt 4: 4s     +/- jitter
...
Attempt N: min(2^(N-1) * InitialBackoff, MaxBackoff) +/- jitter

Jitter is calculated as: backoff * JitterFraction * (2*rand - 1), meaning the actual wait varies by +/- JitterFraction of the current backoff. This prevents thundering-herd reconnections when many clients lose connectivity simultaneously.

Usage Pattern¶

rc := engine.NewReconnector("nats", engine.DefaultReconnectConfig())

// Initial connection
err := rc.Connect(func() error {
    return natsConn.Connect(natsURL)
})
if err != nil {
    return fmt.Errorf("initial NATS connection failed: %w", err)
}

// Later, when a publish fails:
if err := natsConn.Publish(subject, data); err != nil {
    rc.MarkDisconnected()
    go func() {
        if err := rc.RunReconnectLoop(ctx, func() error {
            return natsConn.Reconnect(natsURL)
        }); err != nil {
            slog.Error("reconnection gave up", "component", "nats", "error", err)
        }
    }()
}

// On shutdown:
rc.Reset()

Per-Component Reconnection Behavior¶

NATS (pkg/component/nats/): - Producer: retry-once on publish failure, then triggers RunReconnectLoop in a background goroutine. - Consumer: connection monitor goroutine detects disconnect and runs RunReconnectLoop. - Uses the Reconnector utility.

AMQP (pkg/component/amqp/): - Producer: reconnects on publish failure via RunReconnectLoop, then retries the publish. - Consumer: monitors the delivery channel; when it closes, runs RunReconnectLoop to re-establish the channel and re-declare queues. - Uses the Reconnector utility.

Kafka (segmentio/kafka-go): - Reconnection handled internally by the kafka-go Writer and Reader -- they maintain connection pools and retry automatically. - No Reconnector wrapper needed.

Redis (go-redis/v9): - Reconnection handled internally by go-redis's connection pool with built-in retry and backoff. - No Reconnector wrapper needed.

PostgreSQL (pgx/v5): - Uses pgxpool which manages connection pooling and reconnection internally. - No Reconnector wrapper needed.

MongoDB (mongo-driver/v2): - The driver handles reconnection and server selection automatically. - No Reconnector wrapper needed.

Performance Tuning¶

Exchange Hot Path Optimizations¶

The Exchange struct (pkg/exchange/exchange.go) uses lock-free atomic operations instead of mutexes on all hot-path fields:

type Exchange struct {
    id         string
    pattern    atomic.Int32              // ExchangePattern (InOnly/InOut)
    in         atomic.Pointer[Message]   // input message
    out        atomic.Pointer[Message]   // output message
    properties sync.Map                  // exchange-scoped properties
    exception  atomic.Pointer[error]     // failure state
    routeID    atomic.Value              // string
    fromURI    atomic.Value              // string
    // ...
}

The Message body uses atomic.Value via a bodyWrapper pattern, and headers use sync.Map. These changes eliminate all sync.RWMutex contention on the exchange hot path.

Measured impact (from pkg/exchange/exchange_bench_test.go): - Single-goroutine: ~9.5x faster than the previous mutex-based implementation - Parallel (GOMAXPROCS goroutines): 575-800x faster due to eliminated lock contention

Exchange Pool¶

For high-throughput workflows, use the sync.Pool-backed exchange pool to reduce GC pressure:

import "github.com/arvenza/helix/pkg/exchange"

ex := exchange.AcquireExchange(ctx)
defer exchange.ReleaseExchange(ex)

// Use the exchange normally
ex.In().SetBody("hello")
ex.In().SetHeader("type", "greeting")

AcquireExchange reuses a previously pooled Exchange, resetting all fields (pattern, messages, properties, exception, workflow/trigger metadata). ReleaseExchange clears all references to allow GC of message contents before returning the struct to the pool.

Measured impact: allocation cost drops from ~500ns to ~334ns per exchange (33% reduction), with one fewer heap allocation per exchange lifecycle.

Important: never use an exchange after calling ReleaseExchange. The pool may hand the same struct to another goroutine immediately.

Zero-Alloc Iterators¶

The standard Headers() and Properties() methods allocate a new map[string]any on every call. For hot paths where you only need to read (not modify), use the zero-allocation iterators:

// Instead of: for k, v := range ex.In().Headers() { ... }
ex.In().ForEachHeader(func(key string, value any) bool {
    // Process each header without allocating a map copy
    if key == "Content-Type" {
        // found it
        return false // stop iteration
    }
    return true // continue
})

// Instead of: for k, v := range ex.Properties() { ... }
ex.ForEachProperty(func(key string, value any) bool {
    // Process each property without allocation
    return true
})

Measured impact (from benchmarks): - ForEachHeader: ~22x faster than Headers() (eliminates map allocation + copy) - ForEachProperty: ~15x faster than Properties()

Exchange ID Generation¶

Exchange IDs use strconv.AppendUint with a pre-sized stack buffer instead of fmt.Sprintf:

var exchangeCounter atomic.Uint64

func nextExchangeID() string {
    id := exchangeCounter.Add(1)
    var buf [29]byte
    copy(buf[:9], "exchange-")
    s := strconv.AppendUint(buf[:9], id, 10)
    return string(s)
}

This reduces ID generation from 86ns/2 allocs to 46ns/1 alloc.

SEDA Backpressure Policies¶

The SEDA component (pkg/component/seda/) uses Go channels as async queues. When the queue is full, the overflowPolicy URI parameter controls producer behavior:

Configure via URI parameters:

workflows:
  - from: "kafka://orders"
    steps:
      - target: "seda://process-orders?size=1000&overflowPolicy=block"

  - from: "seda://process-orders?concurrency=10"
    steps:
      - target: "http://processor:8080/orders"

Guidelines for choosing a policy: - fail: Use when message loss is unacceptable and callers can handle errors (e.g., retry or dead-letter). - block: Use when upstream can tolerate backpressure delays (e.g., internal pipelines). Respects context cancellation. - drop: Use for best-effort delivery where some message loss is acceptable (e.g., metrics, telemetry). - newest: Use for "latest value wins" semantics (e.g., sensor readings, status updates).

Script Caching for Starlang Expressions¶

Starlang is the default expression language for choice, filter, transform, and all orchestration pattern predicates. Each expression string is parsed once and cached in a sync.Map (pkg/expression/expression.go):

var scriptCache sync.Map // script string -> *cachedParsed

const maxScriptCacheSize = 10000

On first evaluation, the script is parsed into an AST (*ast.Program), and the result (including parse errors) is cached. Subsequent evaluations of the same script skip parsing entirely.

The cache is bounded at 10,000 entries. When the limit is reached, the entire cache is cleared (simple eviction strategy that avoids LRU overhead). In practice, the number of distinct expressions in a running application is typically well under this limit.

For workflows with many dynamic expressions, ensure scripts are deterministic strings (not generated with varying whitespace or formatting) to maximize cache hit rates.

No configuration is needed -- caching is always active and transparent to workflow definitions.

Strict YAML Validation¶

The validate.go module enforces strict structural validation of YAML flow definitions before workflow construction begins. This catches configuration errors at parse time rather than at runtime.

What is Validated¶

The validation suite covers 27 test cases across all five top-level primitives (agents, tools, pipelines, mcps, workflows).

Monaco Editor Integration¶

For browser-based editing, Helix generates a JSON Schema from the YAML DSL definition types. This schema powers Monaco editor features:

• Real-time validation -- red squiggles on invalid fields as you type
• Autocomplete -- suggests valid keys, enum values, and component URIs
• Hover documentation -- shows field descriptions from the schema

Enable it by serving the schema from the management API:

GET /api/schema/flow.json

Point Monaco's yaml.schemas configuration at this endpoint.

OCR Document Processing¶

Helix supports OCR-based document ingestion through the rag-ingest step with reader: ocr. Eight OCR backends are available, selectable via the ocr_backend parameter.

Supported Backends¶

Example¶

pipelines:
  - id: ocr-ingest
    source: http-server://0.0.0.0:8080/ocr
    steps:
      - rag-ingest:
          reader: ocr
          ocr_backend: docling
          ocr_options:
            languages: ["en", "de"]
            dpi: 300
      - rag-chunk:
          strategy: recursive
          chunk_size: 512
      - rag-embed:
          provider: openai
          model: text-embedding-3-small
          api_key: "${env:OPENAI_API_KEY}"
      - rag-store:
          store: memx
          connection: "http://localhost:8080"
          collection: scanned_docs

Session Management¶

Helix provides built-in session management for stateful protocol servers (AG-UI, A2A, MCP, A2UI). Sessions track conversation state, memory context, and user identity across multiple requests.

Configuration¶

helix:
  sessions:
    enabled: true
    store: memory          # memory | redis | postgres
    ttl: 30m               # session expiration
    max_sessions: 10000    # per-instance limit
    cookie_name: helix-sid

Protocol-Specific Behavior¶

Sessions are automatically created on first request and garbage-collected after TTL expiration. For clustered deployments, use redis or postgres as the session store to share state across instances.