Guide

# Workflows

Sequential, durable steps that span hours or days, with waits, flow control, exception policies and an audit trail.

## Install

Workflows keep their state in four tables. A fresh `pgbus:install` creates them; an existing install adds them with the generator (or `rails generate pgbus:update`, which detects the missing tables):

```shell
rails generate pgbus:add_workflows               # or --database=pgbus for a separate database
rails db:migrate
```

An install that ran `pgbus:add_workflows` before waiting steps existed adds them (the `pgbus_workflow_signals` table and five step columns) with `rails generate pgbus:add_workflow_signals`; `pgbus:update` detects that too. `start!` of a class with a `fan_out` or `wait_for:` step raises `Pgbus::Workflow::WaitingStepsUnmigrated` until it has run.

## Define and start a workflow

A workflow is a class with declared steps. Each step runs once per attempt, in order, as its own job; `wait:` delays it after the one before. Inside a step, `subject` is the record the run was started for and `arguments` the keywords `start!` received.

```ruby
class SignupWorkflow < Pgbus::Workflow
  cancel_if { subject.nil? || subject.closed? }
  on_exception Net::OpenTimeout, action: :reattempt!, wait: 15.seconds, max_reattempts: 5

  step :welcome do
    WelcomeMailer.with(user: subject, plan: arguments[:plan]).welcome.deliver_later
  end

  step :reminder, wait: 2.days, skip_if: -> { subject.onboarded? } do
    ReminderMailer.with(user: subject).remind.deliver_later
  end

  step :complete, wait: 7.days do
    subject.update!(onboarding_completed: true)
  end
end

wf = SignupWorkflow.start!(user, plan: "pro")
# or, started by a person, with who and why on the audit trail (not in arguments):
#   wf = SignupWorkflow.start!(user, plan: "pro", actor: "alice", reason: "invited by support")
wf.status        # => "ready"
wf.workflow_id   # => "6f1c…"
```

Step forms: `step :name do … end`, an anonymous `step { … }` (named `step_1`, `step_2`, …), `step :name` to run the instance method of that name, and `step def name = …`. Options, all checked when the class loads:

- `wait:` — seconds or a Duration after the previous step (for the first step, after `start!`). Underneath it is a PGMQ delay: the StepJob is sent with `wait_until:`.
- `skip_if:` — a lambda or method name; truthy skips the step. On a `wait_for:` step it runs on every claim, the one after `signal!` woke the step included: see "Skip a wait the record already passed" under Signals.
- `timeout:` — how long the step may stay in progress before the sweep treats it as stalled (default `config.workflow_stall_threshold`).
- `job_options:` — `queue`, `priority`, `wait`, `wait_until` for this step's StepJob; `set_step_job_options` sets them for the class. `wait`/`wait_until` only ever move a step later: they are folded into its `scheduled_at` when it is scheduled. Anything else raises.
- `on_exception:` — a step-level exception policy (below).
- `wait_for:`, `wait_timeout:`, `on_timeout:` — wait for a signal, with a fixed deadline or one per run (below).
- `before_step:` / `after_step:` — insert the step before or after another (useful in a subclass).

`cancel_if` runs before every step; truthy cancels the run. Both it and `skip_if` run under the workflow's row lock, before the body.

The subject is optional and loaded through GlobalID; a record that is gone makes `subject` return `nil` (react in `cancel_if`). Arguments go through `ActiveJob::Arguments` (GlobalID-aware, never Marshal) and are capped at `config.workflow_max_payload_size` (128 KB). A subject has at most one ongoing run per workflow class: a second `start!` raises `Pgbus::Workflow::AlreadyRunning`, whose `#workflow` is the run holding the slot. Subject-less runs are never limited.

`idempotency_key:`, `actor:` and `reason:` are reserved keywords of `start!`: they never reach `arguments`. `actor:` (default `"app"`) and `reason:` go on the run's `started` event, and the actor on its first `step_scheduled`, as on every outside verb, so the audit trail says who started a run and why. A workflow that took an argument named `actor` or `reason` renames it (`requested_by:`).

`start!` writes the run and its first step row in one transaction; the StepJob is sent by the step row's after-commit callback, so calling `start!` inside your own transaction sends nothing until it commits (and nothing at all if it rolls back) when the workflow tables share your database. On a separate pgbus database (`connects_to`) the commit is the pgbus connection's: the run is written and its first step sent on their own, so a step can run before your transaction commits the subject (and even if it rolls back). Wait for the subject in the step body — `reattempt!(wait: 5) if subject.nil?` — rather than in `cancel_if`, which cancels the run.

## Flow control

Inside a step body, each verb ends the body on the spot:

- `pause!` — the step completes and the run pauses; `resume!` continues with the next step.
- `cancel!` — the run is cancelled.
- `reattempt!(wait: 30)` — this attempt fails and a new one is scheduled.
- `skip!` — the step is skipped and the run moves on.
- `finished!` — the step completes and the run finishes now.

From outside (the console, your code; the dashboard later), every verb takes `actor:` and `reason:` for the audit trail. Outside a step, `skip!` skips the step a waiting run waits on:

```ruby
wf = SignupWorkflow.find(workflow_id)
wf.pause!(actor: "alice", reason: "SMTP is down")
wf.resume!(actor: "alice", reason: "SMTP fixed")   # a failed step runs again; a completed one is followed by the next
wf.resume_at!(:reminder)                            # paused → run :reminder next
wf.run_now!                                         # the waiting step runs now
wf.reschedule!(at: 1.hour.from_now)
wf.cancel!(reason: "customer asked")
wf.rerun!                                           # a finished or cancelled run, started again
wf.skip!(actor: "alice", reason: "picked up in store")  # waiting → skip the step it waits on
wf.available_actions                                # => [:pause, :cancel, :run_now, :reschedule, :signal]
```

A verb that is not legal in the run's state raises `Pgbus::Workflow::InvalidTransition` (`NotPaused` for `resume!` and `resume_at!`); the table at the bottom of this page is the rule.

## Fan-out steps

A `fan_out` step adds a group of jobs and waits for all of them, as one batch:

```ruby
fan_out :notify_vendors, on_batch_failure: :pause! do |b|
  b.add(*subject.vendors.map { |vendor| NotifyVendorJob.new(vendor.id, "#{step_idempotency_key}:#{vendor.id}") })
end
```

The body adds job instances with `b.add`, as often as it likes (each keeps its own `.set(queue:, wait:, priority:)`; a block given to `b.add` runs inside the batch, so a `perform_later` in it joins too). When the body returns, one batch opens with those jobs, described `OrderWorkflow#notify_vendors <workflow_id>` and carrying `{ workflow_id, step_id }` as its properties, so it shows on the Batches page. The step is `waiting`, and so is the run; no message is sent for it.

When every member has finished (completed or dead-lettered), the batch's `on_complete` callback, `Pgbus::Workflow::BatchFinishedJob` on the step's StepJob queue, resumes the run: with no dead-lettered member the step completes and the next one is scheduled; with any, `on_batch_failure` applies (`:pause!` by default, `:cancel!` or `:skip!`) and `Pgbus::Workflow::FanOutFailed` ("3 of 12 jobs dead-lettered") is recorded on the step. A batch that stalls is repaired by the batch sweep, whose finish fires the same callback; a callback that is lost (dead-lettered, or never sent) is made up by the workflow sweep once its batch has been finished for `config.workflow_stall_threshold`. If the batch row itself is gone (batch retention ran before a long-delayed callback), its outcome is unknown: the step fails with `FanOutFailed` and the run **pauses**, whatever `on_batch_failure` says, for a person to resume or skip.

- The workflow owns the batch's callbacks: `b.on_complete`, `b.on_success` and `b.on_failure` raise `ArgumentError` in the body.
- A flow-control verb (`skip!`, `cancel!`, …) or an exception in the body ends the step before any batch exists. Members join only through `b.add`: a bare `perform_later` in the body is not a member.
- An empty batch finishes at once; the step completes without waiting.
- **A reattempt, or `resume!` after a failure, runs the whole body again and opens a fresh batch.** Every member runs again, so member jobs must be idempotent: derive each member's key from `step_idempotency_key` (stable across attempts) plus something that tells the members apart, such as the vendor id.
- `fan_out` takes every `step` option except `wait_for:`, `wait_timeout:` and `on_timeout:`, and needs the batch callback-job columns (`rails generate pgbus:add_batch_callback_jobs`): without them `start!` raises `Pgbus::Workflow::BatchCallbacksUnmigrated`.

## Signals

A `wait_for:` step waits for an event from outside, delivered with `signal!`:

```ruby
class OrderWorkflow < Pgbus::Workflow
  step :capture, wait_for: :payment_confirmed, wait_timeout: 2.days, on_timeout: :pause! do
    PaymentGateway.capture(subject, amount: received_signal.payload["amount"])
  end

  # Late after an hour, and again every hour, until the approval comes.
  step :approve, wait_for: :approved, wait_timeout: 1.hour, on_timeout: :notify do
    subject.update!(approved_by: received_signal.payload["by"])
  end
end

wf.signal!(:payment_confirmed, payload: { amount: 42 }, idempotency_key: event.id)

# A webhook that knows the order, not the run:
OrderWorkflow.signal!(order, :payment_confirmed, payload: { amount: 42 },
                      idempotency_key: event.id, missing: :ignore)
```

The step reaches its gate after its `wait:` and its guards. It takes the oldest pending signal its matcher takes and runs with it as `received_signal`; with none it parks as `waiting` and nothing is sent. `signal!` stores the signal under the run's row lock and, when the run's step is waiting for one it matches, claims it and schedules the step; its StepJob is sent once the signal commits (so a `signal!` inside your transaction wakes nothing before it commits), into a queue the workers already LISTEN on, so the step runs at once.

| Signal status | Means |
| --- | --- |
| `pending` | Stored, taken by no step: it arrived before the step waited, or no waiting step matches it. |
| `claimed` | A step took it and runs with it (`claimed_by_step_id`). |
| `consumed` | The step that took it ended. |

