Final States

Terminal completion states — signal that a machine or region has finished.

31 min read

A final state is a terminal state — the machine (or a compound state region) has completed its work. No outgoing transitions are allowed. When a final state is entered, it signals done to its parent, enabling powerful composition patterns.

🏁 What Are Final States?

stateDiagram-v2
    direction LR
    [*] --> draft
    draft --> review : SUBMIT
    review --> approved : APPROVE
    review --> draft : REJECT
    approved --> [*] : ✅ done · onDone fires in the parent

Final states represent the end of a workflow. Once a machine (or a nested region) enters a final state, it stops processing events for that branch. There are no outgoing transitions — the machine is finished.

Final states are the statechart equivalent of a function’s return statement. They tell the parent “I’m done” and optionally pass data back.

📄 JSON: `"type": "final"`

{
  "confirmation": {
    "type": "final"
  }
}

🐍 Pythonic: `State("done", final=True)`

from xstate_statemachine import State

done = State("done", final=True)

Warning: A final state cannot have outgoing transitions (on), child states (states), or be marked as parallel. Attempting any of these will raise an InvalidConfigError.

🛒 Basic Example: Checkout Flow

A simple checkout that progresses through three stages:

{
  "id": "checkout",
  "initial": "cart",
  "states": {
    "cart": {
      "on": {
        "CHECKOUT": "payment"
      }
    },
    "payment": {
      "on": {
        "PAY": "confirmation"
      }
    },
    "confirmation": {
      "type": "final"
    }
  }
}

from xstate_statemachine import create_machine, SyncInterpreter

config = {
    "id": "checkout",
    "initial": "cart",
    "states": {
        "cart": {
            "on": {
                "CHECKOUT": "payment",
            },
        },
        "payment": {
            "on": {
                "PAY": "confirmation",
            },
        },
        "confirmation": {
            "type": "final",
        },
    },
}

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

print(interp.active_state_ids)
# {'checkout.cart'}

interp.send("CHECKOUT")
interp.send("PAY")
print(interp.active_state_ids)
# {'checkout.confirmation'}

# Sending further events has no effect — the machine is done
interp.send("CHECKOUT")
print(interp.active_state_ids)
# {'checkout.confirmation'}  — still in final state

interp.stop()

Once the machine reaches confirmation, it stays there. No event can move it out — that’s the guarantee of a final state.

🎬 What Happens When a Final State Is Entered

sequenceDiagram
    participant C as child region
    participant F as final state
    participant P as parent state
    C->>F: transition
    F->>F: ① entry actions
    F-->>P: ② done.state.parent (+ output)
    P->>P: ③ onDone transition

When a final state is entered, several things happen:

Entry actions on the final state execute (if any are defined)
The machine signals completion — no further event processing occurs for this branch
done.state.* event fires on the parent — if the final state is inside a compound state, the parent receives a DoneEvent (and a parallel ancestor whose regions are now all final receives one too; see Which states are told they are done). It is an internal event, like a raise: it is processed right after the current step, before any external event already waiting in the queue (send_events(["FINISH", "NEXT"]) takes the parent’s onDone transition before NEXT, as XState v5 does)
All after timers are cancelled — any pending delayed transitions in the machine/region are cleaned up
A top-level final state completes the machine — the machine’s output is resolved, then the exit actions of every state that is still active (the final state itself and the machine root included) run, innermost first, and status becomes "done". The configuration is kept: a done machine still reports its final state. This matches XState v5; before 0.12.0 no exit actions ran on completion.

# Note: `DoneEvent` is not re-exported from the top-level package — import it
# from the `xstate_statemachine.events` submodule, as shown below.
from xstate_statemachine.events import DoneEvent

# "confirmation" is a top-level final state: entering it completes the
# machine. A final state inside a compound state "review" of this machine
# would instead raise a DoneEvent of type "done.state.checkout.review" --
# the PARENT's id -- which "review"'s `onDone` handles.

