Determinism and Replay Safety

Replay re-runs a workflow body from the top, skipping steps that already wrote step:done. For that to give the right answer, the body has to be deterministic: a step that finished before a crash must produce the same result when it runs again.

The rule is short. Anything non-deterministic or side-effecting goes through ctx. ctx records the result the first time and replays it after.

Call	On replay
ctx.step(id, fn, ...)	Returns the recorded result after step:done exists.
ctx.now()	Recorded timestamp.
ctx.uuid()	Recorded UUID.
ctx.random()	Recorded float in [0.0, 1.0).
ctx.sleep() / ctx.sleep_until()	Skips the wait.
ctx.wait_for_signal()	Returns the recorded signal.
ctx.interrupt()	Returns the recorded response.
ctx.emit()	Replays from the stream. Not re-emitted.

What breaks

Reaching past ctx for time, randomness, or IO diverges on replay:

import time, uuid, random

@workflow
async def unsafe(ctx):
    now = time.time()        # different every run
    uid = uuid.uuid4()       # different every run
    data = await fetch(url)  # different response on replay

Use the ctx versions instead:

@workflow
async def safe(ctx):
    now = ctx.now()
    uid = ctx.uuid()
    data = await ctx.step("fetch", fetch, url)   # runs once, replays the result

The engine has a backstop here: it inspects a step’s bytecode and rejects a direct time, random, uuid, or datetime call inside ctx.step. Treat it as a seatbelt, not a guarantee. It only catches direct module.attr() calls, so an aliased import or a call buried in a helper slips through. Keeping non-determinism behind ctx is still on you.

Idempotent side effects

A step can be interrupted after its side effect lands but before step:done is written. On recovery it can run again because the durable completion was never recorded. ctx.step prevents re-running effects whose completion is already in the event stream, but externally visible effects still need an idempotency key or another natural dedupe guard:

@workflow
async def charge(ctx, amount_cents: int):
    return await ctx.step(
        "charge",
        lambda: payment_client.charge(
            amount_cents,
            idempotency_key=f"{ctx.run_id}:charge",
        ),
    )

For effects you can’t make idempotent, put a human decision in front of them with ctx.interrupt() and then route the approved effect through an external system that can dedupe it. See Human-in-the-Loop.

Version skew

Replay assumes the code matches the events. If you change a workflow’s shape while a run is mid-flight and a worker replays it against the new code, the engine detects the divergence and fails the run loudly rather than producing a wrong result.

Safe to change during a rolling deploy:

• A step’s implementation, as long as its id and the order of steps stay the same.
• Adding new workflows or steps that in-flight runs don’t reach.

Not safe while runs are in flight:

• Renaming or reordering steps, which moves their replay keys.
• Removing a step an in-flight run already recorded.

When in doubt, drain in-flight runs before shipping a change to a workflow’s structure.

Deploy-safe versioning

For a structural change you can’t make backward-compatible — a new step, a reordered branch, a behavior change mid-workflow — gate it behind ctx.patched instead of just shipping it:

@workflow
async def charge(ctx, amount_cents: int):
    await ctx.step("charge", lambda: payment_client.charge(amount_cents))
    if await ctx.patched("refund-flow"):
        await ctx.step("refund-hold", lambda: payment_client.hold_refund_window())
    return "done"

ctx.patched(change_id) decides old-branch-vs-new-branch off the replay cursor, not a run-wide flag: a run that already passed this point without seeing the patch takes the old branch on replay (deterministic); a run that reaches this point live (cursor exhausted) takes the new branch and records a marker. That’s what makes the new code safe to deploy while old runs are still in flight — but it’s still worth proving before you ship, not just trusting the mechanism.

pyrula.workflows.testing gives you a small replay-CI harness for exactly that:

from pyrula.workflows.testing import assert_replay_compatible, export_run_history

# 1. Pull a real in-flight run's journal from your live store (or Postgres archive).
history = export_run_history(store, "charge", run_id)

# 2. Assert the *currently deployed* workflow code can still replay it. Run this
#    in CI against your candidate deploy, before it ships.
assert_replay_compatible(charge, history)  # raises AssertionError if it can't

assert_replay_compatible shadow-replays the history against the workflow code registered under that name and raises with the divergence reason if the run would break — the same check verify_replay_compatible does internally, wrapped for a CI assertion. export_run_history returns JSON-serializable dicts, so you can also commit a run’s history as a fixture in your repo rather than pulling it live.

The full loop:

1. Guard the change with if await ctx.patched("refund-flow"): ....
2. Capture history for a real (or representative) in-flight run with export_run_history.
3. Run assert_replay_compatible(workflow, history) in CI. Deploy only if it’s green.
4. Once every old run has drained past the patch point, call await ctx.deprecate_patch("refund-flow") at the same call site — it keeps writing the marker (so replay of any straggler is still safe) but signals the guard is scheduled for removal.
5. Once you’re confident no run still needs the old branch, delete the else branch and the patch guard entirely.