@spec assert_replays(module(), term(), [map()]) :: term()

Assert that a workflow replays from history without drift.

Returns the replayed result.

@spec assert_replays(module(), term(), [map()], term()) :: term()

Assert that a workflow replays from history to expected.

@spec checkout_sandbox(
  module() | nil,
  keyword()
) :: :ok

Check out an Ecto SQL Sandbox connection.

Pass shared: true when workflow engines or workers need to use the test process' checked-out connection.

@spec crash!(
  binary(),
  keyword()
) :: :ok | {:error, :no_engine}

Kill a run's engine process the way a node loss would.

Returns once the engine is dead and has left the instance registry, so the next drive/2 starts a genuinely fresh engine rather than racing the old one. Pair with expire_lease!/2: a killed engine's lease is still valid until it expires, and rescuing it earlier would be lease theft.

Drive the run to a resting state first (drive_until_state/3). Killing an engine that is mid-statement is fine in production but takes the SQL Sandbox shared connection down with it, which fails the rest of the test for a reason that has nothing to do with the workflow.

@spec drive(
  binary(),
  keyword()
) :: {:ok, map()} | {:error, term()}

Drive the durable runtime until run_id reaches a terminal state.

Test suites normally start Continuum with its pollers disabled so nothing moves behind a test's back. This turns the crank instead: on each tick it rescues expired leases, dispatches runnable runs, runs due activity tasks, and fires due timers, until the run completes, fails, is cancelled, or the deadline passes.

Returns whatever Continuum.await/3 returns for the finished run.

{:ok, run_id} = Continuum.Test.start_postgres(Checkout, order)
assert {:ok, %{state: :completed}} = Continuum.Test.drive(run_id)

Options: :instance, :timeout (default 5_000 ms), :batch_size (default 10).

@spec drive_until_state(binary(), [atom()], keyword()) :: :ok | {:error, :timeout}

Drive the durable runtime until run_id is in one of states.

The same crank as drive/2, stopped earlier. Useful for getting a run to the point you want to crash it at.

Continuum.Test.drive_until_state(run_id, [:suspended])

@spec dump_history!(binary(), Path.t(), keyword()) :: :ok

Persist a run history to path as an Erlang external term.

The resulting file is intended for golden-history tests committed to the repository.

@spec elapse_timers!(
  binary(),
  keyword()
) :: :ok

Make every unfired timer on a run due now.

A multi-day timer/1 is the thing you least want to wait for in a test; this moves its fires_at into the past so the next drive/2 fires it.

@spec expire_lease!(
  binary(),
  keyword()
) :: :ok

Expire a run's lease so recovery may rescue it.

Recovery deliberately refuses to touch a run whose lease is still live, so a crash-resume test has to move the clock rather than wait out a real TTL.

@spec fire_timer(
  binary(),
  keyword()
) :: :ok | {:error, term()}

Inject a fired timer event for the latest pending timer in a run's history.

@spec history(
  binary(),
  keyword()
) :: [map()]

Load a run's event history from a journal.

:journal accepts the shorthand :postgres or :in_memory as well as an adapter module, so a test never has to name a Continuum.Runtime.* module. Defaults to the in-memory journal.

@spec inject_signal(binary(), atom(), term(), keyword()) :: :ok | {:error, term()}

Inject a signal into a run and wake its local engine when one exists.

Delivery goes through the same Continuum.Runtime.SignalRouter path as Continuum.signal/4: in-memory signals are buffered in the run's mailbox and consumed by the matching await signal, journaling signal_received with the await's command identity — injected signals exercise the same command-identity drift detection as production deliveries.

@spec load_history!(Path.t()) :: [map()]

Load a history previously written by dump_history!/3.

@spec replay(module(), term(), [map()], keyword()) :: replay_result()

Replay a workflow from an existing history.

Delegates to Continuum.Replay.run/4, which is read-only by default: stepping past the journaled tail reports {:suspended, {:history_exhausted, _}} instead of executing the next activity for real. Pass journal: explicitly to replay through a live adapter.

Returns {:ok, result} when the workflow completes from history, or {:suspended, reason} if the history ends at a pending effect.

@spec reset_in_memory!() :: :ok

Reset the in-memory journal.

@spec start_postgres(module(), term(), keyword()) ::
  {:ok, binary()} | {:error, term()}

Start a workflow run against the Postgres journal.

@spec start_synchronous(module(), term(), keyword()) ::
  {:ok, binary()} | {:error, term()}

Start a workflow run synchronously against the in-memory journal.

Activities run inline in the engine process — no worker pool, no retry, no timeout. Child workflows run inline too, as their own in-memory engines.

Stubbing activities

Pass :activities to stand in for activity bodies, so a unit test can drive a workflow's branches without the activity's real dependencies:

Continuum.Test.start_synchronous(Checkout, order,
  activities: %{
    {Payments, :charge} => fn _order -> {:ok, "ch_test"} end,
    {Shipping, :book} => {:error, :out_of_stock}
  }
)

Keys are {Module, :function}, or {Module, :function, arity} when one module exports the same activity name at several arities; the more specific key wins. A value that is a function of the activity's arity is called with the activity's arguments, and any other value is returned as-is.

Stub returns are validated with Continuum.DurableTerm, because in-memory writes otherwise skip that check — a stub returning a PID would pass the unit test and be rejected in production.

Stubs are refused on the Postgres journal: a durable activity runs in a worker process out of a claimed task row, which a stub cannot reach. They also cannot influence command identity, which is computed at macro expansion from the call site, so a stubbed run journals the same command ids as a real one.