Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,10 @@ The critical execution order is: `main_event_loop` → `microstep` → `main_eve
`setup(guards=..., actions=..., delays=..., actors=...).create_machine(config)`
- **Snapshot serialization** (0.6.0, `from xstate import serialize_snapshot, deserialize_snapshot`)
— persist and restore State; `create_actor(machine, snapshot=...)` for round-trip replay
- **State tags** (0.7.0) — declare `tags: ["loading"]` (or a single string) on any state node;
query the snapshot with `state.has_tag("loading")` / `state.hasTag(...)` or read the aggregated
`state.tags` frozenset. Tags union across the whole active configuration (compound ancestors +
parallel regions) and are recomputed from the machine definition, so snapshots stay tag-free

Handler-signature note: guards/assigners are invoked arity-aware by
`algorithm._invoke`, which supports four calling conventions: `()`, `(context)`,
Expand Down
18 changes: 18 additions & 0 deletions src/xstate/config_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,10 +105,28 @@ def _resolve_node(self, node: StateNode, *, path: str) -> None:
path=self._path(path, "target"),
)

node.tags = self._build_tags(config.get("tags"), self._path(path, "tags"))

node.donedata = self._build_output(node, path)
self._build_configured_transitions(node, path)
self._build_initial_transition(node, path)

def _build_tags(self, tags_config: Any, path: str) -> tuple[str, ...]:
if tags_config is None:
return ()
if isinstance(tags_config, str):
return (tags_config,)
if isinstance(tags_config, (list, tuple)):
if not all(isinstance(t, str) for t in tags_config):
raise InvalidConfigError(
f"{path}: every tag must be a string, got {tags_config!r}."
)
return tuple(tags_config)
raise InvalidConfigError(
f"{path}: 'tags' must be a string or a list of strings, "
f"got {type(tags_config)!r}."
)
Comment on lines +114 to +128

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Update _build_tags to return a frozenset[str] to align with the suggested frozenset type for StateNode.tags. This also deduplicates any duplicate tags provided in the configuration.

    def _build_tags(self, tags_config: Any, path: str) -> frozenset[str]:\n        if tags_config is None:\n            return frozenset()\n        if isinstance(tags_config, str):\n            return frozenset((tags_config,))\n        if isinstance(tags_config, (list, tuple, set, frozenset)):\n            if not all(isinstance(t, str) for t in tags_config):\n                raise InvalidConfigError(\n                    f\"{path}: every tag must be a string, got {tags_config!r}.\"\n                )\n            return frozenset(tags_config)\n        raise InvalidConfigError(\n            f\"{path}: 'tags' must be a string or a list of strings, \"\n            f\"got {type(tags_config)!r}.\"\n        )


def _build_output(self, node: StateNode, path: str) -> Any:
if node.type != "final":
return None
Expand Down
1 change: 1 addition & 0 deletions src/xstate/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ class StateNodeConfig(TypedDict, total=False):
target: TransitionTarget
output: Any
data: Any
tags: str | list[str]


class MachineConfig(StateNodeConfig, total=False):
Expand Down
19 changes: 19 additions & 0 deletions src/xstate/state.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,25 @@ def __init__(
self.status = "active"
self.output = None

@property
def tags(self) -> frozenset[str]:
"""Union of the ``tags`` declared on every active state node.

XState v5 surfaces ``snapshot.tags`` as the set of tags across the
current configuration; querying it is the idiomatic way to ask "is the
machine loading / busy / editable" without enumerating state values.
"""
return frozenset(
tag for node in self.configuration for tag in node.tags
)
Comment on lines +83 to +85

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If StateNode.tags is updated to be a frozenset, we can compute the union of all active tags more efficiently and cleanly using frozenset().union instead of a nested generator expression.

        return frozenset().union(*(node.tags for node in self.configuration))

Comment on lines +83 to +85

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve root-level tags across transitions

When the machine root declares tags, the initial snapshot includes the root in configuration, but Machine.transition() rebuilds the next configuration from state.value and does not re-add the root node for string or nested values. Because tags is computed only from self.configuration here, any event sent through the pure API or actor runtime drops root-level tags from state.tags/has_tag, even for ignored events. Please include the root in reconstructed configurations or account for root tags explicitly.

Useful? React with 👍 / 👎.


def has_tag(self, tag: str) -> bool:
"""Return True if any active state node declares *tag* (v5 ``hasTag``)."""
return any(tag in node.tags for node in self.configuration)

# XState v5 spells this ``hasTag``; expose both for JS-parity ergonomics.
hasTag = has_tag

def can(self, event: Any) -> bool:
"""Return True if any enabled transition exists for *event* right now.

Expand Down
1 change: 1 addition & 0 deletions src/xstate/state_node.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ class StateNode:
after: list[tuple[Any, str]] = field(default_factory=list)
invoke: list[dict[str, Any]] = field(default_factory=list)
initial_transition: Transition | None = None
tags: tuple[str, ...] = field(default_factory=tuple)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Using a frozenset[str] instead of a tuple[str, ...] for StateNode.tags is more idiomatic and efficient. It automatically deduplicates tags at the node level and allows $O(1)$ membership checks (e.g., tag in node.tags) instead of $O(N)$ scans.

Suggested change
tags: tuple[str, ...] = field(default_factory=tuple)
tags: frozenset[str] = field(default_factory=frozenset)


@property
def history_states(self) -> list[StateNode]:
Expand Down
203 changes: 203 additions & 0 deletions tests/test_tags.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,203 @@
"""Tests for state tags + has_tag()/hasTag() (0.7.0).

XState v5 lets a state declare ``tags: ["loading"]`` and query the running
snapshot with ``state.hasTag("loading")``. Tags aggregate across the whole
active configuration (compound ancestors + parallel regions).
"""

import pytest

from xstate import Machine, create_actor
from xstate.exceptions import InvalidConfigError

# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------


def test_tags_as_list():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {
"loading": {"tags": ["busy", "network"]},
"idle": {},
},
}
)
assert machine.initial_state.tags == frozenset({"busy", "network"})


def test_tags_as_single_string():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {
"loading": {"tags": "busy"},
"idle": {},
},
}
)
assert machine.initial_state.tags == frozenset({"busy"})


def test_no_tags_is_empty_frozenset():
machine = Machine(
{"id": "m", "initial": "idle", "states": {"idle": {}}}
)
assert machine.initial_state.tags == frozenset()


def test_non_string_tag_raises():
with pytest.raises(InvalidConfigError, match="every tag must be a string"):
Machine(
{
"id": "m",
"initial": "a",
"states": {"a": {"tags": ["ok", 123]}},
}
)


def test_invalid_tags_type_raises():
with pytest.raises(InvalidConfigError, match="must be a string or a list"):
Machine(
{
"id": "m",
"initial": "a",
"states": {"a": {"tags": {"not": "valid"}}},
}
)


# ---------------------------------------------------------------------------
# has_tag / hasTag query
# ---------------------------------------------------------------------------


def test_has_tag_true():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {"loading": {"tags": ["busy"]}, "idle": {}},
}
)
assert machine.initial_state.has_tag("busy") is True


def test_has_tag_false():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {"loading": {"tags": ["busy"]}, "idle": {}},
}
)
assert machine.initial_state.has_tag("idle") is False


def test_hasTag_camelcase_alias():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {"loading": {"tags": ["busy"]}, "idle": {}},
}
)
assert machine.initial_state.hasTag("busy") is True
assert machine.initial_state.hasTag("nope") is False


# ---------------------------------------------------------------------------
# Tags update across transitions
# ---------------------------------------------------------------------------


def test_tags_change_on_transition():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {
"loading": {"tags": ["busy"], "on": {"DONE": "ready"}},
"ready": {"tags": ["interactive"]},
},
}
)
state = machine.initial_state
assert state.has_tag("busy")

state = machine.transition(state, "DONE")
assert state.has_tag("interactive")
assert not state.has_tag("busy")


def test_tags_via_actor():
machine = Machine(
{
"id": "m",
"initial": "loading",
"states": {
"loading": {"tags": ["busy"], "on": {"DONE": "ready"}},
"ready": {},
},
}
)
actor = create_actor(machine).start()
assert actor.get_snapshot().has_tag("busy")
actor.send("DONE")
assert not actor.get_snapshot().has_tag("busy")


# ---------------------------------------------------------------------------
# Aggregation across compound ancestors and parallel regions
# ---------------------------------------------------------------------------


def test_tags_aggregate_from_compound_ancestor():
machine = Machine(
{
"id": "m",
"initial": "parent",
"states": {
"parent": {
"tags": ["outer"],
"initial": "child",
"states": {
"child": {"tags": ["inner"]},
},
},
},
}
)
state = machine.initial_state
# Both the active compound ancestor and the leaf contribute tags.
assert state.tags == frozenset({"outer", "inner"})
assert state.has_tag("outer")
assert state.has_tag("inner")


def test_tags_aggregate_across_parallel_regions():
machine = Machine(
{
"id": "m",
"type": "parallel",
"states": {
"a": {
"initial": "a1",
"states": {"a1": {"tags": ["region-a"]}},
},
"b": {
"initial": "b1",
"states": {"b1": {"tags": ["region-b"]}},
},
},
}
)
state = machine.initial_state
assert state.has_tag("region-a")
assert state.has_tag("region-b")
assert {"region-a", "region-b"} <= state.tags
Loading