Delayed Transitions

Timer-based auto-transitions with after — timeouts, polling, and auto-progression.

31 min read

Delayed transitions let a state automatically transition after a specified time delay. No event needed — the machine moves on its own when the timer expires. This is perfect for session timeouts, polling loops, auto-save debouncing, and any workflow that involves waiting.

⏱️ What Are Delayed Transitions?

In a normal state machine, transitions only fire when an event arrives. Delayed transitions break that rule: they fire after a timer expires. You configure them with the after property on a state, mapping millisecond delays to target states.

stateDiagram-v2
    direction LR
    [*] --> active
    active --> warning : after 5 min
    warning --> expired : after 30 s
    warning --> active : EXTEND
    active --> active : ACTIVITY (reenter: timer resets)

When the machine enters a state with after timers, those timers start immediately. If the machine leaves that state before a timer fires (because of an event), the timer is cancelled automatically.

📄 JSON After Syntax

The after property is a dictionary mapping millisecond delays (as strings in JSON) to transition targets:

{
  "after": {
    "3000": "nextState"
  }
}

After 3000 ms (3 seconds) in this state, the machine transitions to "nextState".

⌛ Basic Example: Session Timeout

A session that warns the user before expiring:

{
  "id": "sessionTimeout",
  "initial": "active",
  "states": {
    "active": {
      "after": {
        "300000": "warning"
      },
      "on": {
        "ACTIVITY": {
          "target": "active",
          "reenter": true
        }
      }
    },
    "warning": {
      "after": {
        "30000": "expired"
      },
      "on": {
        "EXTEND": "active"
      }
    },
    "expired": {
      "type": "final"
    }
  }
}

Running it:

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "sessionTimeout",
    "initial": "active",
    "states": {
        "active": {
            "after": {
                "300000": "warning",
            },
            "on": {
                "ACTIVITY": {
                    "target": "active",
                    "reenter": True,
                },
            },
        },
        "warning": {
            "after": {
                "30000": "expired",
            },
            "on": {
                "EXTEND": "active",
            },
        },
        "expired": {
            "type": "final",
        },
    },
}

machine = create_machine(config)
interp = SyncInterpreter(machine).start()

print(interp.active_state_ids)
# {'sessionTimeout.active'}

# User does something — reenter: true restarts the timer
interp.send("ACTIVITY")
print(interp.active_state_ids)
# {'sessionTimeout.active'}  (timer restarted)

interp.stop()

This machine:

Starts in active
After 5 minutes of inactivity, moves to warning
The user gets 30 seconds to click “EXTEND” or it moves to expired
Any ACTIVITY event re-enters active ("reenter": true), resetting the 5-minute timer. Without reenter, a self-transition leaves the state active and the timer keeps its original schedule (XState v5)

🎬 After with Actions

Delayed transitions can trigger actions, just like event-driven transitions:

{
  "monitoring": {
    "after": {
      "5000": {
        "target": "next",
        "actions": "logTimeout"
      }
    }
  }
}

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "timedAction",
    "initial": "waiting",
    "states": {
        "waiting": {
            "after": {
                "5000": {
                    "target": "done",
                    "actions": "logTimeout",
                },
            },
        },
        "done": {
            "type": "final",
        },
    },
}


class TimedLogic(MachineLogic):
    def log_timeout(self, interpreter, context, event, action_def):
        print("Timer expired — transitioning to done")


machine = create_machine(config, logic=TimedLogic())
interp = SyncInterpreter(machine).start()
# After 5 seconds, "log_timeout" fires and machine moves to "done"
interp.stop()

🛡️ After with Guards

You can conditionally block a delayed transition using a guard:

{
  "monitoring": {
    "after": {
      "60000": {
        "target": "stale",
        "guard": "noRecentData"
      }
    }
  }
}

If the guard noRecentData returns False when the timer fires, the transition is skipped — the machine stays in monitoring.

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "guardedTimer",
    "initial": "monitoring",
    "context": {
        "lastPing": 0,
    },
    "states": {
        "monitoring": {
            "after": {
                "60000": {
                    "target": "stale",
                    "guard": "noRecentData",
                },
            },
            "on": {
                "PING": {
                    "actions": "recordPing",
                },
            },
        },
        "stale": {
            "on": {
                "RESET": "monitoring",
            },
        },
    },
}


