xstate-statemachine 0.4.3
A robust Python library for parsing and running XState JSON state machines.
License: MIT
Requires Python: >=3.9

# 🚦 XState - StateMachine for Python
**The Definitive ← _Seriously_ → Guide to Bullet-Proof State Machines**

> _“A good state machine is like a map: once you have it, you’ll never get lost again.”_

---

## 📜 Table of Contents

1.  **[Introduction](#introduction)**
2.  **[Why State Machines?](#why-state-machines)**
3.  **[Event‑Driven Architecture 101](#event-driven-architecture-101)**
4.  **[What is XState-StateMachine?](#what-is-xstate-statemachine)**
5.  **[Key Features](#key-features)**
6.  **[Installation](#installation)**
7.  **[Quick Start](#quick-start)**
8.  **[The State Machine Philosophy](#the-state-machine-philosophy)**
9.  **[Visual-First Development](#visual-first-development)**
10.  **[Anatomy of an XState JSON Blueprint](#anatomy-of-an-xstate-json-blueprint)**
11.  **[States — Atomic, Compound, Parallel, Final](#states-atomic-compound-parallel-final)**
12.  **[Transitions & Events](#transitions--events)**
13.  **[Actions, Guards & Services](#actions-guards--services)**
14.  **[Context - The Machine’s Memory](#context--the-machines-memory)**
15. **[Declarative Timers (after)](#declarative-timers-after)**
16.  **[The Actor Model](#the-actor-model)**
17.  **[Architectural Patterns](#architectural-patterns)**
18.  **[Synchronous vs Asynchronous Execution](#synchronous-vs-asynchronous-execution)**
19.  **[Debugging & Visualization](#debugging--visualization)**
20.  **[CLI: Boilerplate Generation](#cli-boilerplate-generation)**
    - [Aliases & Short Flags](#cli-aliases)
    - [Hierarchical Machine Generation](#cli-hierarchy)
    - [One vs Two Files](#cli-file-count)
    - [Sync vs Async Templates](#cli-async-sync)
21.  **[API Reference](#api-reference)**
23.  **[Advanced Concepts](#advanced-concepts)**
24.  **[Best Practices](#best-practices)**
25.  **[FAQ](#faq)**
26.  **[Roadmap](#roadmap-next-release)**
27.  **[Contributing](#contributing)**
28.  **[License](#license)**
---

## 🏁 Introduction<a name="introduction"></a>

Welcome to **XState-StateMachine for Python**, a **robust**, **async-ready**, and **feature-complete** library for building, parsing, and executing **state machines** and **statecharts** defined in **XState-compatible JSON**.

Whether you’re a **junior dev** struggling with spaghetti `if/else` trees 🌱 _or_ a **senior architect** with half a century of scars and stories 🦖, this README is crafted to be your **_Bible_**. By the time you finish, you’ll know **why** state machines matter, **how** to model them visually, and **exactly** what code to write to run them in production-grade Python.

---

## ❓ Why State Machines?<a name="why-state-machines"></a>

1. **Eliminate Impossible States 🧹**
   Four boolean flags ➡ **16** possible combinations. _How many are valid?_ State machines guarantee you’ll **never** enter a “loading + error” paradox again.

2. **Explode Complexity—In a Good Way 💥**
   Complexity _will_ happen. Put it in a **graph** where it’s explicit, testable, and visualized—rather than hidden in nested conditionals.

3. **Single Source of Truth 🔑**
   Your JSON blueprint declares every state, event, and transition. **No surprises** lurking in random helper functions.

4. **Self-Documenting 📚**
   A statechart _is_ the documentation. No more stale flow-charts stuck in a Confluence graveyard.

5. **Safer Concurrency 🛡️**
   Parallel states and the Actor Model let you reason about multi-threaded logic **without** race-condition nightmares.

---

## ⚡ Event‑Driven Architecture 101<a name="event-driven-architecture-101"></a>

> “Everything is an event.” — That’s the mindset shift.

Most real systems are _reactive_: **something happens → you react**.
Event‑Driven Architecture (EDA) makes that explicit and first‑class.

### 🧠 Core Ideas

- **Emit facts, don’t call functions.** Instead of “do X then Y”, you _announce_ that “X happened”, and anyone interested can react.
- **Decoupling by default.** Producers don’t know who (or how many) consumers will handle an event.
- **Time becomes visible.** Events arrive over time; your logic becomes a timeline, not a call stack.

### 🧩 How XState-StateMachine fits EDA

| EDA Need | Library Feature | Why it helps |
|----------|-----------------|--------------|
| “Everything is an event” | `on`, `after`, `done.invoke.*`, `error.platform.*`, synthetic `entry./exit.` | One uniform pipeline; nothing is “special-cased”. |
| Deterministic reactions | Guards & Actions | Pure decisions (`guard`) vs. side effects (`action`). |
| Temporal logic | Declarative `after` timers | Timeouts, retries, heartbeats—no manual `sleep()`s. |
| Isolation & composition | Actor model (`spawn_*`) | Each actor is its own event-loop bubble. |
| Observability | Plugins & snapshots | Tap and record every heartbeat for logs/tests. |

### 🗺️ Tiny Example: “Upload finished” flow

```mermaid
sequenceDiagram
    participant UI
    participant SM as State Machine
    participant API

    UI->>SM: START_UPLOAD
    SM->>API: invoke uploadFile()
    API-->>SM: done.invoke.uploadFile (data)
    SM->>UI: SHOW_SUCCESS
```

No direct `await upload()` in UI logic; the machine emits/receives events and drives the UI.

### 📦 Example Payload Flow

```text
USER_CLICKED_SUBMIT
           │
           ▼
   [guard: isValid]  ────✗→  (event dropped)
           │
           ▼
   actions: saveDraft → invoke apiCall
                          │
                          ├── done.invoke.apiCall  → actions: cache, target: success
                          └── error.platform.apiCall → actions: setError, target: failure
```

### ✅ Why developers love it

- Clear “When X happens, go to Y” rules—no hidden callbacks.
- Visual diagrams match exactly what runs.
- Tests can fire events in order and assert outcomes—no flaky timing.

> **Takeaway:** State machines are a great _way_ to **implement** EDA.
> Events, timers, and actors are the primitives; this library wires them up so you don’t have to.

---

## ✨ What is XState-StateMachine?<a name="what-is-xstate-statemachine"></a>

* A **Pythonic** runtime for the world-famous **[XState](https://stately.ai)** architecture.
* **100 % JSON-spec-compatible**, so you can design your chart in the Stately editor and run it untouched.
* **Async first** (`asyncio`), yet ships a **fully-featured, blocking** `SyncInterpreter` for CLI tools or tests, complete with support for `after` timers and a threaded **actor model**.
* Packed with goodies: **hierarchy, parallelism, invoke**, **after**, timers, **actors**, auto-binding logic loaders, plugin hooks, diagram generators, and more.

> **TL;DR** — If you know XState in JS, everything 👉 “just works” in Python.
> If you don’t, keep reading—this guide is for you.

---

## 🚀 Key Features<a name="key-features"></a>

| Feature | Why You Care | Best For |
| :--- | :--- | :--- |
| **100% XState JSON Compatibility** | Design visually, export JSON, run in Python. | Teams that collaborate in the Stately Editor. |
| **Async & Sync Interpreters** | One config, two runtimes (asyncio or blocking). | Apps that need to run in CLI scripts *and* async servers. |
| **Multiple `after` Timers per State** | Fine-grained timing logic without hacks. | Complex timeout/backoff flows in synchronous environments. |
| **True Actor Model (spawn_* actions)** | Compose systems from smaller machines; isolate concurrency. | IoT fleets, game entities, micro‑workflows. |
| **Hierarchical & Parallel States** | Model nested flows or concurrent regions explicitly. | UIs, wizards, orchestration logic. |
| **Declarative `invoke` Services** | Async tasks with automatic cancellation & `onDone`/`onError`. | API calls, DB work, long-running jobs. |
| **Smart CLI Code Generator (`xsm`)** | Zero-boilerplate logic/runner stubs from JSON. | Fast scaffolding, consistent team patterns. |
| **Auto Logic Binding (LogicLoader)** | Drop in modules/classes; names are auto-wired. | Big projects with many logic files. |
| **Deep Plugin Hooks** | Observe guards, services, transitions in detail. | Telemetry, tracing, testing. |
| **Snapshots & Restoration** | Time-travel debugging, crash recovery. | Long-lived workflows, CI golden tests. |
| **Diagram Exporters** | Mermaid / PlantUML from code. | Up-to-date docs in CI/CD. |

> 🔥 **Powerful Features**
> *SyncInterpreter:* Non‑blocking actor spawning in threads, multiple concurrent `after` timers per state, and a cleaner shutdown sequence.
> *CLI super-powers:* `xsm`, subcommand aliases, interactive parent detection, stricter arg parsing, and more robust generated runners.
---

## 🛠️ Installation<a name="installation"></a>

```bash
# 1️⃣ Create & activate a virtual env  (recommended)
python -m venv venv
source venv/bin/activate           # Windows → venv\Scripts\activate

# 2️⃣ Install the library
pip install xstate-statemachine
```

> **Requirements**: Python 3.9 +

---

## ⚡ Quick Start<a name="quick-start"></a>

A lightspeed tour: toggle a light 💡—the “Hello World” of state machines.

### 1. Blueprint (`light_switch.json`)

```jsonc
{
  "id": "lightSwitch",
  "initial": "off",
  "context": {
    "flips": 0
  },
  "states": {
    "off": {
      "on": {
        "TOGGLE": {
          "target": "on",
          "actions": "increment_flips"
        }
      }
    },
    "on": {
      "on": {
        "TOGGLE": {
          "target": "off",
          "actions": "increment_flips"
        }
      }
    }
  }
}
```

### 2. Logic (`light_switch_logic.py`)

```python
import logging
from typing import Dict
from xstate_statemachine import Interpreter, Event, ActionDefinition

def increment_flips(i: Interpreter, ctx: Dict, e: Event, a: ActionDefinition):
    ctx["flips"] += 1
    logging.info(f"🔀 Switch flipped {ctx['flips']} time(s).")
```

### 3. Runner (`main.py`)

```python
import asyncio
import json
import light_switch_logic # Use a standard import
from xstate_statemachine import create_machine, Interpreter

async def main():
    with open("light_switch.json") as f:
        config = json.load(f)

    # Pass the imported module object directly for auto-discovery. This is cleaner.
    machine = create_machine(config, logic_modules=[light_switch_logic])

    interpreter = await Interpreter(machine).start()
    await interpreter.send("TOGGLE")
    await interpreter.send("TOGGLE")
    await interpreter.stop()

asyncio.run(main())
```

**Output**

```
INFO 🔀 Switch flipped 1 time(s).
INFO 🔀 Switch flipped 2 time(s).
```

Boom—**no `if current_state == "on"` anywhere**. 🎉

#

### A Very Simple Sync Example: The Toggle Switch

Here is the most basic example of a synchronous state machine. It has only two states (`on` and `off`) and one event (`TOGGLE`). It perfectly illustrates how the `SyncInterpreter` processes events immediately.

#### 1. The Blueprint: `toggle_switch.json`

This JSON defines the structure. It starts `off`, and the `TOGGLE` event switches it to `on` (and vice-versa), running an action called `increment_toggles` each time.

```json
{
  "id": "toggleSwitch",
  "initial": "off",
  "context": {
    "toggleCount": 0
  },
  "states": {
    "off": {
      "on": {
        "TOGGLE": {
          "target": "on",
          "actions": [
            "increment_toggles"
          ]
        }
      }
    },
    "on": {
      "on": {
        "TOGGLE": {
          "target": "off",
          "actions": [
            "increment_toggles"
          ]
        }
      }
    }
  }
}
```

#### 2. The Logic and Runner: `main_sync.py`

For maximum simplicity, we'll define the logic and the simulation in the same file.

```python
import json
import logging
from typing import Dict, Any

from xstate_statemachine import (
    create_machine,
    SyncInterpreter,
    MachineLogic,
    Event,
    ActionDefinition
)

# --- Basic Setup ---
logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s")

# --- 1. The Logic (Action) ---
# This is the single action our machine will execute.
def increment_toggles(i: SyncInterpreter, ctx: Dict, e: Event, a: ActionDefinition) -> None:
    """Action to increment the toggle count in the context."""
    ctx["toggleCount"] += 1
    logging.info(f"💡 Light is now {i.current_state_ids}. Toggle count: {ctx['toggleCount']}")

# --- 2. The Simulation ---
def run_simple_toggle():
    """Initializes and runs the toggle switch simulation."""
    print("\n--- 💡 Simple Synchronous Toggle Switch ---")

    # Load the machine configuration from the JSON file
    with open("toggle_switch.json", "r") as f:
        config = json.load(f)

    # Explicitly bind the action name "increment_toggles" from the JSON
    # to our Python function.
    logic = MachineLogic(
        actions={"increment_toggles": increment_toggles}
    )

    # Create the machine and the synchronous interpreter
    machine = create_machine(config, logic=logic)
    interpreter = SyncInterpreter(machine)

    # Start the machine. It enters the 'off' state.
    interpreter.start()
    logging.info(f"Initial state: {interpreter.current_state_ids}")

    # Send the first event. The machine transitions to 'on'
    # and the 'increment_toggles' action runs before this line finishes.
    print("\n--- Toggling ON ---")
    interpreter.send("TOGGLE")

    # Send the second event. The machine transitions back to 'off'
    # and the action runs again.
    print("\n--- Toggling OFF ---")
    interpreter.send("TOGGLE")

    # Stop the machine
    interpreter.stop()
    print("\n--- ✅ Simulation Complete ---")

if __name__ == "__main__":
    run_simple_toggle()
```

#### Expected Output

When you run `main_sync.py`, you will see the following output, demonstrating that each `send` call completes its work before the next print statement is executed:

```text
--- 💡 Simple Synchronous Toggle Switch ---
[INFO] Initial state: {'toggleSwitch.off'}

--- Toggling ON ---
[INFO] 💡 Light is now {'toggleSwitch.on'}. Toggle count: 1

--- Toggling OFF ---
[INFO] 💡 Light is now {'toggleSwitch.off'}. Toggle count: 2

--- ✅ Simulation Complete ---
```

---

## 🧭 The State Machine Philosophy<a name="the-state-machine-philosophy"></a>

> **Definition ↔ Implementation** separation is the super-power.

* **Definition** (`.json`) 😇 — declares _what can happen_.
  * Finite states
  * Events
  * Valid transitions
  * Timers, services, hierarchy

* **Implementation** (`.py`) 🛠️ — implements _how it happens_.
  * Fetch an API
  * Write a file
  * Update UI

Because the **graph never mutates**, every team-mate sees the same reality. Changing business rules is as easy as editing JSON and re-running tests—**logic stays untouched**.

---

## 🎨 Visual-First Development<a name="visual-first-development"></a>



1. **Design** in the **Stately Editor** → drag states, draw arrows.
2. **Export** to JSON (one click).
3. **Run** with `create_machine(config)` in Python.
4. **Simulate** inside Stately _or_ via Python tests—they behave identically.

### Why It Rocks:

* **Stakeholder Friendly** — Product managers & QA can _read_ and _play_ with the diagram.
* **Zero Drift** — Diagram **is** the code. Update one, you update both.
* **Faster On-Boarding** — New hires grok the flow in minutes, not days.

---


### 🖼️ Designing with Stately Visual Editor<a name="stately-visual-editor"></a>

> “If a picture is worth a thousand words, a *statechart* is worth a thousand **unit-tests**.”

The **[Stately Visual Editor](https://stately.ai/editor) →** is the *single most productive tool* in the XState ecosystem.
It lets you **draw** your machine, **simulate** it in real‑time, and **export** a perfectly–valid JSON blueprint that runs *unchanged* in `xstate‑statemachine` for Python.

### 🔑 Why the Editor Matters

| Benefit | What it Means for You |
|---------|-----------------------|
| **WYSIWYG Modelling** | Drag‑and‑drop states, draw transitions, and tweak guards—no JSON eye‑strain. |
| **Instant Simulation** | Play events, watch timers fire, and inspect context mutations live before writing Python code. |
| **Team Collaboration** | Share a link; PMs and QA can *see* the flow and leave comments, killing a whole back‑and‑forth thread of screenshots. |
| **Source of Truth** | The exported JSON *is* the code—zero drift between docs and implementation. |
| **Version Control Friendly** | Download the JSON (or embed it in your repo) and diff it like any other text asset. |

### 🛠️ Typical Workflow

1. **Sketch** the high‑level flow on a whiteboard (or directly in the editor).
2. **Model** it in the Stately Editor:
   * Add *states* (double‑click canvas)
   * Create *transitions* (drag arrow from one state to another)
   * Configure *events*, *guards*, *actions*, *after* timers, and *invoke* services in the side‑panel UI.
3. **Simulate**: hit ▶️, dispatch events, and watch the visual debugger update in real time.
4. **Export** → **“Machine JSON”** (⚙️ menu → *Export → Machine JSON*).
   Save as `my_machine.json` in your project’s `statecharts/` folder.
5. **Run** it with:

   ```python
   from xstate_statemachine import create_machine, Interpreter
   import json, asyncio

   cfg = json.load(open("statecharts/my_machine.json"))
   machine = create_machine(cfg, logic_modules=[my_logic])
   await Interpreter(machine).start()
   ```

6. **Iterate**: tweak the diagram, re‑export, rerun tests. Your Python logic remains untouched.

### 🎨 Pro Tips for Power Users

| Tip | Shortcut / Action |
|-----|-------------------|
| **Multi‑Select** | <kbd>Shift</kbd> + Click or drag marquee to move/align groups of states. |
| **Quick Transition** | Hold <kbd>A</kbd>, click source state, then click target state. |
| **Relative Targets** | Double‑click a self‑transition arrow to toggle between **internal** and **external** (re‑entry) semantics. |
| **Context Visualisation** | In the *Simulate* tab, expand **Context** to live‑edit values while the machine is running—great for guard testing. |
| **Export Diagram** | *Export → SVG / PNG* to embed in GitHub docs; keep diagrams and JSON in‑sync ✨. |
| **Embed Gist** | Publish the machine as a sharable, *live* gist you can link in PR descriptions. |


---

## 🧬 Anatomy of an XState JSON Blueprint<a name="anatomy-of-an-xstate-json-blueprint"></a>

Every machine is a tree of **StateNodes**. Let’s break down the top-level keys:

| Key | Type | Description |
|---|---|---|
| `id` | **string** | Unique machine ID, root of every absolute state path. |
| `initial` | **string** | State where the interpreter starts. |
| `context` | **object** | Mutable “memory” available to every action/guard. |
| `states` | **object** | Map of state → StateNode definition. |
| `on` | **object** | _Global_ event handlers (catch-all). |

Example from your files: The `flightBooking.json` machine has a root `id` of `"flightBooking"`.

Example from your files: The `ciCdPipeline.json` defines an initial `context` to track the state of a deployment process:
```jsonc
"context": {
  "build_artifact": null,
  "commit_hash": null,
  "deployment_url": null,
  "error": null,
  "scan_results": null,
  "test_results": null
}
```

### Target Resolution: How the Machine Finds the Next State

When you define a transition with a `target`, the library uses a powerful resolution mechanism to find the destination state. This allows for flexible and intuitive state navigation.

-   **Sibling State (Most Common):** If you provide a simple name, the interpreter looks for a state with that name within the *same parent*.

    ```json
    "green": {
      "on": { "TIMER": "yellow" } // Looks for "yellow" alongside "green"
    },
    "yellow": {},
    "red": {}
    ```

-   **Child State (Dot Notation):** To target a descendant state, use dot notation.

    ```json
    "on": {
      "GO_TO_DEEP_STATE": "parent.child.grandchild"
    }
    ```

-   **Relative Path (Leading Dot):** A leading dot (`.`) makes the path relative to the *parent* of the current state. This is extremely useful for sibling-to-sibling transitions inside a compound state.

    ```json
    "parent": {
      "initial": "child1",
      "states": {
        "child1": { "on": { "NEXT": ".child2" } }, // Correctly targets parent.child2
        "child2": {}
      }
    }
    ```

-   **Absolute Path (Leading Hash):** A leading hash (`#`) makes the path absolute from the root of the machine, using the machine's `id`. This is the safest way to target a state from a deeply nested location without ambiguity.

    ```json
    "deeply": {
      "nested": {
        "state": {
          "on": {
            "GO_HOME": "#myMachine.idle" // Always goes to the top-level idle state
          }
        }
      }
    }
    ```

#### 🛣️ Edge‑Cases & Safety Nets

| Target Syntax | Resolved To | When to Use |
|---------------|------------|-------------|
| `"."` | **Parent state** of the current state &nbsp;▸ if already at the root, it resolves to the **root itself**. | Jump back one level without hard‑coding the parent’s ID. Handy inside deeply nested compound states. |
| `a..b` &nbsp;or&nbsp; `state.` | **❌ Invalid** — the library rejects any target that contains empty path segments (double dots `..` or a trailing dot `.`) and raises **`StateNotFoundError`**. | Typos happen! The explicit error prevents silent mis‑navigation and keeps your diagrams truthful. |

> 💡 **Tip:**  When debugging a mysterious *StateNotFoundError*, check for an accidental double‑dot or dangling dot in your `target` strings.


---

## 🏛️ States — Atomic • Compound • Parallel • Final <a name="states-atomic-compound-parallel-final"></a>

### 1️⃣ Atomic

> **No children.** Think “leaf node.”

```jsonc
"idle": {
  "on": { "FETCH": "loading" }
}
```

### 2️⃣ Compound

> **Has its own `initial` + `states`.**

Example from your files: In installWizard.json, the "configuring" state is a compound state that contains its own sub-machine for handling the configuration steps.

```jsonc
"configuring": {
  "initial": "network",
  "onDone": "installing",
  "states": {
    "network": {
      "on": {
        "SUBMIT_NETWORK": "database"
      }
    },
    "database": {
      "on": {
        "SUBMIT_DATABASE": "admin_user"
      }
    },
    "admin_user": {
      "on": {
        "SUBMIT_ADMIN": "config_complete"
      }
    },
    "config_complete": {
      "type": "final"
    }
  }
}
```

### 3️⃣ Parallel

> **All child regions active at once.** Perfect for independent subsystems.

Example from your files: The smartHome.json machine uses a parallel state at its root to manage lighting, climate, and security as independent, concurrent regions.

```jsonc
"dashboard": {
  "type": "parallel",
  "states": {
    "notifications": {
      /* handles toasts */
    },
    "socket": {
      /* websocket connect/close */
    },
    "theme": {
      /* dark / light */
    }
  }
}
```

### 4️⃣ Final

> **Signals completion** to parent; triggers `onDone` transition.

```jsonc
"success": { "type": "final" }
```

---

## 🔄 Transitions & Events<a name="transitions--events"></a>

### 🌊 Event Lifecycle & Synthetic Events<a name="event-lifecycle"></a>

A state machine **lives and breathes events**.
Besides the ones *you* dispatch, the runtime forges its own messages and even
loops internal “always” transitions until the graph stabilises.
Understanding these moving parts lets you write bullet‑proof tests, guards and
plugins. 🔍

---

#### 1️⃣ `.send()` – One API, Three Input Flavours

| What you call | What the helper returns | Notes |
|---------------|------------------------|-------|
| `service.send("CLICK", x=1)` | `Event(type="CLICK", payload={"x": 1})` | Snack‑size syntax |
| `service.send({"type": "CLICK", "x": 1})` | ditto | Handy when forwarding raw JSON |
| `service.send(Event("CLICK", {"x": 1}))` | *unchanged* | Already a proper `Event` |

`BaseInterpreter._prepare_event` does the coercion, so every path into the
interpreter is **consistent & type‑safe**. ✔️

---

#### 2️⃣ Runtime‑Generated (Synthetic) Events

| Pattern | ✨ When it fires | Typical purpose |
|---------|-----------------|-----------------|
| `entry.<stateId>` | Right before a state’s **entry actions** run | Side‑effect hooks, analytics |
| `exit.<stateId>` | After **exit actions** finish | Cleanup metrics, audit |
| `___xstate_statemachine_init___` | Once, at machine start‑up | Kick‑start transient guards |
| `after.<delay>.<stateId>` | `after { "<delay>": … }` timer expires | Declarative timeouts / polling |
| `done.state.<stateId>` | A **compound / parallel** state reaches all its finals | Bubble completion upward |
| `done.invoke.<src>` | An **invoked service** returns successfully | Happy‑path transitions |
| `error.platform.<src>` | Invoked service raised / rejected | Failure branch |

Because they are regular events you can:

```python
await interp.send("after.5000.flightBooking.loading")   # force timeout in tests
plugin.on_event_received = lambda _, e: print(e.type)
```
#### Quick Example: `onDone` from a Parallel Parent

```jsonc
"checkout": {
  "type": "parallel",
  "states": {
    "shipping": { "type": "final" },
    "billing":  { "type": "final" }
  },
  "onDone": "receipt"  // fires when BOTH regions hit their finals
}

```

---

#### 3️⃣ Transient (“Always”) Transitions `""`

An empty‑string event (`""`) models *automatic* logic that should run
**immediately** after a state becomes active:

```jsonc
"checking": {
  "on": {
    "": [
      {
        "guard": "isValid",
        "target": "approved"
      },
      {
        "target": "rejected"
      }
    ]
  }
}
```

* Both interpreters keep looping
  `while optimal_transition.event == "": …`
  (`Interpreter._run_event_loop`, `SyncInterpreter._process_transient_transitions`)
  until **no** guard passes.
* Guards must be **pure & synchronous** — they run potentially many times per
  event cycle.
* Great for conditional redirects, validation gates and hierarchical
  “initial” logic.

---

📝 **Cheat‑sheet**

```text
You           Engine                          What you observe
──────────── ─────────────────────────────── ─────────────────────────────────
service.send("CLICK")      ─────▶  CLICK
state entry                ─────▶  entry.someState
timer 2 s later            ─────▶  after.2000.someState
service success            ─────▶  done.invoke.fetchData
compound finished          ─────▶  done.state.parent.compound
```

> **Clarification — arrows show the *event type*, not the payload**
> In the cheat‑sheet diagram, `service.send("CLICK", x=1)  ─────▶  CLICK` means the runtime receives an event whose **type** is `"CLICK"`. Any payload you pass (e.g. `x=1`) still travels inside the event object (`Event.type == "CLICK"`, `Event.payload == {"x": 1}`); it’s just omitted for brevity.


Now you can assert, spy and debug every heartbeat of your machine. 🎉

* **Event-driven** — under `on`.
* **Time-driven** — under `after`.
* **Done/Error** — from invoke services → auto-events `done.invoke.<src>` & `error.platform.<src>`.

```jsonc
"loading": {
  "invoke": {
    "src": "fetchData",
    "onDone": {
      "actions": "cacheData",
      "target": "success"
    },
    "onError": {
      "actions": "setError",
      "target": "failure"
    }
  },
  "after": {
    "5000": {
      "target": "timeout"
    }
    // If it hangs for 5 s
  }
}
```

#### 📨 How service results travel back into the machine

When an **invoked service** (sync *or* async) finishes **successfully**,
its *return value* is baked into a **`DoneEvent`**:

```
type =  "done.invoke.<serviceId>"
data =  <return value of your Python function>
```

That payload is now available to **guards** & **actions** via
`event.data` — use it to make decisions or stash the result in context.

```python
# guards.py
def is_valid_response(ctx, event):
    # event.data is whatever the service returned 🙌
    return event.data.get("status") == 200

# actions.py
def store_payload(i, ctx, event, a):
    ctx["payload"] = event.data["body"]
```

```jsonc
"loading": {
  "invoke": {
    "src": "fetchData",
    "onDone": [
      { "target": "success", "guard": "is_valid_response", "actions": "store_payload" },
      { "target": "failure" }
    ],
    "onError": "failure"
  }
}
```

> 🔎 **Tip:** In unit tests you can *stub* the service to return a canned
> object and assert that `ctx["payload"]` matches it, without hitting the
> network. Fast & deterministic! 🧪


### Internal vs External Transitions & The `reenter` Flag<a name="internal-vs-external-transitions"></a>

When an event matches a transition, the library distinguishes between two types: **internal** and **external**.

-   **Internal Transition**: An internal transition only executes its actions and does not exit or re-enter the current state. This is useful for updating context without resetting the state's timers or invoked services. This occurs when a transition has **no `target`**.

-   **External Transition**: An external transition will exit the current state (running exit actions, cancelling timers/services) and re-enter a new state (or the same one), running all entry actions. This occurs whenever a transition **has a `target`**.

#### The `reenter` Flag for Self-Transitions

By default, a self-transition (where the `target` is the same as the source state) is **internal**. However, you can force it to be **external** by adding `"reenter": true`. This provides explicit control, aligning with XState v5.

| Transition Config | Behavior | Use Case |
| :--- | :--- | :--- |
| `{"actions": "update_ctx"}` | **Internal** (no `target`) | Update context without resetting the state. |
| `{"target": "myState"}` | **Internal** (self-target, `reenter` is `false` by default) | Same as above. The machine remains in `myState` without exit/entry actions. |
| `{"target": "myState", "reenter": true}` | **External** (self-target, explicit `reenter`) | Force a full exit/re-entry of `myState` to reset its timers or re-run entry actions. |

**Example:**

Consider a state that needs to reset an inactivity timer whenever the user performs an action.

```jsonc
"active": {
  "initial": "idle",
  "states": {
    "idle": {
      "after": {
        "5000": { "target": "#myMachine.timedOut" }
      },
      "on": {
        "USER_ACTIVITY": {
          "target": "idle",
          "reenter": true // This forces an exit/re-entry, resetting the 5s timer
        }
      }
    }
  }
}
```

---

## 🛠️ Actions, Guards & Services<a name="actions-guards--services"></a>

#### ✅ Good – deterministic & side‑effect‑free

```python
def can_retry(ctx, event) -> bool:
    return ctx["attempts"] < 3
```

#### 🚫 Bad – asynchronous

```python
async def remote_rule(ctx, event):
    result = await fetch_flag()            # blocking the event loop = 💥
    return result == "ALLOW"
```

#### 🚫 Bad – non‑boolean return

```python
def non_bool(ctx, event):
    return "yes" if ctx["foo"] else ""     # Truthy string! ⚠️
```

While `"yes"` passes today, **explicit `True`/`False` is required** for
readability and future compatibility.

---

##### Tip 💡 – Keep Them 100 % Pure

* **No logging** inside guards – use an action instead.
* **No mutation** of `ctx`. Guards run many times (transients!), so mutating
  state here creates elusive bugs.

---

Now your transitions obey the *Law of Least Surprise*: one question, one crisp
answer, synchronously. 🏁

| Kind | When Runs | Signature | Return |
|------|-----------|-----------|--------|
| **Action** | On entry/exit/transition | `(interp, ctx, event, action_def)` | `None` |
| **Guard**  | Before transition decision | `(ctx, event)` | `bool` |
| **Service**| Inside `invoke` (async or sync) | `(interp, ctx, event)` | `value` or **raise** |


### 🔥 Error Handling (Actions / Guards / Services)

**TL;DR**

| Component | If it raises… | What the interpreter does | How to observe / recover |
|-----------|----------------|---------------------------|---------------------------|
| **Guard** | Any `Exception` | Treated as **False** (transition candidate fails), error is logged; the interpreter keeps evaluating the rest of the transitions. If none pass, the event is dropped. | Use `LoggingInspector` or a custom plugin (`on_guard_evaluated`) to surface failures. |
| **Action** | Any `Exception` | Logs the error, **skips the remaining actions** in that list, continues processing. In `SyncInterpreter` the exception is re‑raised to the caller of `.send()`. In `Interpreter` it’s captured inside the event-loop task (logged, does not crash the loop). | Wrap `.send()` in `try/except` for sync; add a plugin or await `send()` and inspect logs for async. |
| **Service (`invoke`)** | Exception / rejection | Fires the configured `onError` transition with `event.data = error`; if no handler exists, the error bubbles to the parent (or is logged). | Handle `onError` branches explicitly; inspect `event.data` for details. |


```python
# Guard example: an exception becomes "False", not a crash
def might_blow_up(ctx, event):
    return 1 / ctx["divisor"] > 0   # ZeroDivisionError → guard == False

# Action example: catching sync errors
try:
    sync_service.send("SAVE")
except Exception as exc:
    logger.error("SAVE failed: %s", exc)

```

> 🐍 **Snake → Camel Autowiring** 🐫
> The loader automatically converts **`snake_case`** Python function names
> to **`camelCase`** keys expected in your JSON.
> No manual mapping needed — simply define:
>
> ```python
> def increment_flips(i, ctx, e, a): ...
> ```
>
> …and reference it in JSON as:
>
> ```jsonc
> "actions": "incrementFlips"
> ```
>
> The helper `logic_loader._snake_to_camel()` does the heavy lifting, so you
> stay idiomatic in Python **and** compliant with XState’s camel‑cased world. ✨


> ℹ️ Automatic Logic Discovery binds JSON names to Python callables **by convention** (`snake_case ⇌ camelCase`). Anything unmatched raises `ImplementationMissingError`.

---

## 🧠 Context - The Machine’s Memory<a name="context--the-machines-memory"></a>

A plain dict **shared** across all states. Mutate it inside **actions** and **services**; read-only in **guards**.

```jsonc
"context": {
  "retryCount": 0,
  "payload": null,
  "error": null
}
```

```python
def increment_retry(i, ctx, e, a):
    ctx["retryCount"] += 1
```

Rule of thumb 🧘: **Pure guards** & **deterministic actions** → easier tests.

---

## ⏰ Declarative Timers (`after`)<a name="declarative-timers-after"></a>

> **Key idea:** you declare **_intent_**, not implementation.
> The interpreter owns the stopwatch so *you* don’t have to.

| What you write | What the engine does |
|----------------|----------------------|
| `after: { "500": "blink" }` | Starts a 500 ms countdown *every* time the state is entered |
| `after: { "1000": { "target": "retry", "actions": "backoff" } }` | Schedules an internal event, then cancels it automatically if the state exits early |
| `after: { "…": { …, "guard": "stillRelevant" } }` | Evaluates guard *right before* firing—handy for stale timers |

> **SyncInterpreter parity:**
> Everything you see here—including multiple `after` entries per state—works in the synchronous engine.
> Each `after` gets a dedicated thread-backed timer and a unique ID for cancellation. No extra code from you.

> **SyncInterpreter parity addendum – guard timing:**
> In `SyncInterpreter`, exactly like in the async `Interpreter`, any `guard` on an `after` transition is evaluated **when the timer fires**, *not* when it’s scheduled on state entry.
> If the guard returns `False` at fire time, that particular delayed transition is skipped (other `after` entries can still fire), and no state change occurs.

```jsonc
"retrying": {
  "after": {
    "1000": { "target": "fetching", "guard": "fewAttempts" },
    "5000": { "target": "abandon" }
  }
}
```
```python
def fewAttempts(ctx, event):
    # evaluated only when the 1000ms timer actually fires
    return ctx["attempts"] < 3
```

### Anatomy of an `after` Event

```
after(<delay>)#<absoluteStateId>
└──── ─────── ── ───────────────
   |     |          |
   |     |          ↳ state that was active when timer started
   |     |
   |     ↳ milliseconds (integer)
   |
   ↳ literal string “after”
```

Knowing the auto‑generated `type` helps if you need to **assert** or **spy** on timers in tests:

```python
await interp.send("START_LOAD")
await asyncio.sleep(0.01)                  # Allow event loop to process
assert "loading" in interp.current_state_ids

# Fast‑forward virtual time by monkey‑patching loop.time() — or simpler:
await asyncio.sleep(5.1)                   # let after(5000) fire
assert "timeout" in interp.current_state_ids
```

### ⚡ Multiple Timers in One State

Attach *several* `after` clauses—think **progressive back‑off**:

```jsonc
"retrying": {
  "entry": "incAttempts",
  "after": {
    "1000": {
      "target": "fetching",
      "guard": "fewAttempts"
    },
    "5000": {
      "target": "giveUp",
      "actions": "alertOps"
    }
  }
}
```

If `fewAttempts` returns **false**, the 1‑second timer is ignored; the 5‑second timer is still pending.

### ⏱️ Looping Timers (Self‑Transition)

```jsonc
"flashing": {
  "after": {
    "250": {
      "actions": "toggleLED",
      "target": ".flashing"
    }
  }
}
```

Because the target is **relative** (`.flashing`), the state re‑enters itself, creating a persistent 250 ms blink. The interpreter guarantees only **one live timer** at a time—previous handles are cancelled on re‑entry.

### Cancellation & Clean‑Up

Leaving a state *always* cancels its timers.

* No memory leaks
* No stray events after the user navigates away
* Predictable “at‑most‑once” semantics

### Testing Timers 🧪

1. **Unit** – patch the event loop clock (`pytest‑asyncio`’s `advance_time`) and assert the synthetic event.
2. **Integration** – run under a virtual loop with controlled time.
3. **Snapshots** – diff context before & after the timer fires.

### Common Pitfalls & Remedies

| Pitfall | Symptom | Remedy |
|---------|---------|--------|
| Timer never fires | Guard returns `False`; or state changed before delay | Log guard or use `LoggingInspector` |
| Fires twice | Re‑entering state via **different absolute ID** in hierarchy | Use absolute target (`"#machine.state"`) or guard |
| Delay starts late | Long CPU loop in entry action | Make actions async & yield `await asyncio.sleep(0)` |

#

#### 🗂️ TaskManager – Zero‑Leak Guarantees<a name="taskmanager"></a>

Every **`after` timer** ⏱️ and each **`invoke` service** 📞 is tracked by the TaskManager: the async `Interpreter` wraps them in `asyncio.Task`, while the `SyncInterpreter` uses `threading.Timer`/`threading.Thread`; both are **registered per‑state** and cancelled on exit.


```text
stateId ─┬─ after‑5000 timer        ─┐
         ├─ after‑10000 timer  ──────┤───▶ TaskManager.add(owner_id, task)
         └─ invoke.fetchData task ───┘
```

Why it matters:

| ⭐ Benefit | How it works |
|-----------|--------------|
| **No orphaned coroutines** | When a state exits, the interpreter calls **`TaskManager.cancel_by_owner(state.id)`**, which iterates over every recorded task, `task.cancel()`s them