FSM History

State transition audit log stored as a ring-buffer in FSM storage. Useful for debugging, analytics, and undo/back navigation. Added in v0.5.0.

Recording Transitions

Call push_history() whenever you transition to a new state. Each entry stores the state name and a Unix timestamp.

from kurigram_addons import FSMContext, StatesGroup, State

class OrderFlow(StatesGroup):
    cart     = State()
    checkout = State()
    payment  = State()
    done     = State()

@router.on_command("start_order")
async def start(client, message, state: FSMContext):
    await state.set_state(OrderFlow.cart)
    await state.push_history("OrderFlow:cart")
    await message.reply("Cart ready.")

@router.on_callback_query(OrderFlow.checkout.filter())
async def go_checkout(client, query, state: FSMContext):
    await state.set_state(OrderFlow.checkout)
    await state.push_history(str(OrderFlow.checkout))
    await query.answer()

Reading History

@router.on_command("history")
async def show_history(client, message, state: FSMContext):
    entries = await state.get_history(limit=5)

    lines = []
    for e in entries:
        import datetime
        ts = datetime.datetime.fromtimestamp(e["at"]).strftime("%H:%M:%S")
        lines.append(f"{ts} → {e['state']}")

    await message.reply("\n".join(lines) or "No history.")

Entries are returned newest-last. The ring-buffer caps at 50 entries— older entries are dropped automatically.

Back Navigation

Use history to implement a "go back" button by reading the previous state from the log and transitioning back to it.

@router.on_callback_query(filters.regex(r"^back$"))
async def go_back(client, query, state: FSMContext):
    entries = await state.get_history(limit=2)
    if len(entries) < 2:
        await query.answer("Nothing to go back to.")
        return
    prev_state = entries[-2]["state"]   # entry before current
    await state.set_state(prev_state)
    await query.answer(f"Back to {prev_state}")

Clearing History

# Clear only history (state & data are unaffected)
await state.clear_history()

# Or clear everything at end of flow
await state.clear_state()
await state.clear_history()

API Reference

Method	Description
`await push_history(state_name, *, ttl=None)`	Append state_name + current timestamp to the ring-buffer
`await get_history(limit=10)`	Return up to limit most recent entries as `[{"state": str, "at": float}]`, newest last
`await clear_history()`	Delete the entire history for this identifier

History is stored under a separate key (identifier + "__history__") so it never interferes with state or data.