Testing workflows

#makeWorkflowRunner

A workflow's logic — steps, retries, branching on a signal — deserves a test that doesn't boot the cluster engine. makeWorkflowRunner({ ctx, workflows }) runs a workflow end-to-end in-process on an in-memory engine and hands back the same step log the workflow dashboard reads, so a test can assert on attempts, durations, and intermediate values.

Discovery is explicit: pass each workflow under test as a { workflow, execute } entry. The workflow is the value from workflow({ name, ... }) (the descriptor); the execute is the same (payload, executionId) => Effect the framework hands to workflow.toLayer(...) at registration time — i.e. buildExecute(ctx) from your *.workflow.server.tsx. The runner matches start(name, …) against entry.workflow.name.

import { makeTestContext, mockStore, makeWorkflowRunner } from '@voltro/testing'
import { SummarizeTodos } from '../workflows/todos.summarize.workflow'
import buildExecute from '../workflows/todos.summarize.workflow.server'

test('summarize walks fetch → summarize and succeeds', async () => {
  const ctx = makeTestContext({ store: mockStore({ todos: [{ id: 't1', title: 'x' }] }) })

  const runner = makeWorkflowRunner({
    ctx,
    workflows: [
      { workflow: SummarizeTodos as never, execute: buildExecute(ctx) as never },
    ],
  })

  const result = await runner.start('todos.summarize', { tenantId: 't1' })

  expect(result.status).toBe('succeeded')
  expect(result.steps.map((s) => s.name)).toEqual(['fetch-todos', 'summarize'])
})

The as never casts keep the call-site ergonomic — the runner is type-erased over each workflow's payload / success / error shapes (it provides the engine layer itself), and never assigns to any field type without an any leaking into the public type.

#The result

start(name, payload) blocks until the run is terminal and resolves a WorkflowRunResult:

Field	Meaning
status	'succeeded' or 'failed'.
output	The workflow's success value (null on failure).
error	{ tag, message } on failure (null on success). tag is the typed error's _tag when there is one.
steps	One InspectedStep per step name — { name, attempt, attempts, status, input, output, error, durationMs }.
runId	The synthetic run id, for inspect(...).

#Asserting on retries

steps[i].attempts reflects the real retry count: a step that throws twice then succeeds records three attempt rows, so attempts === 3. This is the cleanest way to prove a step's retry policy actually fired.

const charge = result.steps.find((s) => s.name === 'charge-card')
expect(charge?.status).toBe('completed')
expect(charge?.attempts).toBe(3)   // failed twice, succeeded on the third

#Inspecting a finished run

inspect(runId) returns the recorded InspectedWorkflow for a previously-started run without re-running it, or null for an unknown id — the same shape the dashboard renders ({ id, name, status, input, output, subject, source: 'test-runner', steps, startedAt, finishedAt, … }).

const run = await runner.inspect(result.runId)
expect(run?.source).toBe('test-runner')

#Notes

• The runner uses the in-memory workflow engine and an in-memory recorder — nothing persists, nothing schedules on the real clock. Pair it with the mock clock for any time-dependent step.
• Starting a workflow the runner doesn't know about throws a clear error naming the missing workflow — register it in the workflows array.
• This tests the workflow's logic. To exercise the durable engine itself (crash/resume, cluster handoff), that's an integration concern beyond the in-process runner.