- **Buffering.** A signal sent before its step waits stays pending and is taken when the step reaches its gate (unless a guard skips the step first; see below).
- **Idempotency.** With an `idempotency_key`, a redelivery stores nothing, wakes nothing and returns the stored signal, whose `duplicate_delivery?` is true. A blank key (`""`) is no key.
- **Audit.** `actor:` and `reason:` (both optional, on the handle and by subject) go on the run's `signal_received` event: `wf.signal!(:shipped, actor: "webhook", reason: "carrier callback")`.
- **Matchers.** `wait_for:` takes a signal name, a `Pgbus::Workflow::SignalMatcher.new(:shipped) { |payload| payload[:region] == "eu" }`, or any object answering `matches?(signal)`. A matcher runs under the run's row lock, so it must be fast and free of side effects. One that raises at the gate is handed to the exception policies; in `signal!` the error reaches the caller and nothing is stored.
- **Payloads** go through `ActiveJob::Arguments`, are capped at `config.workflow_max_payload_size`, and read back with indifferent access (`received_signal.payload["amount"]`).
- **A finished or cancelled run** raises `Pgbus::Workflow::NotOngoing`. By subject, `missing: :raise` (default) raises it when the subject has no ongoing run of the class; `missing: :ignore` returns `nil`. A subject-less run is signalled through its handle (`OrderWorkflow.find(workflow_id).signal!`).
- **Lifecycle.** A claimed signal is consumed when its step ends, and kept when the attempt reattempts or fails into a pause: the next attempt (`reattempt!`, `resume!`, or `resume_at!` of that step) runs with the same signal, without matching again.
- **A paused run** still takes a signal for its waiting step, but stays paused; `resume!` runs the step. `resume!` of a run paused while it waited puts it back to waiting.

**Skip a wait the record already passed.** A run that starts late, or a later signal that overtakes an earlier wait, finds the record past the state the step waits for. The guard for that reads `signal_ready?`:

```ruby
class OrderWorkflow < Pgbus::Workflow
  step :await_shipped, wait_for: :shipped,
                       skip_if: -> { subject.shipped? && !signal_ready? } do
    output(shipped_at: subject.shipped_at)
  end
end
```

Guards run before the gate, on every claim of the step. Without the reader, `skip_if: -> { subject.shipped? }` would also skip the step `signal!` woke once the order shipped (its body would never run), and a signal sent before the step's first claim would stay pending for good. Two step readers, usable in `skip_if:`, `cancel_if` and the body, tell the cases apart:

- `signal_ready?` — the gate runs this attempt with a signal: one the step holds, or a pending one its `wait_for:` matcher takes. The guard above skips only when no signal is held or pending for the step; a woken step and one whose signal came first both complete: the gate claims the signal, and it is consumed when the step ends.
- `woken?` — narrower: this attempt holds a signal before its gate (`signal!` woke it, or an earlier attempt took one and reattempted or failed into a pause). False on a first claim, even when a buffered signal waits for the gate, so `!woken?` in that guard skips the step and leaves that signal pending.
- Both only read (the gate claims, after the guards), keep one answer for the whole attempt, are false after an `on_timeout: :run` wake (the gate takes no signal then) and in a step without `wait_for:`, and raise `Pgbus::Workflow::NotInStep` outside a step.
- **A guard that skips** (or a `cancel_if` that cancels) takes no signal: a pending one stays pending for a later `wait_for:` step that matches it, as with an outside `skip!`. A signal the step already holds is consumed, as when the body calls `skip!`.

**Timeouts.** A signal that never comes is the usual way a workflow gets stuck, so a waiting step can carry a deadline: `wait_timeout:` (seconds or a Duration) and `on_timeout:`, one of `:pause!` (default: the step fails with `Pgbus::Workflow::WaitTimedOut`), `:cancel!`, `:skip!`, `:run` (the body runs with `received_signal` nil and `timed_out?` true) or `:notify` (below). `timeout:` keeps its meaning on every step: it bounds the in-progress attempt from its claim, while `wait_timeout:` bounds the wait before it. The sweep fires timeouts, so one fires within `config.workflow_sweep_interval` of its deadline. A paused run's wait does not time out while it is paused, and the time spent paused does not count against it: `resume!` moves the deadline by the length of the pause. A signal that arrives after the timeout fired stays pending, except under `:notify`.

**A deadline per run.** `wait_timeout:` also takes a method name or a lambda, evaluated on the workflow instance (`subject`, `arguments`) when the step parks, returning a duration or a `Time`, or `nil`:

```ruby
class OfferWorkflow < Pgbus::Workflow
  # The offer row carries expires_at, set when it was created.
  step :await_reply, wait_for: :replied, wait_timeout: -> { subject.expires_at }, on_timeout: :run do
    timed_out? ? subject.expire! : subject.accept!(received_signal.payload)
  end

  step :await_payment, wait_for: :paid, wait_timeout: :payment_deadline, on_timeout: :pause! do
    subject.record_payment!(received_signal.payload)
  end

  # start!(offer, ttl: 3600): an hour from park; without ttl, no timeout.
  def payment_deadline = arguments[:ttl]&.seconds
end
```

- **A `Time`** (or anything with `#to_time` that is not a String) is the deadline itself, however long the run took to reach the wait: the steps before it, queue latency, a reattempt. A deadline already past is `timeout_at = now`, so the next sweep applies `on_timeout`.
- **A Duration or a number of seconds** is counted from park, as a static `wait_timeout:` is, and must be positive.
- **`nil`** parks with no deadline for this attempt.
- **Anything else** raises `ArgumentError` at the gate, which, like an error the callable raises, goes to the step's exception policies (the default pauses the run): the step never parks silently.