🪆 Final States in Compound States

The most powerful use of final states is inside compound (hierarchical) states. When a child reaches a final state, the parent receives a done.state.* event and can react via onDone:

{
  "id": "workflow",
  "initial": "processing",
  "states": {
    "processing": {
      "initial": "step1",
      "states": {
        "step1": {
          "on": {
            "NEXT": "step2"
          }
        },
        "step2": {
          "on": {
            "NEXT": "step3"
          }
        },
        "step3": {
          "on": {
            "FINISH": "complete"
          }
        },
        "complete": {
          "type": "final"
        }
      },
      "onDone": "finished"
    },
    "finished": {
      "type": "final"
    }
  }
}

from xstate_statemachine import create_machine, SyncInterpreter

config = {
    "id": "workflow",
    "initial": "processing",
    "states": {
        "processing": {
            "initial": "step1",
            "states": {
                "step1": {
                    "on": {
                        "NEXT": "step2",
                    },
                },
                "step2": {
                    "on": {
                        "NEXT": "step3",
                    },
                },
                "step3": {
                    "on": {
                        "FINISH": "complete",
                    },
                },
                "complete": {
                    "type": "final",
                },
            },
            "onDone": "finished",
        },
        "finished": {
            "type": "final",
        },
    },
}

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

print(interp.active_state_ids)
# {'workflow.processing.step1'}

interp.send("NEXT")  # step1 → step2
interp.send("NEXT")  # step2 → step3
interp.send("FINISH")  # step3 → complete (final)
# → triggers onDone → processing exits → finished

print(interp.active_state_ids)
# {'workflow.finished'}

interp.stop()

The flow:

step3 receives FINISH → transitions to complete
complete is final → done.state.workflow.processing.complete fires
processing has onDone: "finished" → transitions to finished
finished is also final → the entire machine is done

Tip: onDone is the key mechanism for composing sub-workflows. Each compound state can encapsulate a multi-step process, and the parent only needs to know “when it’s done.”

🎬 onDone with Actions

The onDone transition can include actions, just like any other transition:

{
  "processing": {
    "initial": "validating",
    "states": {
      "validating": {
        "on": {
          "VALID": "complete"
        }
      },
      "complete": {
        "type": "final"
      }
    },
    "onDone": {
      "target": "finished",
      "actions": "logCompletion"
    }
  }
}

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "withActions",
    "initial": "processing",
    "states": {
        "processing": {
            "initial": "validating",
            "states": {
                "validating": {
                    "on": {
                        "VALID": "complete",
                    },
                },
                "complete": {
                    "type": "final",
                },
            },
            "onDone": {
                "target": "finished",
                "actions": "logCompletion",
            },
        },
        "finished": {
            "type": "final",
        },
    },
}


class CompletionLogic(MachineLogic):
    def log_completion(self, interpreter, context, event, action_def):
        print("Processing complete! Moving to finished state.")


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

interp.send("VALID")
# Output: Processing complete! Moving to finished state.
print(interp.active_state_ids)
# {'withActions.finished'}

interp.stop()

🎛️ Final States in Parallel Regions

In a parallel state, each region can have its own final state. Important rules:

A region reaching its final state does not stop other regions
All regions must reach a final state for the parent’s onDone to fire
Other regions continue processing events normally until they also finish

{
  "id": "parallelFinal",
  "initial": "processing",
  "states": {
    "processing": {
      "type": "parallel",
      "states": {
        "upload": {
          "initial": "uploading",
          "states": {
            "uploading": {
              "on": {
                "UPLOAD_DONE": "complete"
              }
            },
            "complete": {
              "type": "final"
            }
          }
        },
        "validate": {
          "initial": "checking",
          "states": {
            "checking": {
              "on": {
                "VALID": "complete"
              }
            },
            "complete": {
              "type": "final"
            }
          }
        }
      },
      "onDone": "finished"
    },
    "finished": {
      "type": "final"
    }
  }
}