class MonitorLogic(MachineLogic):
    def no_recent_data(self, context, event):
        """Returns True if no PING has ever been received (lastPing is still 0)."""
        return context.get("lastPing", 0) == 0

    def record_ping(self, interpreter, context, event, action_def):
        import time

        context["lastPing"] = time.time()


machine = create_machine(config, logic=MonitorLogic())
interp = SyncInterpreter(machine).start()

# If PING arrives before 60s, lastPing becomes non-zero
# and the guard returns False, preventing the stale transition
interp.send("PING")
print(interp.active_state_ids)
# {'guardedTimer.monitoring'}

interp.stop()

Note: Guard functions receive (context, event) and must return a bool. They must be synchronous.

⏲️ Multiple Timers per State

A single state can have multiple after timers running simultaneously:

{
  "monitoring": {
    "after": {
      "5000": {
        "target": "monitoring",
        "reenter": true,
        "actions": "heartbeat"
      },
      "60000": {
        "target": "stale",
        "guard": "noRecentData"
      },
      "300000": "timeout"
    }
  }
}

All three timers start when monitoring is entered:

5 seconds: self-transition with heartbeat action ("reenter": true re-enters the state, restarting all timers)
60 seconds: conditional move to stale (only if guard passes)
5 minutes: unconditional move to timeout

Tip: The shortest timer fires first. With reenter: true the 5-second heartbeat re-enters the state and all its timers reset — so the 60s and 300s timers effectively restart too.

🔄 Timer Reset

stateDiagram-v2
    direction LR
    [*] --> active
    active --> active : ACTIVITY (reenter: true → timer restarts)
    active --> idle : after 60 s
    idle --> active : ACTIVITY

A transition that exits and re-enters a state cancels and restarts its after timers. A self-transition does so only with "reenter": true; without it the state stays active and its timers keep running (XState v5). This is key to implementing patterns like “idle timeout” — every user action resets the clock.

{
  "active": {
    "after": {
      "300000": "warning"
    },
    "on": {
      "ACTIVITY": {
        "target": "active",
        "reenter": true
      }
    }
  }
}

Every ACTIVITY event re-enters active (because of reenter: true), which cancels the existing 5-minute timer and starts a fresh one.

Warning: A self-transition without reenter does nothing to the state: no exit or entry actions run and no timer is cancelled or restarted. With "reenter": true it runs exit actions, cancels the timers, runs entry actions and restarts the timers. This is XState v5 behaviour.

🏷️ Named and Computed Delays

An after key does not have to be a number. A non-numeric key is a named delay resolved from MachineLogic(delays={...}) when the state is entered; the same names work as the delay of a raise / sendTo / sendParent action. A delay is a number of milliseconds or a callable that returns one. A callable receives a single {"context": ..., "event": ...} mapping (the legacy (context, event) two-argument form is also accepted).

from xstate_statemachine import (
    create_machine,
    SyncInterpreter,
    MachineLogic,
    SimulatedClock,
)

config = {
    "id": "retry",
    "initial": "waiting",
    "context": {
        "attempt": 2,
    },
    "states": {
        "waiting": {
            "after": {
                "BACKOFF": "trying",
            },
            "on": {
                "NUDGE": {
                    "target": "waiting",
                    "reenter": True,
                },
            },
        },
        "trying": {},
    },
}

logic = MachineLogic(
    delays={
        "BACKOFF": lambda args: 1000 * args["context"]["attempt"],
    }
)
clock = SimulatedClock()
interp = SyncInterpreter(
    create_machine(config, logic=logic), clock=clock
).start()

clock.increment(1500)
interp.send("NUDGE")  # reenter: the timer restarts from zero
clock.increment(1500)
print(sorted(interp.active_state_ids))
# ['retry.waiting']
clock.increment(500)
print(sorted(interp.active_state_ids))
# ['retry.trying']
interp.stop()

Gotchas: a name that is not in MachineLogic.delays logs a warning and disables that timer (the machine still starts); a delay that resolves to something that is not a number is logged as an error and ignored the same way.