It is evaluated once per attempt, when the step parks, under the run's row lock, so it must be fast and free of side effects, as a matcher is. It runs in a savepoint: a query of its own that fails in the database goes to the exception policies like any other error. Under `on_timeout: :run`, a step that reattempts parks again and evaluates it again: a callable that keeps returning the same past deadline times out each attempt at once. A paused run's deadline moves by the length of the pause, a `Time` one too. A static `wait_timeout:` is unchanged, and checked when the class loads.

**A soft deadline.** `on_timeout: :notify` reports a late signal and keeps waiting for it: a late notice writes a `wait_timed_out` event (`action: "notify"`), sends `pgbus.workflow.wait_timed_out` and counts once in `pgbus_workflow_wait_overdue_total`, while the same attempt stays waiting and its deadline moves one `wait_timeout:` on. No attempt fails, so the step's outcomes and the step-failing alert below are untouched, and a signal that arrives later wakes that attempt as usual (`timed_out?` is never true under `:notify`). A sweep that missed several deadlines (the dispatcher was down) reports them once and moves to the next one ahead. A per-run `wait_timeout:` is evaluated again at each notice: a `Time` after now is the next deadline, a duration moves the deadline one period on as a static one does, and `nil` or a `Time` not after now (the same `expires_at`) leaves the wait without a deadline: it was reported once and keeps waiting for its signal (the Workflows list says since when it is late, and that no further late notice is due). One that raises there, or returns anything else, pauses the run with that error: the sweep holds the run's row lock, under which no `on_exception` block may run.

**A wait without a timeout** has three escape hatches: `skip!` (the run moves on), `cancel!`, or `pause!` then `resume_at!`. One waiting longer than `config.workflow_waiting_visibility_threshold` (7 days) counts as stuck (`waiting_overdue`).

## Exceptions