from xstate_statemachine import create_machine, SyncInterpreter

config = {
    "id": "parallelFinal",
    "initial": "processing",
    "states": {
        "processing": {
            "type": "parallel",
            "states": {
                "upload": {
                    "initial": "uploading",
                    "states": {
                        "uploading": {
                            "on": {
                                "UPLOAD_DONE": "complete",
                            },
                        },
                        "complete": {
                            "type": "final",
                        },
                    },
                },
                "validate": {
                    "initial": "checking",
                    "states": {
                        "checking": {
                            "on": {
                                "VALID": "complete",
                            },
                        },
                        "complete": {
                            "type": "final",
                        },
                    },
                },
            },
            "onDone": "finished",
        },
        "finished": {
            "type": "final",
        },
    },
}

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

print(interp.active_state_ids)
# {'parallelFinal.processing.upload.uploading',
#  'parallelFinal.processing.validate.checking'}

# Upload finishes first — validate is still running
interp.send("UPLOAD_DONE")
print(interp.active_state_ids)
# {'parallelFinal.processing.upload.complete',
#  'parallelFinal.processing.validate.checking'}

# Now validate finishes — BOTH regions are final → onDone fires
interp.send("VALID")
print(interp.active_state_ids)
# {'parallelFinal.finished'}

interp.stop()

Note: The onDone on processing only fires when both upload and validate have reached their final states. This is how parallel states enable “wait for all” patterns.

Which states are told they are done

Entering a final state raises done.state.* events exactly as XState v5 does:

Its parent, when that is a compound state (a region, or any other compound), receives done.state.<parent> carrying the final state’s output.
Then, walking up from there: while the next ancestor is a parallel state whose regions are all in a final state, that parallel state receives done.state.<parallel> too (a history child is not a region). A region that finishes last therefore produces two events in a row, the region’s and the parallel state’s, and both onDone handlers run, in that order. Nested parallel states continue the chain.
Nothing else is notified. In particular a compound state whose grandchild is final is not done: for a → b → f (f final inside b), b’s onDone runs, a’s does not — a completes only when one of its own final children is entered.

Every event is raised whether or not the state declares onDone; one nobody handles reaches a "*" handler (or is dropped like any unhandled engine event). When the walk reaches the root — a final state directly under it, or a parallel root whose regions are all final — the machine completes.

Changed in 0.12.0: only the first ancestor with an onDone that was “done” was notified (a compound counted as done when its active child was, recursively), so a grandparent’s onDone fired in step 3, a parallel state’s onDone was skipped when the finishing region had its own, and a parallel root never completed the machine.

🔀 Multiple Final States: Success/Failure Patterns

A machine can have multiple final states to represent different outcomes:

{
  "id": "payment",
  "initial": "processing",
  "states": {
    "processing": {
      "invoke": {
        "src": "chargeCard",
        "onDone": "success",
        "onError": "failed"
      }
    },
    "success": {
      "type": "final"
    },
    "failed": {
      "type": "final"
    }
  }
}

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "payment",
    "initial": "processing",
    "states": {
        "processing": {
            "invoke": {
                "src": "chargeCard",
                "onDone": "success",
                "onError": "failed",
            },
        },
        "success": {
            "type": "final",
        },
        "failed": {
            "type": "final",
        },
    },
}


class PaymentLogic(MachineLogic):
    def charge_card(self, interpreter, context, event):
        # Simulate successful payment
        return {
            "transactionId": "TXN-9876",
        }


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

print(interp.active_state_ids)
# {'payment.success'}

# Determine which final state we ended in
if "payment.success" in interp.active_state_ids:
    print("Payment succeeded!")
elif "payment.failed" in interp.active_state_ids:
    print("Payment failed!")

interp.stop()

This pattern is useful when a parent needs to distinguish between different completion outcomes:

{
  "checkout": {
    "initial": "paying",
    "states": {
      "paying": {
        "initial": "processing",
        "states": {
          "processing": {
            "invoke": {
              "src": "chargeCard",
              "onDone": "success",
              "onError": "failed"
            }
          },
          "success": {
            "type": "final"
          },
          "failed": {
            "type": "final"
          }
        },
        "onDone": "confirmation"
      }
    },
    "confirmation": {}
  }
}

🐍 Pythonic Final States

In StateMachine Class

from xstate_statemachine import State, StateMachine, SyncInterpreter

class OrderMachine(StateMachine):
    machine_id = "order"

    pending    = State("pending", initial=True)
    processing = State("processing")
    completed  = State("completed", final=True)
    cancelled  = State("cancelled", final=True)

    start  = pending.to(processing, event="START")
    finish = processing.to(completed, event="COMPLETE")
    cancel = pending.to(cancelled, event="CANCEL")

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

interp.send("START")
interp.send("COMPLETE")
print(interp.active_state_ids)
# {'order.completed'}

interp.stop()

In MachineBuilder

from xstate_statemachine import MachineBuilder, SyncInterpreter

machine = (
    MachineBuilder("order")
    .state("pending",    initial=True)
    .state("processing")
    .state("completed",  final=True)
    .state("cancelled",  final=True)
    .transition("pending",    "START",    "processing")
    .transition("processing", "COMPLETE", "completed")
    .transition("pending",    "CANCEL",   "cancelled")
    .build()
)

interp = SyncInterpreter(machine).start()
interp.send("CANCEL")
print(interp.active_state_ids)
# {'order.cancelled'}

interp.stop()

In Functional API

from xstate_statemachine import State, build_machine, SyncInterpreter

pending    = State("pending", initial=True)
processing = State("processing")
completed  = State("completed", final=True)
cancelled  = State("cancelled", final=True)

t1 = pending.to(processing, event="START")
t2 = processing.to(completed, event="COMPLETE")
t3 = pending.to(cancelled, event="CANCEL")

machine = build_machine(
    id="order",
    states=[pending, processing, completed, cancelled],
    transitions=[t1, t2, t3],
)

interp = SyncInterpreter(machine).start()
interp.send("START")
interp.send("COMPLETE")
print(interp.active_state_ids)
# {'order.completed'}

interp.stop()

📦 Complete Example: Order Processing with Multiple Final Outcomes

An order processing system with three possible terminal states:

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "orderProcess",
    "initial": "received",
    "context": {
        "orderId": None,
        "reason": None,
        "refundAmount": 0,
    },
    "states": {
        "received": {
            "entry": "assignOrderId",
            "on": {
                "APPROVE": "fulfilling",
                "REJECT": {
                    "target": "cancelled",
                    "actions": "setRejectionReason",
                },
            },
        },
        "fulfilling": {
            "initial": "picking",
            "states": {
                "picking": {
                    "on": {
                        "PICKED": "packing",
                        "OUT_OF_STOCK": {
                            "target": "stockFailed",
                            "actions": "setStockFailure",
                        },
                    },
                },
                "packing": {
                    "on": {
                        "PACKED": "shipping",
                    },
                },
                "shipping": {
                    "invoke": {
                        "src": "shipOrder",
                        "onDone": "shipped",
                        "onError": {
                            "target": "shipFailed",
                            "actions": "setShipFailure",
                        },
                    },
                },
                "shipped": {
                    "type": "final",
                },
                "stockFailed": {
                    "type": "final",
                },
                "shipFailed": {
                    "type": "final",
                },
            },
            "onDone": "evaluating",
        },
        "evaluating": {
            "on": {
                "": [
                    {
                        "target": "completed",
                        "guard": "wasShipped",
                    },
                    {
                        "target": "refunded",
                        "actions": "calculateRefund",
                    },
                ],
            },
        },
        "completed": {
            "type": "final",
            "entry": "sendConfirmationEmail",
        },
        "cancelled": {
            "type": "final",
            "entry": "sendCancellationEmail",
        },
        "refunded": {
            "type": "final",
            "entry": "processRefund",
        },
    },
}