🐍 Pythonic After

The State class accepts an after parameter — a dict mapping millisecond delays to targets:

from xstate_statemachine import State, StateMachine, SyncInterpreter, action


class SessionMachine(StateMachine):
    machine_id = "session"

    active = State(
        "active",
        initial=True,
        after={
            300000: "warning",
        },
        on={
            "ACTIVITY": "active",
        },
    )
    warning = State(
        "warning",
        after={
            30000: "expired",
        },
        on={
            "EXTEND": "active",
        },
    )
    expired = State("expired", final=True)


machine = SessionMachine.create_machine()
interp = SyncInterpreter(machine).start()

print(interp.active_state_ids)
# {'session.active'}

interp.send("ACTIVITY")
print(interp.active_state_ids)
# {'session.active'}

interp.stop()

Using the functional API:

from xstate_statemachine import State, build_machine, SyncInterpreter

active = State(
    "active",
    initial=True,
    after={
        300000: "warning",
    },
    on={
        "ACTIVITY": "active",
    },
)
warning = State(
    "warning",
    after={
        30000: "expired",
    },
    on={
        "EXTEND": "active",
    },
)
expired = State("expired", final=True)

machine = build_machine(id="session", states=[active, warning, expired])
interp = SyncInterpreter(machine).start()
interp.stop()

Using the builder API:

from xstate_statemachine import MachineBuilder, SyncInterpreter

machine = (
    MachineBuilder("session")
    .state(
        "active",
        initial=True,
        after={
            300000: "warning",
        },
        on={
            "ACTIVITY": "active",
        },
    )
    .state(
        "warning",
        after={
            30000: "expired",
        },
        on={
            "EXTEND": "active",
        },
    )
    .state("expired", final=True)
    .build()
)

interp = SyncInterpreter(machine).start()
interp.stop()

🪆 After in Nested States

Delayed transitions work inside compound (hierarchical) states. Timers in a child state are cancelled when the parent state is exited:

{
  "id": "nestedTimer",
  "initial": "loggedIn",
  "states": {
    "loggedIn": {
      "initial": "dashboard",
      "states": {
        "dashboard": {
          "after": {
            "60000": {
              "target": "dashboard",
              "actions": "refreshData"
            }
          }
        },
        "settings": {}
      },
      "on": {
        "LOGOUT": "loggedOut"
      }
    },
    "loggedOut": {}
  }
}

When LOGOUT fires, the machine exits dashboard (cancelling its 60-second refresh timer) and then exits loggedIn.

⚖️ Sync vs Async Timers

Interpreter	Timer Behavior
`Interpreter` (async)	Schedules through the interpreter’s `Clock`. By default (`RealClock`) that’s `loop.call_later`, the same primitive `asyncio.sleep` uses, so timers run concurrently with your event loop.
`SyncInterpreter`	Also schedules through a `Clock`. With the default `RealClock` a timer is a deadline recorded in a heap — no thread is started. Due timers fire when `send()` or `tick()` next runs, on the caller’s own thread.

Tip: For real-time timer behavior (actual wall-clock delays), use the async Interpreter with asyncio, or call RealClock.pump() / SyncInterpreter.tick() periodically to fire due SyncInterpreter timers if you aren’t calling send() often enough on your own.

after guarantees “not before”, never “at”. Both engines route every timer through the same priority lane so a due after or delayed send does not queue behind a backlog of already-pending external events — see Controlling Time below. It can still fire late while other machines on the same loop finish their macrosteps — measured in Production Characteristics § 2. On the SyncInterpreter, timers now fire on the thread that calls send()/tick(), never on a background thread (§ 3).

Measuring lateness with `event.lateness_ms`

Every after and delayed-send() event delivered to your actions is an AfterEvent. It carries two raw clock readings — scheduled_for (when the timer was due) and fired_at (when it actually fired) — plus a convenience property, lateness_ms, that computes max(0.0, (fired_at - scheduled_for) * 1000.0). Read it from inside the action that handles the fired timer to alarm on real drift instead of inferring timer starvation from symptoms:

import asyncio
from xstate_statemachine import create_machine, Interpreter, MachineLogic