An exception in a step body never reaches ActiveJob. The step row records the error (class, message, cleaned backtrace, cause chain, the worker's `hostname:pid` and the app revision), the error goes to `Pgbus::ErrorReporter`, and by default the run pauses. A policy says otherwise:

```ruby
class OrderWorkflow < Pgbus::Workflow
  on_exception PaymentGateway::Transient, action: :reattempt!, wait: 30.seconds,
                                          max_reattempts: 5, terminal_action: :cancel!
  on_exception PaymentGateway::Declined, action: :cancel!, report: :never
  on_exception action: :pause!                           # blanket: every other StandardError

  on_exception Timeout::Error do |error|                 # imperative: call a verb, or the run pauses
    error.message.include?("read") ? reattempt!(wait: 5) : skip!
  end

  step :capture, on_exception: { matching: "Faraday::TimeoutError", action: :reattempt! } do
    PaymentGateway.capture(subject)
  end
end
```

Resolution order: the step's own policies, then class-level policies that name the error, then blanket ones, then `pause!`. `matching:` takes classes, class names or a list of either. `max_reattempts:` counts consecutive reattempted attempts of the step; once spent, `terminal_action:` (default `pause!`) applies. `report:` is `:always` (default), `:never`, or `:terminal_only` (not on a reattempt that still has attempts left). A plain `rescue` in the body bypasses the policies: your code decides.

To the executor the StepJob always succeeds, whatever the step's outcome; job stats count it as completed. The step row and the `pgbus.workflow.*` events are the truth about the step.

## Idempotency

A step can run more than once: a worker that dies mid-step leaves part of the body done, and the stall action may run it again. Four things make that safe:

```ruby
class CheckoutWorkflow < Pgbus::Workflow
  step :charge do
    charge = PaymentGateway.charge(subject, idempotency_key: step_idempotency_key)  # same key on every attempt
    output(charge_id: charge.id)                                                    # durable once the step completes
  end

  step :receipt do
    ReceiptMailer.with(charge_id: output_of(:charge)[:charge_id]).receipt.deliver_later
  end
end

CheckoutWorkflow.start!(order, idempotency_key: "checkout-#{order.id}")  # a retried webhook gets the same run back
```

- **Outputs.** `output(**values)` records values on the attempt, in the same UPDATE that completes the step, so they exist exactly when the step does. `output_of(:name)` returns the latest completed attempt's output (with indifferent access), or `nil`; `outputs` returns all of them. A failed attempt keeps its output on its own row for debugging, but `output_of` never returns it. Outputs are capped like arguments; an unserialisable value raises inside the step and goes through the policies. The body's return value is ignored.
- **Step keys.** `step_idempotency_key` is `"pgbus-wf:<workflow_id>:<step>"`, the same on every attempt and on a `resume_at!` re-run: hand it to the external API so the side effect happens once. `attempt_idempotency_key` adds `:<attempt>` for the rare call that must differ per attempt.
- **Idempotent start.** `start!(…, idempotency_key:)` returns the run that already has that key, in whatever status, with `duplicate_start?` true, no new step and no new event: the first caller's `actor:` stays on the trail. The key is a reserved keyword. A different key for a subject with an ongoing run still raises `AlreadyRunning`.
- **Completed steps stay completed.** `resume!` and the policies never re-run a completed step; only `resume_at!` does, on purpose.

The engine's own messages are idempotent too: the claim (scheduled → in_progress, only once due) makes a duplicate StepJob a no-op — except a redelivery of the very message that claimed a still-running step, which applies the stall action (see Housekeeping) — and the unique `(workflow_id, step_name, attempt)` index makes a duplicate attempt impossible. A step still has to be idempotent: the key is how.

## Renaming and removing steps

A run that is scheduled on a step the class no longer declares pauses with `Pgbus::Workflow::StepNotDefined`. To retire a step, replace it with a gravestone where it used to be:

```ruby
class SignupWorkflow < Pgbus::Workflow
  step(:welcome) { … }
  removed_step :reminder      # a run still on :reminder skips it and moves on to :complete
  step(:complete, wait: 7.days) { … }
end

SignupWorkflow.removed_steps_in_flight   # => { "reminder" => 0 }: safe to delete the gravestone
```

Every run stores the SHA256 of its class's ordered step list when it starts (`definition_digest`); `Pgbus::WorkflowEntry#definition_outdated?` tells a run that began on an older definition.

## Housekeeping

The dispatcher looks after workflows on its own:

- **A dead worker is noticed in seconds**, while `config.visibility_heartbeat` is on (the default). StepJob keeps the heartbeat, so a live worker keeps its message invisible; when the very message that claimed a step comes back while the row is still in progress, its worker died (or its heartbeat lapsed, for example through a long pause), and `config.workflow_stall_action` applies at once. With the heartbeat off, only the sweep below catches a dead worker.
- **A hung step** (in progress past its `timeout:`, or past `config.workflow_stall_threshold`) gets the same stall action from the sweep, every `config.workflow_sweep_interval`. `:pause` (default) records `Pgbus::Workflow::StepStalled`, `:reattempt` schedules a new attempt, `:cancel` cancels the run.
- **A lost message** — a scheduled step of a ready run, due more than the threshold ago and not sent within it — is sent again.
- **A wait past its `wait_timeout:`** gets its `on_timeout:` (see Signals). A waiting step is never treated as stalled.
- **A lost fan-out callback** — a `fan_out` step still waiting on a batch that finished more than the threshold ago — is resolved as its `BatchFinishedJob` would have. One whose batch row is gone pauses with `FanOutFailed` (the outcome is unknown).
- **Retention**: finished and cancelled runs older than `config.workflow_retention` (30 days) are deleted hourly, with their steps and events.

A step that was declared stalled while it was in fact still running finds its row no longer in progress when it ends: its outcome and output are discarded and a `late_completion_ignored` event says so.

| Setting | Default | What it does |
| --- | --- | --- |
| `workflow_queue` | `"default"` | Queue of the StepJob messages. |
| `workflow_retention` | `30.days` | Finished/cancelled runs kept this long; nil keeps them. |
| `workflow_sweep_interval` | `5.minutes` | How often the sweep runs. |
| `workflow_stall_threshold` | `1.hour` | When an in-progress step is stalled, or a scheduled one overdue. |
| `workflow_stall_action` | `:pause` | `:pause`, `:reattempt` or `:cancel`. |
| `workflow_max_payload_size` | `131_072` | Byte cap on arguments and on each step output. |
| `workflow_error_backtrace_lines` | `50` | Backtrace lines kept on a failed step. |
| `workflow_waiting_visibility_threshold` | `7.days` | A wait without a deadline (no `wait_timeout:`, or a per-run one that set none) this long counts as stuck (`waiting_overdue`). |

## Operating

The dashboard's **Workflows** page (`/pgbus/workflows`) lists every run, newest first: the ongoing ones by default, or one state, or all. Each row says in words what the run is doing and what happens next: "Performing `capture` (attempt 2)", "Waiting for signal `shipped` since 3h ago", "Waiting for batch `OrderWorkflow#notify_vendors` (7 of 12 jobs done)", "Paused on `capture`: Timeout::Error — resume or cancel", "Scheduled: `reminder` runs in 2d". Filter by class, current step, error class, subject, workflow id and a created-at range; the filters combine and survive paging.

A run's page shows its arguments, a **timeline** (one range bar per attempt, split into waited, queued and ran, coloured by how it ended, with waits and pauses as their own segments; zoom for runs that span days), its newest 100 attempts with the same durations as text, its worker, app revision, error, backtrace and cause chain, the signals it received, and the audit trail ("Resumed by dashboard:alice: SMTP fixed"). A run that started on an older step list says so.

**Actions** show only when `available_actions` allows them: Resume (its label says whether it retries the failed step or continues with the next), Resume at a chosen step, Run now (a scheduled step's wait cut short), Skip step (a waiting step), Send signal (a JSON object; the form's idempotency key makes a double submit send one signal), Rerun (finished or cancelled runs; the new run links back) and Cancel. Each takes an optional reason and reports the real outcome: when the run moved on first, the flash says where it is now.

**Needs attention** (`/pgbus/workflows/attention`) groups every stuck run by class, step, stuck reason and error class, with a count, when the group was first and last seen and the newest error message (the 200 biggest groups; past that the page is not exhaustive). Once the cause is fixed, Resume or Cancel the whole group (at most 500 runs per press, each through its own guarded transition): "Resumed 212; 2 had already moved on". The home page counts ongoing, paused and stuck runs.

Two settings connect the page to your app:

```ruby
Pgbus.configure do |config|
  # Who a dashboard action is recorded as on the audit trail:
  # "dashboard:<name>", or "dashboard" when this returns nil.
  config.web_actor = ->(request) { request.env["warden"]&.user&.email }

  # Where the subject links to. Only an http(s) URL or a path on this
  # site is linked; anything else, or a lambda that raises, is text.
  config.workflow_subject_url = ->(type, id) { "/admin/#{type.underscore.pluralize}/#{id}" }
end
```

The MCP server's read tools have the same views: `pgbus_workflows` (the list, with the page's filters), `pgbus_workflow` (one run, with `available_actions`) and `pgbus_workflows_attention`. Arguments, step outputs, signal payloads and error text are redacted unless the server allows payloads and the call sets `include_payloads`.

