Workflows & Graphs

ask-graph lets you define durable, multi-step workflows with conditional routing, parallel execution, human approval steps, crash recovery, and sub-workflow composition. Each step is a plain Ruby class.

Use ask-graph when you need a reliable multi-step process — order fulfillment, document processing pipelines, approval workflows, data ETL. Steps checkpoint after every completion, so a crash mid-way means you resume where you left off, not from the start.

gem "ask-graph"

Quick Start

require "ask-graph"

# Define a workflow
module ProcessOrder
  class Workflow < Ask::Graph
    step ValidateOrder
    step ChargeCustomer,  if: :valid?
    step SendConfirmation, if: :valid?
    step NotifyAdmin,     unless: :valid?

    private

    def valid?
      context.order.valid?
    end
  end
end

# Run it
result = ProcessOrder::Workflow.call(order: order)
result.order      # => the validated order
result.charged    # => true if payment succeeded

The call class method creates an instance, runs through all steps, and returns a Context object with every step’s output accessible by name.

Defining Steps

Steps are plain Ruby classes with a call(context) method. They read from and write to the context — no return values, no special interfaces.

class ValidateOrder
  def call(context)
    order = context.input[:order]
    context.order = OrderValidator.validate(order)
    context.valid = context.order.errors.empty?
  end
end

When a Hash is passed as input to a workflow, its keys are directly accessible on the context:

result = MyWorkflow.call({ order: order })
result.input       # => { order: order }
result.order       # => the order — same as input[:order]

This works for all hash inputs, including data passed to sub-workflows.

The step contract

A step must implement call(context). That’s it. No base class to inherit, no module to include. The context is an Ask::Graph::Context object — use method access or bracket access, whichever reads better:

class SetValue
  def call(context)
    context.greeting = "hello"      # method setter
    context[:color] = "blue"        # bracket setter
  end
end

class ReadValue
  def call(context)
    puts context.greeting           # => "hello"
    puts context[:color]            # => "blue"
  end
end

Conditional Steps

Use if: or unless: to control whether a step runs. The condition is a method name on the workflow class:

module HandleCall
  class Workflow < Ask::Graph
    step BookAppointment,  if: :booking?
    step EmergencyAlert,   if: :emergency?
    step HandleInquiry,    unless: :known_intent?

    private

    def booking?   = context.intent == "booking"
    def emergency? = context.intent == "emergency"
    def known_intent? = %w[booking emergency inquiry].include?(context.intent)
  end
end

A step is skipped when its condition returns false (if:) or true (unless:). Skipped steps don’t affect the context or trigger hooks. Steps after a skipped step run normally.

Parallel Execution

Use steps (plural) to run multiple steps simultaneously. All steps run in parallel threads, and the workflow waits for all to finish before continuing:

module SyncData
  class Workflow < Ask::Graph
    step FetchRecords

    # All three run in parallel
    steps CrmUpdate, CalendarSync, SendNotification

    step ConfirmResponse
  end
end

Parallel steps share the same context — each step reads from and writes to it concurrently. Use thread-safe operations (the context is mutex-protected). If any step fails, the entire group fails.

Sub-Workflows

A step can delegate to another workflow. This is how you compose larger processes from smaller, reusable workflows.

module NotifyCustomer
  class Workflow < Ask::Graph
    step SendEmail
    step LogNotification
  end
end

Wrap it in a PORO step that calls Workflow.call(context):

module OrderFulfillment
  class NotifyCustomer
    def call(context)
      NotifyCustomer::Workflow.call(context)
    end
  end

  class Workflow < Ask::Graph
    step ValidatePayment
    step NotifyCustomer
    step ShipOrder
  end
end

NotifyCustomer::Workflow.call(context) exports the current context, creates the sub-workflow, runs it, and merges every result back. The sub-workflow sees all the outer context’s data — no special wiring required.

You can also use context.run for the same thing:

class NotifyCustomer
  def call(context)
    context.run(NotifyCustomer::Workflow)
  end
end

Sub-workflows support nesting — a sub-workflow can call another sub-workflow. Timeouts, retries, and conditions on the outer step apply to the entire sub-workflow as a unit.

Suggested directory layout

app/workflows/
  notify_customer/
    workflow.rb          # module NotifyCustomer; class Workflow < Ask::Graph
    steps/
      send_email.rb
      log_notification.rb
  order_fulfillment/
    workflow.rb          # module OrderFulfillment; class Workflow < Ask::Graph
    steps/
      validate_payment.rb
      notify_customer.rb
      ship_order.rb

The directory name acts as the module namespace. workflow.rb holds the graph class. steps/ holds the step POROs.

Human-in-the-Loop (Approve)

Use approve to pause a workflow and wait for external input. The step runs, then the workflow saves its checkpoint and raises a pause signal. Call resume on the same instance to continue:

module ProcessBooking
  class Workflow < Ask::Graph
    step BookAppointment
    approve ReviewBooking, if: :expensive?
    step ConfirmBooking
  end
end