config = {
    "id": "watchdog",
    "initial": "waiting",
    "states": {
        "waiting": {
            "after": {
                "50": {
                    "target": "waiting",
                    "actions": "checkLateness",
                },
            },
        },
    },
}


def check_lateness(interpreter, context, event, action_def):
    print(f"lateness_ms={event.lateness_ms:.2f}")


async def main():
    machine = create_machine(
        config,
        logic=MachineLogic(
            actions={
                "check_lateness": check_lateness,
            }
        ),
    )
    interp = await Interpreter(machine).start()
    await asyncio.sleep(0.2)
    await interp.stop()


asyncio.run(main())

If lateness_ms trends upward over time, the event loop (or, for the SyncInterpreter, the calling thread) isn’t being given enough opportunities to pump() due timers — see Production Characteristics § 2 for the underlying guarantee.

🕹️ Controlling Time

Every timer in the library — after transitions and delayed send() — is scheduled through an injectable Clock, not called directly against asyncio.sleep() or a background thread. This is the same seam XState v5 uses (createActor(machine, { clock })), and it solves two problems at once:

Tests don’t have to wait. A 30-second after timeout can be fired in a fraction of a millisecond of real time by advancing a SimulatedClock instead of sleeping.
A due timer isn’t starved by a busy inbox. Both engines run timers through a dedicated priority lane, so a timer that becomes due is not queued behind thousands of already-pending external events.

The `Clock` protocol

from xstate_statemachine import Clock, RealClock, SimulatedClock

A Clock exposes now(), set_timeout(fn, delay_sec, owner=None), clear_timeout(handle), and pump(). Both built-in clocks also provide wall_now() (epoch seconds, used for snapshot deadlines) and, since 0.12.0, pump_filtered(owner_filter) – a pump restricted to one owner’s deadlines, which is how a SyncInterpreter child sharing its parent’s SimulatedClock avoids firing the parent’s timers from its own thread. Both Interpreter and SyncInterpreter accept a clock= constructor argument; if you don’t pass one, each gets its own RealClock().

`RealClock` — wall time, the default

Inside a running event loop (the async Interpreter), a timeout is loop.call_later — identical to today’s asyncio.sleep-based timing.
Outside a loop (the SyncInterpreter), a timeout is a deadline in a heap. No thread is created. Due callbacks run only when something calls pump() — which SyncInterpreter.send() and tick() do automatically, on the caller’s own thread.

from xstate_statemachine import RealClock

clock = RealClock()
clock.set_timeout(lambda: print("fired"), 0.0)
clock.pump()          # runs every due callback now; returns how many fired
clock.pending          # number of deadlines still waiting (sync timers only)

If your SyncInterpreter isn’t receiving events regularly enough for its own send() calls to drain due timers, call interp.clock.pump() (or interp.tick()) yourself on a schedule.

`SimulatedClock` — virtual time for tests

Time does not pass on its own. Call increment(ms) to advance virtual time and fire everything that became due, in due order:

from xstate_statemachine import create_machine, SyncInterpreter, SimulatedClock

config = {
    "id": "sessionTimeout",
    "initial": "active",
    "states": {
        "active": {
            "after": {
                "300000": "warning",
            },
            "on": {
                "ACTIVITY": {
                    "target": "active",
                    "reenter": True,
                },
            },
        },
        "warning": {
            "after": {
                "30000": "expired",
            },
            "on": {
                "EXTEND": "active",
            },
        },
        "expired": {
            "type": "final",
        },
    },
}
machine = create_machine(config)

clock = SimulatedClock()
interp = SyncInterpreter(machine, clock=clock).start()
print(interp.active_state_ids)  # {'sessionTimeout.active'}

clock.increment(300_000)  # 5 minutes pass, instantly
print(interp.active_state_ids)  # {'sessionTimeout.warning'}

clock.increment(30_000)  # another 30 seconds pass
print(interp.active_state_ids)  # {'sessionTimeout.expired'}

interp.stop()

On the async Interpreter, increment() returns an awaitable so the caller can be sure the fired timer has finished being processed (including any onward transitions) before checking state:

clock = SimulatedClock()
i = await Interpreter(create_machine(config), clock=clock).start()
await clock.increment(300_000)
await clock.increment(30_000)
await i.stop()

Calling increment()/set() inside a running event loop without await raises a warning at the next garbage collection — a forgotten await would otherwise silently race the loop instead of failing loudly.

set(ms) jumps to an absolute virtual time instead of advancing by a delta, and refuses to move backwards. pending reports how many live timers are still scheduled, next_due() returns the virtual time (seconds) of the earliest one (or None), and fire_until(seconds) advances to an absolute time synchronously, firing due timers in order (for code that holds no running loop). SimulatedClock(wall_start=...) sets the virtual wall clock (wall_now() is wall_start plus elapsed virtual time), so a test can express “restarted an hour later”.

A Clock is not tied to one engine. Pass the same instance to a parent and to the children it invokes — including a mix of async and sync interpreters — and they share a single timeline. A parent’s invoked child inherits its clock automatically; if you construct a standalone SyncInterpreter you want on the same timeline, pass clock= explicitly:

from xstate_statemachine import (
    create_machine,
    SyncInterpreter,
    Interpreter,
    SimulatedClock,
)

child = {
    "id": "kid",
    "initial": "w",
    "states": {
        "w": {
            "after": {
                "100": "d",
            },
        },
        "d": {
            "type": "final",
        },
    },
}
parent = {
    "id": "p",
    "initial": "a",
    "states": {
        "a": {
            "after": {
                "500": "b",
            },
        },
        "b": {},
    },
}

clock = SimulatedClock()
kid = SyncInterpreter(create_machine(child), clock=clock).start()


async def main():
    i = await Interpreter(create_machine(parent), clock=clock).start()
    await clock.increment(100)
    print(
        set(kid.current_state_ids), set(i.current_state_ids)
    )  # {'kid.d'} {'p.a'}
    await clock.increment(400)
    print(
        set(kid.current_state_ids), set(i.current_state_ids)
    )  # {'kid.d'} {'p.b'}
    await i.stop()

One increment() call fires timers for every interpreter attached to the clock, in due order, regardless of which engine owns each one.

Why a busy inbox no longer delays a due timer

Before 0.8.0, a fired async timer’s continuation had to queue behind every external event already sitting in the inbox, which could add measurable lateness on top of the OS timer floor under heavy load. Both engines now run clock-fired callbacks through a dedicated priority lane, so a due after or delayed send is processed promptly even while thousands of send() calls are backlogged. This is what makes after usable as a watchdog alongside high event volume.

⏲️ Delayed Self-Sends Are Timers Too