class OrderLogic(MachineLogic):
    def assign_order_id(self, interpreter, context, event, action_def):
        import random

        context["orderId"] = f"ORD-{random.randint(10000, 99999)}"
        print(f"Order created: {context['orderId']}")

    def set_rejection_reason(self, interpreter, context, event, action_def):
        context["reason"] = event.data.get("reason", "Unknown")

    def set_stock_failure(self, interpreter, context, event, action_def):
        context["reason"] = "Out of stock"

    def set_ship_failure(self, interpreter, context, event, action_def):
        context["reason"] = "Shipping failed"

    def ship_order(self, interpreter, context, event):
        print(f"Shipping order {context['orderId']}...")
        return {
            "trackingNumber": "TRACK-12345",
        }

    def was_shipped(self, context, event):
        return context.get("reason") is None

    def calculate_refund(self, interpreter, context, event, action_def):
        context["refundAmount"] = 49.99
        print(f"Refund calculated: ${context['refundAmount']}")

    def send_confirmation_email(self, interpreter, context, event, action_def):
        print(f"Order {context['orderId']} completed!")

    def send_cancellation_email(self, interpreter, context, event, action_def):
        print(f"Order {context['orderId']} cancelled: {context['reason']}")

    def process_refund(self, interpreter, context, event, action_def):
        print(
            f"Refund of ${context['refundAmount']} processed for {context['orderId']}"
        )


machine = create_machine(config, logic=OrderLogic())

# Happy path: order ships successfully
interp = SyncInterpreter(machine).start()
interp.send("APPROVE")
interp.send("PICKED")
interp.send("PACKED")
# ship_order service runs → shipped (final) → onDone → evaluating → completed

print(interp.active_state_ids)
# {'orderProcess.completed'}

interp.stop()

✅ Complete Example: Multi-Step Approval Workflow

A document approval process that requires multiple sign-offs:

from xstate_statemachine import (
    State,
    StateMachine,
    SyncInterpreter,
    action,
    guard,
)


class ApprovalWorkflow(StateMachine):
    machine_id = "approval"
    # `input` (passed to the interpreter) seeds the document; the rest
    # is fixed initial context.
    initial_context = {
        "document": None,
        "managerApproved": False,
        "directorApproved": False,
        "rejectedBy": None,
    }

    # States
    draft = State(
        "draft",
        initial=True,
        on={
            "SUBMIT": {
                "target": "review",
                "guard": "hasDocument",
            },
        },
    )

    review = State(
        "review",
        states=[
            State(
                "managerReview",
                initial=True,
                on={
                    "MANAGER_APPROVE": {
                        "target": "directorReview",
                        "actions": "recordManagerApproval",
                    },
                    "MANAGER_REJECT": {
                        "target": "rejected",
                        "actions": "recordRejection",
                    },
                },
            ),
            State(
                "directorReview",
                on={
                    "DIRECTOR_APPROVE": {
                        "target": "approved",
                        "actions": "recordDirectorApproval",
                    },
                    "DIRECTOR_REJECT": {
                        "target": "rejected",
                        "actions": "recordRejection",
                    },
                },
            ),
            State("approved", final=True),
            State("rejected", final=True),
        ],
        on_done="evaluating",
    )

    evaluating = State(
        "evaluating",
        on={
            "": [
                {
                    "target": "published",
                    "guard": "wasApproved",
                },
                {
                    "target": "returned",
                },
            ],
        },
    )

    published = State("published", final=True)
    returned = State(
        "returned",
        on={
            "REVISE": "draft",
        },
    )

    # Guards
    @guard
    def has_document(self, context, event):
        return (
            context.get("document") is not None
            or (context.get("input") or {}).get("document") is not None
        )

    @guard
    def was_approved(self, context, event):
        return context.get("managerApproved") and context.get(
            "directorApproved"
        )

    # Actions
    @action
    def record_manager_approval(self, interpreter, context, event, action_def):
        context["managerApproved"] = True
        print("Manager approved")

    @action
    def record_director_approval(
        self, interpreter, context, event, action_def
    ):
        context["directorApproved"] = True
        print("Director approved")

    @action
    def record_rejection(self, interpreter, context, event, action_def):
        context["rejectedBy"] = event.data.get("by", "unknown")
        print(f"Rejected by: {context['rejectedBy']}")