graph = ProcessBooking::Workflow.new(input)
result = graph.call                  # runs, pauses after ReviewBooking

# Later, when the operator responds:
result = graph.resume(input: "approved")  # resumes, runs ConfirmBooking

The approve step’s context changes are preserved in the checkpoint. When you resume, the workflow skips the approve step and continues with the remaining steps.

Timeouts

Set a timeout on any step. If the step takes longer than the given seconds, it raises Ask::Graph::StepFailed.

step FetchApi, timeout: 10, retry: 2

Set a default timeout for all steps in a workflow:

module ApiWorkflow
  class Workflow < Ask::Graph
    timeout 30
    step FastOp           # uses 30s
    step SlowOp, timeout: 120  # overrides
  end
end

Set a global default for every workflow:

Ask::Graph.timeout 30

Resolution order: step timeout: → workflow timeout → Ask::Graph.timeout. Set timeout nil on a child workflow to explicitly clear the parent’s default.

Retry

Retry a step when it fails. The step is retried up to the specified number of times with exponential backoff:

step FlakyOp, retry: 3

If all retries are exhausted, Ask::Graph::StepFailed is raised. The on_failure hook fires before each retry and on the final failure.

step FetchApi, timeout: 10, retry: 2

Timeouts also trigger retries. Each attempt runs within the timeout window, and a timeout counts as a failure for retry purposes.

Lifecycle Hooks

Observe or intervene at each step boundary:

Hook	When it fires
before_step	Before every step
after_step	After every successful step
on_failure	When a step fails

module MonitoredWorkflow
  class Workflow < Ask::Graph
    before_step :log_start
    after_step  :log_completion
    on_failure  :alert_team

    step ProcessPayment, timeout: 15, retry: 2
  end
end

Hooks receive declaration: and context: keyword arguments. on_failure also receives error: with the failure message:

def log_start(declaration:, context:)
  Rails.logger.info "Starting #{declaration[:name]}"
end

def alert_team(declaration:, context:, error:)
  SlackNotifier.alert("Step failed: #{error.message}")
end

Hooks are inherited by subclasses and can be stacked — multiple before_step declarations all fire in order.

Crash Recovery (Checkpointing)

By default, workflows use an in-memory store. If the process crashes, the workflow restarts from the beginning. Pass a storage backend to make workflows durable across restarts:

store = Ask::State::Memory.new   # default, dies on restart
store = RedisStore.new            # survives restarts
store = PostgresStore.new         # or any backend with #set and #get

result = OrderFulfillment::Workflow.call(input, storage: store)

# If a crash occurs, resume from the last completed step:
result = OrderFulfillment::Workflow.call(input, storage: store)

Set a default storage for all workflows:

Ask::Graph.storage RedisPool.new

# Every workflow uses it automatically
result = MyWorkflow.call(input)

Override per-graph or per-call:

class MyWorkflow < Ask::Graph
  storage PostgresStore.new
end

MyWorkflow.call(input)                          # uses PostgresStore
MyWorkflow.call(input, storage: InMemory.new)   # overrides per-call

Per-Item Iteration

When a step needs to process a list of items — sending notifications, processing records — use context.each inside the step. It automatically checkpoints after each item, so a crash mid-list resumes from the last completed item:

class SendVoiceReminders
  def call(context)
    context.each(context.appointments) do |appt|
      PhoneService.call(appt.number, appt.message)
    end
  end
end

module DailyReminders
  class Workflow < Ask::Graph
    step FetchAppointments
    step SendVoiceReminders
    step MarkComplete
  end
end

context.item gives you the current item being processed, accessible anywhere inside the block.

Error Handling

When a step fails, Ask::Graph::StepFailed is raised with the step name in the message:

begin
  MyWorkflow::Workflow.call(input)
rescue Ask::Graph::StepFailed => e
  puts e.message  # => "RiskyOperation failed: connection timeout"
end

The on_failure hook fires before the exception is raised, giving you a chance to log, alert, or clean up.

Step Metadata

Attach a human-readable description to any step:

step ValidateOrder, description: "Verify order details with inventory system"
step ChargeCustomer, description: "Process payment via Stripe"

The description is available in the declaration for debugging, monitoring, and hook callbacks.

Inheritance

Child workflow classes inherit steps and hooks from their parent. Adding new steps to the child doesn’t affect the parent:

class Parent < Ask::Graph
  step ValidateOrder
end

class Child < Parent
  step ChargeCustomer
end

Parent.declarations.size  # => 1
Child.declarations.size   # => 2

Configuration Reference

Option	Where	Purpose
timeout	step option	Max seconds per step
timeout	class method	Default timeout for all steps in this workflow
Ask::Graph.timeout	global	Default timeout for every workflow
retry	step option	Number of retries on failure
storage	class method	Checkpoint backend for this workflow
Ask::Graph.storage	global	Default checkpoint backend for all workflows
storage:	.call param	Override storage per-call

Next Steps

• Explore all core components
• Build custom tools to use inside steps
• Set up monitoring for production workflows