raise (or send to the machine itself) with a delay is a timer with exactly the standing of after (0.9.0, #212): arming it ends the current step’s self-generated chain, its firing is a clock event, and it is never counted against maxIterations. A self-paced heartbeat or poller written either way runs indefinitely:

"up": {
  "entry": [
    {
      "type": "raise",
      "params": {
        "event": "BEAT",
        "delay": 1000
      }
    }
  ],
  "on": {
    "BEAT": "down"
  }
},
"down": {
  "entry": [
    {
      "type": "raise",
      "params": {
        "event": "BEAT",
        "delay": 1000
      }
    }
  ],
  "on": {
    "BEAT": "up"
  }
}

Only a zero-delay self-raise cycle is charged to the budget and cut with RunawayChainError. Give the send an id and cancel it to disarm. Leaving the state that armed it does not cancel a delayed send (XState v5): it is still delivered after the machine has moved on, and only cancel disarms it. The clock handle is released when the send fires or is cancelled (#218).

Timers and snapshots

A snapshot records each armed after timer’s wall-clock deadline (deadlines, layout v4). After a static from_snapshot() restore the timer is dormant — interp.has_dormant_timers is True — until you pass restart_timers= (defaults to the value of restart_services): True / "restart" re-arms every dormant timer from zero on start(), "resume" re-arms the remaining time, and "fire_due" additionally fires matured timers during start(). persisted() defaults to "resume". A timer that had already fired and was waiting in the priority lane is persisted and replays. An armed delayed self-send is persisted with its remaining delay (scheduled_sends, layout v4) and resumes where it would have been — and it survives a restore → re-persist hop that never calls start(). See Snapshots.

📡 Complete Example: Polling Machine

A machine that polls an API at regular intervals, with error handling and backoff:

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "poller",
    "initial": "idle",
    "context": {
        "data": None,
        "errors": 0,
        "maxErrors": 3,
    },
    "states": {
        "idle": {
            "on": {
                "START": "polling",
            },
        },
        "polling": {
            "invoke": {
                "src": "fetchData",
                "onDone": {
                    "target": "waiting",
                    "actions": "storeData",
                },
                "onError": {
                    "target": "retrying",
                    "actions": "incrementErrors",
                },
            },
        },
        "waiting": {
            "after": {
                "10000": "polling",
            },
            "on": {
                "STOP": "idle",
            },
        },
        "retrying": {
            "after": {
                "30000": {
                    "target": "polling",
                    "guard": "belowMaxErrors",
                },
                "30001": {
                    "target": "failed",
                    "guard": "atMaxErrors",
                },
            },
        },
        "failed": {
            "on": {
                "RESET": "idle",
            },
        },
    },
}


class PollerLogic(MachineLogic):
    def fetch_data(self, interpreter, context, event):
        return {
            "status": "ok",
            "value": 42,
        }

    def store_data(self, interpreter, context, event, action_def):
        context["data"] = event.data
        context["errors"] = 0

    def increment_errors(self, interpreter, context, event, action_def):
        context["errors"] += 1

    def below_max_errors(self, context, event):
        return context["errors"] < context["maxErrors"]

    def at_max_errors(self, context, event):
        return context["errors"] >= context["maxErrors"]


machine = create_machine(config, logic=PollerLogic())
interp = SyncInterpreter(machine).start()

interp.send("START")
print(interp.active_state_ids)
# {'poller.waiting'}  (fetch_data succeeded, now waiting 10s to poll again)

print(interp.context["data"])
# {'status': 'ok', 'value': 42}

interp.send("STOP")
print(interp.active_state_ids)
# {'poller.idle'}

interp.stop()

💾 Complete Example: Auto-Save with Debounce

A document editor that auto-saves 2 seconds after the last edit:

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "autoSave",
    "initial": "clean",
    "context": {
        "content": "",
        "lastSaved": None,
    },
    "states": {
        "clean": {
            "on": {
                "EDIT": {
                    "target": "dirty",
                    "actions": "updateContent",
                },
            },
        },
        "dirty": {
            "after": {
                "2000": "saving",
            },
            "on": {
                "EDIT": {
                    "target": "dirty",
                    "reenter": True,
                    "actions": "updateContent",
                },
            },
        },
        "saving": {
            "invoke": {
                "src": "saveDocument",
                "onDone": {
                    "target": "clean",
                    "actions": "markSaved",
                },
                "onError": {
                    "target": "dirty",
                },
            },
        },
    },
}


class AutoSaveLogic(MachineLogic):
    def update_content(self, interpreter, context, event, action_def):
        context["content"] = event.data.get("text", context["content"])
        print(f"Content updated: {context['content']!r}")

    def save_document(self, interpreter, context, event):
        print(f"Saving: {context['content']!r}")
        return {
            "saved": True,
        }

    def mark_saved(self, interpreter, context, event, action_def):
        import time

        context["lastSaved"] = time.time()
        print("Document saved!")


machine = create_machine(config, logic=AutoSaveLogic())
interp = SyncInterpreter(machine).start()

# Type something
interp.send(
    {
        "type": "EDIT",
        "text": "Hello",
    }
)
print(interp.active_state_ids)
# {'autoSave.dirty'}

# Type more — reenter: True resets the 2-second debounce timer
interp.send(
    {
        "type": "EDIT",
        "text": "Hello, world!",
    }
)
print(interp.active_state_ids)
# {'autoSave.dirty'}  (timer restarted)

# After 2 seconds of no edits, auto-save would fire
interp.stop()

The key insight: each EDIT event re-enters dirty ("reenter": true), which resets the 2-second timer. The save only fires after the user stops typing for 2 full seconds. This is the classic debounce pattern, expressed declaratively.

Edit this page on GitHub