Mix.install([
{:choreo, path: Path.expand("~/repos/elixir/choreo")},
# {:choreo, github: "code-shoily/choreo", branch: "main"},
{:finitomata, "~> 0.41"},
{:kino_vizjs, "~> 0.9.0"}
])Choreo and Finitomata are complementary Elixir libraries for finite-state machines:
| Choreo.FSM | Finitomata | |
|---|---|---|
| Feature | Design-time analysis & visualization | Runtime FSM execution with OTP |
| FSM Format | Graph-based (Yog.Multi.Graph) |
Transition list ([%Transition{}]) |
| Use case | Model, visualize, validate, minimize | Spawn, transition, persist, supervise |
This notebook demonstrates how to bridge them — design in Choreo, run in Finitomata, then import back into Choreo for post-hoc analysis.
This module provides bidirectional conversion between Choreo.FSM and Finitomata.
defmodule FinitomataAdapter do
@moduledoc """
Bidirectional adapter between Choreo.FSM and Finitomata.
- `to_finitomata_dsl/1` — Choreo FSM → Mermaid DSL string for `use Finitomata`
- `from_finitomata/1` — Finitomata module or transition list → Choreo FSM
"""
alias Choreo.FSM
@synthetic_events ~w(__start__ __end__ finitomata_back)a
# ── Choreo → Finitomata DSL ─────────────────────────────────────────────
@doc "Converts a Choreo FSM to a Finitomata-compatible Mermaid DSL string."
def to_finitomata_dsl(%FSM{} = fsm) do
fsm.graph.edges
|> Enum.map(fn {_eid, {from, to, label}} -> {from, to, label} end)
|> Enum.reject(fn {from, to, _} -> from == :* or to == :* end)
|> Enum.sort()
|> Enum.map_join("\n", fn {from, to, label} ->
"#{from} --> |#{label}| #{to}"
end)
end
# ── Finitomata → Choreo ─────────────────────────────────────────────────
@doc "Builds a Choreo FSM from a compiled Finitomata module."
def from_finitomata(finitomata_module) when is_atom(finitomata_module) do
unless function_exported?(finitomata_module, :__config__, 1) do
raise ArgumentError,
"#{inspect(finitomata_module)} does not appear to `use Finitomata`"
end
finitomata_module.__config__(:fsm) |> from_finitomata()
end
def from_finitomata(transitions) when is_list(transitions) do
user_transitions = Enum.reject(transitions, &(&1.event in @synthetic_events))
initial_states =
transitions
|> Enum.filter(&(&1.from == :*))
|> Enum.flat_map(&List.wrap(&1.to))
|> Enum.uniq()
final_states =
transitions
|> Enum.filter(&(&1.to == :* or (is_list(&1.to) and :* in &1.to)))
|> Enum.map(& &1.from)
|> Enum.uniq()
all_states =
transitions
|> Enum.flat_map(&[&1.from | List.wrap(&1.to)])
|> Enum.reject(&(&1 == :*))
|> Enum.uniq()
choreo =
Enum.reduce(all_states, FSM.new(), fn state, acc ->
acc =
if state in initial_states,
do: FSM.add_initial_state(acc, state),
else: FSM.add_state(acc, state)
if state in final_states,
do: put_in(acc.meta.final_states, MapSet.put(acc.meta.final_states, state)),
else: acc
end)
Enum.reduce(user_transitions, choreo, fn t, acc ->
if t.from == :* or t.to == :* or (is_list(t.to) and :* in t.to) do
acc
else
targets = List.wrap(t.to)
event_str = to_string(t.event)
if length(targets) > 1 do
Enum.reduce(targets, acc, fn target, inner_acc ->
FSM.add_transition(inner_acc, t.from, target, label: "#{event_str}→#{target}")
end)
else
FSM.add_transition(acc, t.from, hd(targets), label: event_str)
end
end
end)
end
endLet's model an order processing workflow in Choreo, then generate the Finitomata DSL from it.
alias Choreo.FSM
alias Choreo.FSM.Analysis
order_fsm =
FSM.new()
|> FSM.add_initial_state(:cart)
|> FSM.add_state(:checkout)
|> FSM.add_state(:payment_pending)
|> FSM.add_state(:payment_failed)
|> FSM.add_state(:processing)
|> FSM.add_state(:shipped)
|> FSM.add_final_state(:delivered)
|> FSM.add_final_state(:cancelled)
|> FSM.add_final_state(:refunded)
# Happy path
|> FSM.add_transition(:cart, :checkout, label: "begin_checkout")
|> FSM.add_transition(:checkout, :payment_pending, label: "submit_payment")
|> FSM.add_transition(:payment_pending, :processing, label: "payment_confirmed")
|> FSM.add_transition(:processing, :shipped, label: "ship")
|> FSM.add_transition(:shipped, :delivered, label: "deliver")
# Payment retry
|> FSM.add_transition(:payment_pending, :payment_failed, label: "payment_declined")
|> FSM.add_transition(:payment_failed, :payment_pending, label: "retry_payment")
# Cancellation
|> FSM.add_transition(:cart, :cancelled, label: "cancel")
|> FSM.add_transition(:checkout, :cancelled, label: "cancel")
|> FSM.add_transition(:payment_failed, :cancelled, label: "abandon")
# Refund
|> FSM.add_transition(:shipped, :refunded, label: "refund")
:okmermaid =
order_fsm
|> FSM.to_mermaid()
|> Kino.Mermaid.new()
mermaid_state_diagram =
order_fsm
|> FSM.to_mermaid(syntax: :state_diagram)
|> Kino.Mermaid.new()
dot =
order_fsm
|> FSM.to_dot()
|> Kino.VizJS.render(height: "600px")
Kino.Layout.tabs(
Graphviz: dot,
Mermaid: mermaid,
"Mermaid (State Diagram)": mermaid_state_diagram
)Before we turn this into a Finitomata module, let's verify structural soundness.
IO.puts("═══ Structural Validation ═══\n")
validation = Analysis.validate(order_fsm)
case validation do
[] -> IO.puts("✅ No structural issues found.")
issues -> Enum.each(issues, fn {level, msg} -> IO.puts(" #{level}: #{msg}") end)
end
IO.puts("\n═══ Reachability ═══\n")
reachable = Analysis.reachable_states(order_fsm)
all_states = FSM.states(order_fsm)
unreachable = all_states -- reachable
IO.puts(" Reachable: #{inspect(Enum.sort(reachable))}")
IO.puts(" Unreachable: #{inspect(unreachable)}")
IO.puts("\n═══ Dead States ═══\n")
dead = Analysis.dead_states(order_fsm)
case dead do
[] -> IO.puts(" ✅ No dead states (all states can reach a final state).")
states -> IO.puts(" ⚠️ Dead: #{inspect(states)}")
end
IO.puts("\n═══ Livelock Detection ═══\n")
livelocks = Analysis.livelock_states(order_fsm)
case livelocks do
[] -> IO.puts(" ✅ No livelock states.")
states -> IO.puts(" ⚠️ Livelock: #{inspect(states)}")
end
IO.puts("\n═══ Determinism ═══\n")
if Analysis.deterministic?(order_fsm) do
IO.puts(" ✅ FSM is deterministic.")
else
IO.puts(" ⚠️ FSM is non-deterministic (same label from same state to different targets).")
end
IO.puts("\n═══ Shortest Accepting Path ═══\n")
case Analysis.shortest_accepting_path(order_fsm) do
{:ok, path} -> IO.puts(" #{Enum.join(path, " → ")}")
:error -> IO.puts(" ❌ No accepting path found.")
endChoreo can automatically generate test paths covering all states or all transitions:
IO.puts("═══ State-Coverage Test Cases ═══\n")
state_cases = Analysis.generate_test_cases(order_fsm, :state)
state_cases
|> Enum.with_index(1)
|> Enum.each(fn {path, i} ->
IO.puts(" Test #{i}: #{Enum.join(path, " → ")}")
end)
IO.puts("\n═══ Transition-Coverage Test Cases ═══\n")
transition_cases = Analysis.generate_test_cases(order_fsm, :transition)
transition_cases
|> Enum.with_index(1)
|> Enum.each(fn {path, i} ->
IO.puts(" Test #{i}: #{Enum.join(path, " → ")}")
end)Now that the FSM is validated, generate the DSL string for Finitomata:
dsl = FinitomataAdapter.to_finitomata_dsl(order_fsm)
IO.puts(dsl)This DSL can be directly pasted into use Finitomata, fsm: ~S"""...""".
Let's take the generated DSL and use it to define a real Finitomata process.
defmodule OrderProcess do
@moduledoc "Order processing FSM powered by Finitomata."
use Finitomata,
fsm: """
cart --> |begin_checkout| checkout
cart --> |cancel| cancelled
checkout --> |cancel| cancelled
checkout --> |submit_payment| payment_pending
payment_failed --> |abandon| cancelled
payment_failed --> |retry_payment| payment_pending
payment_pending --> |payment_confirmed| processing
payment_pending --> |payment_declined| payment_failed
processing --> |ship| shipped
shipped --> |deliver| delivered
shipped --> |refund| refunded
""",
auto_terminate: true,
impl_for: :all
@impl Finitomata
def on_transition(current_state, event, _event_payload, state_data) do
updated = Map.update(state_data, :history, [{current_state, event}], &[{current_state, event} | &1])
Finitomata.Transition.guess_next_state(__config__(:fsm), current_state, event, updated)
end
endIO.puts("═══ Compiled Finitomata Config ═══\n")
IO.puts("Entry event: #{inspect(OrderProcess.__config__(:entry))}")
IO.puts("States: #{inspect(OrderProcess.__config__(:states))}")
IO.puts("Events: #{inspect(OrderProcess.__config__(:events))}")
IO.puts("\n═══ Transition List ═══\n")
OrderProcess.__config__(:fsm)
|> Enum.reject(&(&1.from == :* or &1.to == :*))
|> Enum.each(fn t ->
IO.puts(" #{t.from} --#{t.event}--> #{inspect(t.to)}")
end)# Ensure the Finitomata supervision tree is started in our Livebook session
unless Process.whereis(Finitomata.Supervisor), do: Finitomata.start_link()
# Start the happy path
Finitomata.start_fsm(OrderProcess, "order-42", %{user_id: 42, items: ["widget"]})
Process.sleep(50)
# Walk through the happy path
[:begin_checkout, :submit_payment, :payment_confirmed, :ship, :deliver]
|> Enum.each(fn event ->
Finitomata.transition("order-42", {event, nil})
Process.sleep(20)
end)
# Give auto_terminate time
Process.sleep(100)
IO.puts("✅ Order 42 completed the happy path.")# Ensure Finitomata is running
unless Process.whereis(Finitomata.Supervisor), do: Finitomata.start_link()
# Start the retry + cancel path
Finitomata.start_fsm(OrderProcess, "order-99", %{user_id: 99, items: ["gizmo"]})
Process.sleep(50)
[:begin_checkout, :submit_payment, :payment_declined, :retry_payment, :payment_declined, :abandon]
|> Enum.each(fn event ->
Finitomata.transition("order-99", {event, nil})
Process.sleep(20)
end)
Process.sleep(100)
IO.puts("✅ Order 99 completed the retry→cancel path.")Now let's demonstrate the reverse — taking a compiled Finitomata module and importing it into Choreo for analysis.
imported_fsm = FinitomataAdapter.from_finitomata(OrderProcess)
IO.puts("═══ Imported FSM ═══\n")
IO.puts("States: #{inspect(Enum.sort(FSM.states(imported_fsm)))}")
IO.puts("Initial: #{inspect(FSM.initial_state(imported_fsm))}")
IO.puts("Final: #{inspect(Enum.sort(FSM.final_states(imported_fsm)))}")
IO.puts("Transitions: #{length(FSM.transitions(imported_fsm))}")mermaid =
imported_fsm
|> FSM.to_mermaid()
|> Kino.Mermaid.new()
mermaid_state_diagram =
imported_fsm
|> FSM.to_mermaid(syntax: :state_diagram)
|> Kino.Mermaid.new()
dot =
imported_fsm
|> FSM.to_dot()
|> Kino.VizJS.render(height: "600px")
Kino.Layout.tabs(
Graphviz: dot,
Mermaid: mermaid,
"Mermaid (State Diagram)": mermaid_state_diagram
)Let's verify that the round-trip preserves the topology:
IO.puts("═══ Round-Trip Check ═══\n")
original_states = FSM.states(order_fsm) |> Enum.sort()
imported_states = FSM.states(imported_fsm) |> Enum.sort()
original_transitions =
FSM.transitions(order_fsm)
|> Enum.map(fn {from, to, label} -> {from, to, label} end)
|> Enum.sort()
imported_transitions =
FSM.transitions(imported_fsm)
|> Enum.map(fn {from, to, label} -> {from, to, label} end)
|> Enum.sort()
IO.puts("States match: #{original_states == imported_states}")
IO.puts("Transitions match: #{original_transitions == imported_transitions}")
IO.puts("Initial match: #{FSM.initial_state(order_fsm) == FSM.initial_state(imported_fsm)}")
final_match =
Enum.sort(FSM.final_states(order_fsm)) == Enum.sort(FSM.final_states(imported_fsm))
IO.puts("Finals match: #{final_match}")Here's where the real power lies — importing an FSM you didn't design in Choreo and finding issues.
defmodule BuggyWorkflow do
@moduledoc "A workflow with hidden structural issues."
use Finitomata,
fsm: """
idle --> |start| validating
validating --> |valid| processing
validating --> |invalid| rejected
processing --> |complete| done
processing --> |hang| stuck
stuck --> |recover| looping
looping --> |fail| stuck
""",
auto_terminate: true,
impl_for: :all
@impl Finitomata
def on_transition(current, event, _payload, state_data) do
Finitomata.Transition.guess_next_state(__config__(:fsm), current, event, state_data)
end
endbuggy = FinitomataAdapter.from_finitomata(BuggyWorkflow)IO.puts("═══ Analysis of BuggyWorkflow ═══\n")
IO.puts("── Validation ──")
case Analysis.validate(buggy) do
[] -> IO.puts(" ✅ No issues")
issues -> Enum.each(issues, fn {level, msg} -> IO.puts(" #{level}: #{msg}") end)
end
IO.puts("\n── Dead States ──")
dead = Analysis.dead_states(buggy)
case dead do
[] -> IO.puts(" ✅ None")
_ -> IO.puts(" ⚠️ #{inspect(dead)} — these states can never reach a final state!")
end
IO.puts("\n── Livelock ──")
livelocks = Analysis.livelock_states(buggy)
case livelocks do
[] -> IO.puts(" ✅ None")
_ -> IO.puts(" 🔄 #{inspect(livelocks)} — these states can cycle forever without reaching acceptance!")
end
IO.puts("\n── Determinism ──")
IO.puts(" #{if Analysis.deterministic?(buggy), do: "✅ Deterministic", else: "⚠️ Non-deterministic"}")
IO.puts("\n── Completeness ──")
IO.puts(" #{if Analysis.complete?(buggy), do: "✅ Complete", else: "⚠️ Incomplete — some states don't handle all events"}")
IO.puts("\n── Shortest Accepting Path ──")
case Analysis.shortest_accepting_path(buggy) do
{:ok, path} -> IO.puts(" #{Enum.join(path, " → ")}")
:error -> IO.puts(" ❌ None")
endmermaid =
buggy
|> FSM.to_mermaid()
|> Kino.Mermaid.new()
mermaid_state_diagram =
buggy
|> FSM.to_mermaid(syntax: :state_diagram)
|> Kino.Mermaid.new()
dot =
buggy
|> FSM.to_dot()
|> Kino.VizJS.render(height: "600px")
Kino.Layout.tabs(
Graphviz: dot,
Mermaid: mermaid,
"Mermaid (State Diagram)": mermaid_state_diagram
)Check if specific properties hold:
IO.puts("═══ Invariant Checks ═══\n")
# Can every state reach at least one final state?
IO.puts("All states productive: #{not Analysis.violates_invariant?(buggy, reachable_finals: true)}")
# Are there any unreachable states?
unreachable = FSM.states(buggy) -- Analysis.reachable_states(buggy)
IO.puts("Unreachable states: #{inspect(unreachable)}")Choreo can minimize an FSM by merging equivalent states, producing a smaller machine that accepts the same language.
redundant =
FSM.new()
|> FSM.add_initial_state(:s0)
|> FSM.add_state(:s1)
|> FSM.add_state(:s2)
|> FSM.add_state(:s3)
|> FSM.add_state(:s4)
|> FSM.add_final_state(:s5)
|> FSM.add_final_state(:s6)
# s1 and s2 are equivalent (same outgoing transitions to equivalent targets)
|> FSM.add_transition(:s0, :s1, label: "a")
|> FSM.add_transition(:s0, :s2, label: "b")
|> FSM.add_transition(:s1, :s3, label: "x")
|> FSM.add_transition(:s2, :s4, label: "x")
# s3,s4 → equivalent final states
|> FSM.add_transition(:s3, :s5, label: "done")
|> FSM.add_transition(:s4, :s6, label: "done")
IO.puts("Before: #{length(FSM.states(redundant))} states, #{length(FSM.transitions(redundant))} transitions")redundant
|> FSM.to_dot()
|> Kino.VizJS.render(height: "600px")minimized = Analysis.minimize(redundant)
IO.puts("After: #{length(FSM.states(minimized))} states, #{length(FSM.transitions(minimized))} transitions")
IO.puts("Equivalent? #{Analysis.equivalent?(redundant, minimized)}")minimized
|> FSM.to_dot()
|> Kino.VizJS.render(height: "600px")minimized_order = Analysis.minimize(order_fsm)
IO.puts("Order FSM: #{length(FSM.states(order_fsm))} states → #{length(FSM.states(minimized_order))} states")
IO.puts("Equivalent? #{Analysis.equivalent?(order_fsm, minimized_order)}")fsms = %{
"Order (Choreo-designed)" => order_fsm,
"Order (Finitomata-imported)" => imported_fsm,
"BuggyWorkflow (imported)" => buggy,
"Redundant (before minimize)" => redundant,
"Redundant (after minimize)" => minimized
}
header = "| FSM | States | Transitions | Deterministic? | Dead States | Livelocks | Shortest Path |"
separator = "|-----|--------|-------------|----------------|-------------|-----------|---------------|"
rows =
Enum.map(fsms, fn {name, fsm} ->
states = length(FSM.states(fsm))
transitions = length(FSM.transitions(fsm))
det = if Analysis.deterministic?(fsm), do: "✅", else: "❌"
dead = length(Analysis.dead_states(fsm))
livelocks = length(Analysis.livelock_states(fsm))
shortest =
case Analysis.shortest_accepting_path(fsm) do
{:ok, path} -> "#{length(path)} steps"
:error -> "none"
end
"| #{name} | #{states} | #{transitions} | #{det} | #{dead} | #{livelocks} | #{shortest} |"
end)
[header, separator | rows]
|> Enum.join("\n")
|> Kino.Markdown.new()Choreo can not only validate and design FSMs, but it can also serve as a live monitoring and visualization tool for running Finitomata instances.
In this section, we will:
- Spawn 50 simultaneous
OrderProcessinstances representing active user shopping sessions. - Advance them randomly to different stages of the checkout lifecycle.
- Query the live state of all running Finitomata FSMs using
Finitomata.all/1. - Aggregate state frequencies and render a live traffic heatmap of our production system using Choreo's heatmap generator.
# Ensure Finitomata is running
unless Process.whereis(Finitomata.Supervisor), do: Finitomata.start_link()
# Spawn 50 random orders
order_ids = Enum.map(1..50, fn i -> "order-sim-#{i}" end)
Enum.each(order_ids, fn id ->
Finitomata.start_fsm(OrderProcess, id, %{user_id: :rand.uniform(10_000)})
end)
# Let the FSMs initialize
Process.sleep(50)
# Simulate different user behaviors by transitioning FSMs randomly:
# - Some users stay in cart
# - Some go to checkout
# - Some complete, fail, retry, or cancel
Enum.each(order_ids, fn id ->
# Roll a die to decide how far the order goes
die = :rand.uniform(6)
cond do
die == 1 ->
# Abandon immediately
Finitomata.transition(id, {:cancel, nil})
die == 2 ->
# Stay in checkout
Finitomata.transition(id, {:begin_checkout, nil})
die == 3 ->
# Fail payment and abandon
Finitomata.transition(id, {:begin_checkout, nil})
Process.sleep(10)
Finitomata.transition(id, {:submit_payment, nil})
Process.sleep(10)
Finitomata.transition(id, {:payment_declined, nil})
Process.sleep(10)
Finitomata.transition(id, {:abandon, nil})
die == 4 ->
# Fail payment, retry, and stay pending
Finitomata.transition(id, {:begin_checkout, nil})
Process.sleep(10)
Finitomata.transition(id, {:submit_payment, nil})
Process.sleep(10)
Finitomata.transition(id, {:payment_declined, nil})
Process.sleep(10)
Finitomata.transition(id, {:retry_payment, nil})
die >= 5 ->
# Happy path (processing, shipped, or delivered)
Finitomata.transition(id, {:begin_checkout, nil})
Process.sleep(10)
Finitomata.transition(id, {:submit_payment, nil})
Process.sleep(10)
Finitomata.transition(id, {:payment_confirmed, nil})
if die == 6 do
Process.sleep(10)
Finitomata.transition(id, {:ship, nil})
if :rand.uniform(2) == 1 do
Process.sleep(10)
Finitomata.transition(id, {:deliver, nil})
end
end
end
end)
# Wait a moment for transitions to settle
Process.sleep(100)
IO.puts("Simulation finished. Live instances are resting in various states.")Now we query the registry of all active Finitomata processes, read their current state, and pass those counts to Choreo.Analysis.heatmap/2.
# Retrieve all active FSM names
active_fsms = Finitomata.all() |> Map.keys()
# Count the frequency of each state.
state_counts =
active_fsms
|> Enum.reduce(%{}, fn name, acc ->
case Finitomata.state(name) do
%Finitomata.State{current: state} ->
Map.update(acc, state, 1, &(&1 + 1))
nil ->
acc
end
end)
# Also account for auto-terminated ones (cancelled, delivered, refunded)
# by comparing total simulated (50) with active.
total_active = Enum.sum(Map.values(state_counts))
terminated_count = 50 - total_active
IO.puts("Active FSMs: #{total_active}, Auto-terminated: #{terminated_count}")
IO.inspect(state_counts, label: "Active State Frequencies")
# Render the heatmap
order_fsm
|> Choreo.Analysis.heatmap(scores: state_counts, palette: :heat)
|> Choreo.FSM.to_dot()
|> Kino.VizJS.render(height: "600px")This notebook demonstrated the full lifecycle:
- Design an FSM in Choreo with rich visualization
- Validate it with reachability, deadlock, livelock, and determinism checks
- Generate test cases automatically for state and transition coverage
- Export to Finitomata DSL and run it as a supervised OTP process
- Import an existing Finitomata module back into Choreo
- Analyze imported FSMs for structural issues you couldn't see at runtime
- Minimize redundant FSMs while preserving behavioral equivalence
- Monitor live running Finitomata workloads in real-time using Choreo heatmaps