**Acting over MCP** is a separate opt-in. With the write tools on, an agent that found the paused runs can resume them after the fix:

```ruby
Pgbus.configure do |config|
  # Default false: the write tools do not exist at all. Read when the
  # MCP server is built (pgbus mcp, Pgbus::MCP.rack_app).
  config.mcp_write_tools = true
end
```

That adds `pgbus_workflow_resume`, `pgbus_workflow_cancel`, `pgbus_workflow_skip_step`, `pgbus_workflow_resume_at` (with `step`), `pgbus_workflow_run_now` and `pgbus_workflow_signal` (with `name`, a JSON object `payload` of at most `workflow_max_payload_size` and an optional `idempotency_key`, which a retried call should pass; a blank one is no key). Each takes `workflow_id`, a required `reason` (1–500 characters after trimming whitespace) and an optional `expected_status`, the status the agent read: a run that has moved on by the time the call starts answers `ok: false` and nothing is done (the check runs just before the action, whose transition is guarded again under the run's row lock). A verb the run's `available_actions` does not allow answers `ok: false` with the allowed list. Otherwise the call goes through the same action as the dashboard's button and lands on the audit trail with actor `mcp` (`mcp:<client name>` when the transport knows the client) and the reason it was given; a refused call changes nothing and records nothing. Every call answers `{ ok:, reason:, workflow: }` with the run as it is after the call (payloads always redacted; a transition refused under the row lock adds `code` and the `status` the run is in now; an unexpected failure answers a fixed reason, the error is in the pgbus log). They are marked `destructiveHint: true` (`idempotentHint: true` on cancel only), and there are no bulk tools: `pgbus_workflows_attention` groups the stuck runs, `pgbus_workflows` lists a group's runs (`status`, `workflow_class`, `step`, `error_class`), and each run gets one audited call.

Enable them only on an authenticated endpoint: `Pgbus::MCP.rack_app` with `token:` or `auth:`, or mounted inside your app's authenticated routes, never public. An unauthenticated `rack_app` with the write tools on logs a warning saying anyone who reaches it can change workflow runs.

## Monitoring

**Workflow insights** (`/pgbus/workflows/definitions`, linked from the Workflows page as "Insights by class") answers how a workflow class is doing, not where one run is. The list has one row per class (every registered `Pgbus::Workflow` subclass and every class name still in `pgbus_workflows`): runs going on now, paused, needing attention, the runs that finished or were cancelled in the window, and the p50/p95 end-to-end duration (created → finished). A class that no longer loads says "definition not loaded"; one with ongoing runs on a `removed_step` says how many.

A class's page (`/pgbus/workflows/definitions/OrderWorkflow`, or `Billing::OrderWorkflow`) shows its steps in order as a **pipeline**: each step with the runs sitting at it now (scheduled, running, waiting, paused), the p50/p95/p99 run time and the pickup latency (scheduled → started) of its attempts that finished in the window, and their failure (an error was recorded), reattempt and skip rates. A step that waits for a signal also shows its wait timeout: its `wait_timeout:`, or "dynamic (per run)" for a method name or a lambda. Percentiles over fewer than 20 attempts carry their sample size and a "few samples" note. Below: runs started, finished and cancelled per bucket (5 min for 1 h, 1 h for 24 h, 6 h for 7 d, 1 d for 30 d), the end-to-end duration distribution, a **funnel** (of the runs started in the window, how many reached each step and where each one is now) and the same numbers for the last five app revisions that finished an attempt, newest first, so a regression after a deploy shows as a row (a failure rate more than five points above the revision before is marked). The window is 1 h, 24 h (default), 7 d or 30 d, capped at `workflow_retention`. Every count of runs a list filter can express (runs at a step now, by status, started or ended in the window, where the funnel's runs are now) links to the Workflows list filtered to exactly those runs; attempt-level numbers (percentiles, rates, the funnel's "reached", the per-revision rows) are text. The list gained `ended_from`/`ended_to` (when a run finished or was cancelled) for it, and the need-attention counts open Needs attention narrowed to the class (`/pgbus/workflows/attention?workflow_class=OrderWorkflow`).

The numbers come from the step and run rows, on the indexes the tables already have: on about 1 M step rows over 20 classes a class's page reads in about 25 ms and the list in about 100 ms over 24 hours (the budgets `rake bench:workflow_insights` checks: 250 and 500 ms); over 30 days both take about 300 ms, which the bench reports without a budget. The MCP server's read tools have the same: `pgbus_workflow_definitions(window:)` and `pgbus_workflow_stats(workflow_class:, window:)`, and `pgbus_workflows_attention` takes `workflow_class:`.

**Alerts** belong in Prometheus or AppSignal, on the metrics below. Three starting points: a step whose attempts end in a reattempt or a pause (an exception, or the body's own `reattempt!`/`pause!`) more than 5 % of the time for 15 minutes, runs needing a person for an hour, and a `:notify` wait that missed its deadline:

```yaml
groups:
  - name: pgbus-workflows
    rules:
      - alert: PgbusWorkflowStepFailing
        expr: |
          sum by (workflow_class, step) (rate(pgbus_workflow_steps_total{outcome=~"reattempted|paused"}[15m]))
            / sum by (workflow_class, step) (rate(pgbus_workflow_steps_total[15m])) > 0.05
        for: 15m
        annotations:
          summary: "{{ $labels.workflow_class }}#{{ $labels.step }} fails on {{ $value | humanizePercentage }} of its attempts"
      - alert: PgbusWorkflowsNeedAttention
        expr: sum by (workflow_class) (pgbus_workflows_stuck) > 0
        for: 1h
        annotations:
          summary: "{{ $value }} {{ $labels.workflow_class }} runs need a person: /pgbus/workflows/attention"
      - alert: PgbusWorkflowWaitLate
        expr: sum by (workflow_class, step) (increase(pgbus_workflow_wait_overdue_total[15m])) > 0
        annotations:
          summary: "{{ $labels.workflow_class }}#{{ $labels.step }} is still waiting past its deadline"
```

On AppSignal, the bundled `main` dashboard (`pgbus dashboard main`) has a workflows row: steps per outcome, step run time, workflows by status, stuck workflows and late waits.

## Observability

**Audit trail.** Every transition and every outside verb writes a `pgbus_workflow_events` row in the same transaction, with its actor (`"app"` by default, `"system"` for the StepJob and the sweep, or the `actor:` you pass) and reason. `wf.events` returns them in order: `started`, `step_scheduled`, `step_started`, `step_completed`, `step_failed`, `step_skipped`, `step_reattempted`, `paused`, `resumed`, `cancelled`, `finished`, `run_now`, `rescheduled`, `rerun`, `stall_detected`, `late_completion_ignored`, and for waiting steps `step_waiting` (details: `wait_for` or `batch_id`), `signal_received` (`name`, `idempotency_key`, `duplicate`), `signal_claimed`, `wait_timed_out` (`wait_for`, `action`, the missed `timeout_at`), `fan_out_opened` (`batch_id`, `size`) and `fan_out_finished` (`failed_jobs`, `total_jobs`). `start!` and `signal!` take `actor:` too.

**Metrics** reach Prometheus, StatsD and AppSignal with the same names (one exception for StatsD, below the table):

| Metric | Kind | Labels |
| --- | --- | --- |
| `pgbus_workflow_step_duration_seconds` | histogram | workflow_class, step, outcome |
| `pgbus_workflow_steps_total` | counter | workflow_class, step, outcome |
| `pgbus_workflows_finished_total` | counter | workflow_class |
| `pgbus_workflows_paused_total` | counter | workflow_class, error_class |
| `pgbus_workflows_cancelled_total` | counter | workflow_class |
| `pgbus_workflow_stalls_total` | counter | workflow_class, step, detection |
| `pgbus_workflow_wait_overdue_total` | counter | workflow_class, step |
| `pgbus_workflows` | gauge | workflow_class, status |
| `pgbus_workflows_stuck` | gauge | workflow_class, reason |

StatsD reads timer values as milliseconds, so it receives the step histogram as `pgbus_workflow_step_duration_ms` (seconds × 1000). `pgbus_workflow_wait_overdue_total` counts the late notices of `on_timeout: :notify` waits (one per deadline, or one for a sweep that missed several); the other `on_timeout:` values end the wait and show in the step and run counters instead.

**Stuck reasons** (the `pgbus_workflows_stuck` gauge): `paused_error` (paused on a step that failed), `stalled` (paused on `StepStalled`), `paused_manual` (the step called `pause!`, or an outside `pause!`), `overdue` (a scheduled step nobody picked up), `waiting_overdue` (a wait without a deadline older than `workflow_waiting_visibility_threshold`: no `wait_timeout:`, or a per-run one that set none or cleared it after a `:notify` notice; a `:notify` wait with a deadline has `pgbus_workflow_wait_overdue_total` instead).

**Errors** reach `Pgbus::ErrorReporter` with `{ workflow_class:, workflow_id:, step_name:, attempt:, outcome: }` as context. **Logs**: while a step runs, Pgbus's own log lines carry `workflow=… step=… attempt=…` through `Pgbus::LogFormatter`, and your code reads `Pgbus::Workflow::StepContext.workflow_id` (and `step_name`, `attempt`) to tag its own. The instrumentation events are `pgbus.workflow.step`, `pgbus.workflow_finished`, `pgbus.workflow_paused`, `pgbus.workflow_cancelled`, `pgbus.workflow.stall_detected`, `pgbus.workflow.signal`, `pgbus.workflow.wait_timed_out` and `pgbus.workflow.gauge`.

## Testing

`Pgbus::Workflow::TestHelpers` runs steps inline against your test database, ignoring their `wait:`, through the real StepJob:

```ruby
RSpec.describe SignupWorkflow do
  include Pgbus::Workflow::TestHelpers

  it "onboards the user" do
    wf = SignupWorkflow.start!(user)

    perform_next_step(wf)            # runs :welcome
    run_step_inline(wf, :reminder)   # raises if another step is next
    speedrun_workflow(wf)            # runs steps until the run stops being ready

    expect(wf.status).to eq("finished")
  end

  it "ships the order" do
    wf = OrderWorkflow.start!(order)
    speedrun_workflow(wf)            # raises StepWaiting at notify_vendors
  rescue Pgbus::Workflow::StepWaiting
    complete_fan_out(wf)             # every member completes; the run moves on
    deliver_signal(wf, :shipped, payload: { at: Time.current })  # signal!, then the step it woke
    expire_wait(wf)                  # or: the waiting step's on_timeout, now
  end
end
```

`speedrun_workflow` stops at `paused`, `finished` and `cancelled`, and raises `Pgbus::Workflow::StepWaiting` on `waiting`, naming the signal or the batch the step waits for. `complete_fan_out` completes every member of the waiting step's batch (without running them) and runs its `BatchFinishedJob` inline; `deliver_signal` calls `signal!` and runs the step it scheduled; `expire_wait` applies the waiting step's `on_timeout` whatever its deadline (under `:notify` the step stays waiting: assert the `wait_timed_out` event and the moved `timeout_at`).

## How it works

- `pgbus_workflows` — one row per run (`Pgbus::WorkflowEntry`): the class, the subject, the arguments, the status and the current step. A partial unique index keeps one ongoing run per class and subject; another dedupes `idempotency_key`s.
- `pgbus_workflow_steps` — one row per step attempt (`Pgbus::WorkflowStep`), unique on `(workflow_id, step_name, attempt)`, with the timings, the worker, the app revision, the output and the error.
- `pgbus_workflow_events` — the append-only audit trail (`Pgbus::WorkflowEvent`).
- `pgbus_workflow_signals` — one row per signal (`Pgbus::WorkflowSignal`), unique on `(workflow_id, idempotency_key)`.

Every StepJob is sent by `WorkflowStep`'s after-commit callback, and only there: a new scheduled row, a row going back to scheduled, or a bumped `enqueued_at` (how the sweep and `resume!` re-send). A StepJob claims its row (scheduled → in_progress, only once due) and the run (ready → performing) under the run's row lock, runs the guards and the body, then applies the outcome under the lock again: one step at a time per run, every change a guarded UPDATE from the table below. A waiting step goes back to scheduled when it is woken (a signal, or a timeout with `:run`), so its resume is a StepJob like any other; a fan-out's batch, a timeout or an operator `skip!` resolves the waiting row directly.

## The transition table

Pgbus::Workflow::States is the one rule for which verb is legal in which state, for the runner, `available_actions` and this page. The outside verbs:

| Verb | From | To | Steps |
| --- | --- | --- | --- |
| `pause` | ready, performing, waiting | paused | a running step finishes and records its outcome; nothing new is claimed |
| `resume` | paused | ready | a failed step runs again as a new attempt; a completed or skipped one is followed by the next step; a scheduled one is sent again; a waiting one waits again |
| `resume_at` | paused | ready | the named step runs as a new attempt |
| `cancel` | ready, performing, waiting, paused | cancelled | scheduled and in_progress steps are cancelled; a running step's outcome is ignored |
| `run_now` | ready | ready | the scheduled step's wait is cut short and it is sent now |
| `reschedule` | ready | ready | the scheduled step moves to the given time |
| `rerun` | finished, cancelled | ready | a new run starts with the same subject and arguments |
| `skip` | waiting | ready | the waiting step is skipped (a signal it holds is consumed) and the next step is scheduled, or the run finishes after the last |
| `signal` | ready, performing, waiting, paused | (unchanged) | the signal is stored; a waiting step it matches is scheduled again, otherwise it stays pending until a step claims it |

The StepJob's and the sweep's own moves, never offered as actions:

| Verb | From | To | Steps |
| --- | --- | --- | --- |
| `claim` | ready | performing | the scheduled step moves to in_progress |
| `advance` | performing | ready | the step is completed or skipped and the next step is scheduled |
| `finish` | performing, paused | finished | the last step (or one that called finished!) completed |
| `reattempt` | performing | ready | the step failed and its next attempt is scheduled |
| `step_pause` | performing | paused | the step completed after pause!, or failed (an exception or a stall) |
| `step_cancel` | performing, paused | cancelled | the step called cancel!, or cancel_if or a policy cancelled it |
| `unpause` | paused | performing | resume! while the step that was running at the outside pause! is still running: the run goes on performing it |
| `hold` | paused | paused | an outcome that arrived after an outside pause!: recorded on the step; a completed, skipped or reattempted step has its next row scheduled; the workflow stays paused |
| `park` | performing | waiting | the step waits for a matching signal or for its fan-out batch; nothing is sent until it is woken |
| `wake` | waiting | ready | a matching signal arrived, or the wait timed out with on_timeout: :run: the waiting step is scheduled again |
| `resolve` | waiting | performing | the fan-out batch finished, the wait timed out, or an operator skipped the step: its outcome is applied as for a finished attempt |
| `rewait` | paused | waiting | resume! on a run paused while its step was waiting: it waits again |

Workflow states: `ready`, `performing`, `waiting`, `paused` (the four ongoing ones), `finished`, `cancelled`. Step states: `scheduled`, `waiting`, `in_progress`, `completed`, `failed`, `cancelled`, `skipped`. A run is `waiting` while its step waits for a signal or a fan-out batch.