machine = ApprovalWorkflow.create_machine()

# Test the approval flow. Initial context comes from the class; per-run
# data goes in through `input`, readable as `context["input"]`.
interp = SyncInterpreter(
    machine,
    input={
        "document": "Q4 Report",
    },
).start()

print(interp.active_state_ids)
# {'approval.draft'}

interp.send("SUBMIT")
print(interp.active_state_ids)
# {'approval.review.managerReview'}

interp.send("MANAGER_APPROVE")
print(interp.active_state_ids)
# {'approval.review.directorReview'}

interp.send("DIRECTOR_APPROVE")
# directorReview → approved (final) → onDone → evaluating → published
print(interp.active_state_ids)
# {'approval.published'}

interp.stop()

This workflow demonstrates how final states compose with compound states:

The review compound state contains two final states: approved and rejected
When either is reached, onDone fires on review, transitioning to evaluating
The evaluating state uses an eventless transition (always) with guards to route to the correct outcome
published is the top-level final state for successful approvals
returned allows the document to be revised and resubmitted

📤 Carrying Data Out with `output`

A final state can declare an output value — the data it hands back to whatever consumes the completion. output may be a literal (dict, string, number) or a callable of ({context, event}), matching XState’s dynamic-output form. Whichever final state is entered, its resolved output becomes the machine’s interp.output.

This is the natural way to carry an outcome (a transaction ID, an error reason, a computed total) out of a machine, instead of inferring the outcome only by checking which state ID is active:

from xstate_statemachine import create_machine, SyncInterpreter

config = {
    "id": "payment",
    "initial": "processing",
    "states": {
        "processing": {
            "on": {
                "SUCCEED": "success",
                "FAIL": "failed",
            },
        },
        "success": {
            "type": "final",
            # Callable output: resolved from context/event when entered.
            "output": lambda ev: {
                "transactionId": "TXN-001",
            },
        },
        "failed": {
            "type": "final",
            # Literal output also works.
            "output": {
                "reason": "declined",
            },
        },
    },
}

machine = create_machine(config)
interp = SyncInterpreter(machine).start()
interp.send("SUCCEED")

print(interp.status)
# 'done'
print(interp.output)
# {'transactionId': 'TXN-001'}

Detecting Completion Programmatically

Besides checking active_state_ids for a known final-state ID, the interpreter exposes completion as first-class, checkable state:

interp.status — becomes "done" once the machine reaches its top-level final state (it starts as "running").
interp.output — the resolved output (see above) from the final state that was entered, or None if none was declared.
await interp.wait_done() (async Interpreter only) — an asyncio.Future that resolves once the machine finishes.
to_promise(interp) — a standalone helper that awaits completion and returns interp.output, mirroring XState’s toPromise(actor).

import asyncio
from xstate_statemachine import create_machine, Interpreter, to_promise

config = {
    "id": "payment",
    "initial": "processing",
    "states": {
        "processing": {
            "on": {
                "SUCCEED": "success",
            },
        },
        "success": {
            "type": "final",
            "output": {
                "transactionId": "TXN-001",
            },
        },
    },
}


async def main():
    machine = create_machine(config)
    interp = await Interpreter(machine).start()
    interp.send("SUCCEED")

    output = await to_promise(interp)
    print(interp.status, output)
    # done {'transactionId': 'TXN-001'}


asyncio.run(main())

See also: Interpreters, Services & Invoke.

Edit this page on GitHub