Wunderspec syntax (Python DSL)

Basics

Literals & imports

from wunderspec import *

Import all definitions

Val(False) Val(True)

Boolean literals; Python bool auto-coerced in expressions

Val(234) Val(0xabcdef)

Integer literals; Python int auto-coerced in expressions

Val("hello")

String literal; Python str auto-coerced in expressions

Val(MyEnum.Foo)

Enum literal; enum member auto-coerced in expressions

Var("x", int) Var("s", MyRecord)

Symbolic variable of the given type/sort

Types / Sorts

Used in StateVar[T], Param[T], Field[T], Variant[T] hints

bool, int, str

Booleans, big integers, strings

MyEnum

Python enum.Enum subclass

MyRecord

Class decorated with @record

MyUnion

Class decorated with @union

tuple[A, B, C]

Tuple of the given element types

set[E]

Set with elements of type E

dict[K, V]

Map from keys of type K to values of type V

list[E]

List with elements of type E

Booleans

BoolExpr

And(p, q) or p & q or p.and_(q)

Logical conjunction

Or(p, q) or p | q or p.or_(q)

Logical disjunction

Not(p) or ~p

Logical negation

Implies(p, q) or p.implies(q)

Implication: p → q

p == q or p != q

Boolean equality / inequality

Ite(cond, then_expr, else_expr) or p.if_(cond).else_(q)

If-then-else expression; all three arguments are expressions

Integers

IntExpr

x + y x - y x * y x / y x % y x ** y

Arithmetic operators; division is integer division

x == y x != y x < y x <= y x > y x >= y

Comparison operators; return BoolExpr

Set(lo, ..., hi)

Inclusive integer set {lo, …, hi}; distinct from list Range(start, end)

Ints or UnsignedInts

All signed / unsigned integers (infinite sets)

Sets

Immutable; may be infinite

Set(2, 5, 8, 8)

New set with the given elements, duplicates ignored

Set(lo, ..., hi)

Inclusive integer set {lo, …, hi}

Set(E)

Empty set whose elements have sort E

S.contains(x) or x.in_(S)

True if x ∈ S

S | T or S.union(T)

Union: elements in S or T

S & T or S.intersect(T)

Intersection: elements in both

S - T or S.difference(T)

Difference: in S but not in T

S <= T or S.issubset(T)

True if S ⊆ T

S < T S >= T S > T

Proper subset, superset, and proper superset predicates

S.exists(lambda x: P(x))

True if some element satisfies P

S.forall(lambda x: P(x))

True if every element satisfies P

S.filter(lambda x: P(x))

New set: elements satisfying P

S.map(lambda x: f(x))

New set: image of S under f

S.map_to(lambda x: f(x))

New map binding each x ∈ S to f(x)

S.reduce(lambda acc, x: acc + x, i)

Fold over elements in some order; i is the initial value

S.choose(lambda x: P(x))

Pick one element satisfying P (deterministic)

S.size

Cardinality (finite sets only)

S.is_empty

True when S has no elements

S.flattened

Union of all sets in S (⋃ S)

Ints or UnsignedInts

All signed / unsigned integers (infinite sets)

Quantifiers & comprehensions

Generator-expression style

Forall(P(x) for x in S)

∀ x ∈ S : P(x) — equivalent to S.forall(lambda x: P(x))

Exists(P(x) for x in S)

∃ x ∈ S : P(x) — equivalent to S.exists(lambda x: P(x))

Forall(P(x,y) for x in S for y in T)

Multiple binders — ranges over the Cartesian product

Forall(P(x) for x in lst)

Over a list: binds x to each element in order — same as lst.forall(lambda x: P(x))

SetIf(P(x) for x in S)

{ x ∈ S : P(x) } — equivalent to S.filter(lambda x: P(x))

Set(f(x) for x in S)

{ f(x) : x ∈ S } — image set; equivalent to S.map(lambda x: f(x))

Map(f(x) for x in domain)

[ x ↦ f(x) ] — map expression; equivalent to domain.map_to(lambda x: f(x))

Lists

Ordered, 0-based, immutable

List(2, 3, 3, 4)

New list with the given elements in order, duplicates allowed

List(E)

Empty list with elements of sort E

Range(start, end)

New list: [start, start+1, …, end-1] (end-exclusive)

lst[i]

Element at index i (0-based)

lst[s:e] or lst[s:] or lst[:e]

Slice [lst[s], …, lst[e-1]]

lst1 + lst2

Concatenation

lst + List(x)

New list with x added at the end (single-element concatenation; there is no append method)

lst.replace(i, x)

New list with element at index i replaced by x

lst.filter(lambda x: x > 5)

New list: elements satisfying the condition

lst.reduce(lambda acc, x: acc + x, initial)

Ordered left fold; initial is the starting accumulator

lst.exists(lambda x: P(x))

True if some element satisfies P

lst.forall(lambda x: P(x))

True if every element satisfies P

lst.size

Number of elements

lst.is_empty

True when lst has no elements

lst.keys

Set of valid indices: Set(0, 1, …, size-1)

Maps (dictionaries)

Immutable key/value bindings

Map(K, V)

Empty map from type K to type V

Map((k1, v1), (k2, v2), ...)

Explicit finite map; literals are auto-coerced and sorts are inferred from the first pair

MapTo(f(x) for x in domain)

Map binding each element of domain to f(x)

M[key]

Look up value for key (key must exist)

M.insert(k, v)

New map with k → v added or updated

M.replace(k, v)

New map with k → v updated (k must exist)

M.keys

Set of all keys (domain)

M.size

Number of entries

M.is_empty

True when M has no entries

upd = M.edit(); upd[k] = v; M2 = upd.result

Multi-key functional update; .result returns the new map

Records

from wunderspec import Field, record

Record(x=Val(1), y=Val(True))

Anonymous record

@record class Point: x: Field[int] y: Field[int]

Named record type

Point(x=3, y=4)

Construct a record (all fields required; Python literals auto-coerced)

r.x r["x"]

Field access via dot or subscript

r.name r.replace r._.replace(...) upd._.result

Record fields win over expression API names; use ._ for hidden methods or metadata such as r._.sort and r._.node

r.replace(x=Val(10))

New record with the named field(s) changed

upd = r.edit(); upd.x = Val(5); r2 = upd.result

Deep nested update; .result returns the new record

Tuples

Heterogeneous, 0-based, immutable

Tuple("alice", 2025, True)

New tuple (auto-coercion applies)

t[0] t[1] t[2]

Element access (0-based)

(a, b, c) = t

Python unpacking into symbolic variables

t.replace(1, Val(2026))

New tuple with element at the given index replaced

Unions

from wunderspec import Variant, Unit, union

@union class Result: Ok: Variant[int] Err: Variant[str]

Declare a union type; each field is a variant

Result.Ok(Val(42)) Result.Err(Val("oops"))

Construct a variant

@union class Flag: Set_: Variant[Unit] Unset: Variant[Unit]

No-payload variant: use Variant[Unit]

Flag.Set_() Flag.Unset()

Construct a no-payload variant

v.tag

String expression holding the active variant name

v.match(Ok=lambda n: n * Val(2), Err=lambda s: Val(-1))

Exhaustive pattern match — every variant must be handled

v.match(Ok=lambda n: n, default=Val(0))

Non-exhaustive match with a default fallback

All-sets

Higher-order set constructors

AllSubsets(S)

Power set of S: all subsets

AllMaps(keys, values)

All total maps from keys to values

AllTuples(S1, S2, ...)

Cartesian product S1 × S2 × …

AllRecords(x=S1, y=S2)

All records where each field ranges over its set

Decorators

Classify functions and classes

@expr

Symbolic helper function with auto-coercion and LET-binding

@expr(cache_args=False) @expr(pure=True)

Skip argument LET-aliases, or reject state-like arguments for pure helpers

@expr(coerce=False) @expr(inline=False)

Require Expr inputs/outputs, or extract the helper as a named TLA+ operator

@record

Named record type

@union

Tagged union type

@state

State schema (variables + parameters) for a machine

@action @action(init=True) @action(inline=False) @action(coerce=False)

Transition action; init marks init; inline=False is required for fairness/Enabled; coerce=False requires Expr arguments

@invariant

Safety predicate — must hold in every reachable state

@example

Reachability predicate — checker searches for a state where it holds

@temporal

Liveness property returning a TemporalExpr

@instance

Zero-arg factory returning a concrete @state with parameters set

@coverage

State-shape function for coverage tracking during fuzzing / model checking

@with_flow

Imperative if/else expression builder (see flow section)

Temporal expressions

TemporalExpr — used in @temporal properties

Always(P)

□P — P holds in every state of the execution

Eventually(P)

◇P — P holds in some future state

AndT(t1, t2) OrT(t1, t2) NotT(t) ImpliesT(t1, t2)

Temporal connectives; Boolean arguments are lifted to TemporalExpr

Enabled(my_action, arg)

True if my_action can fire; requires @action(inline=False)

Enabled(action_expr)

Direct action-expression form of ENABLED A

WeakFair(my_action, arg, vars=("x",))

WF — if action is continuously enabled it must eventually fire

StrongFair(my_action, arg, vars=("x",))

SF — if action is enabled infinitely often it must eventually fire

Forall(WeakFair(send, p, vars=("msgs",)) for p in PROCS)

Temporal quantification over a parameter set

State machines

from wunderspec import Param, StateVar, Context, MachineStateBase, action, instance, invariant, state

@state class S(MachineStateBase): N: Param[int] ctr: StateVar[int]

Param[T] marks a constant; StateVar[T] marks a mutable variable. The underlying Annotated[...] form is still accepted.

@action(init=True) def init(c: Context[S]): c.state.ctr = Val(0)

Initialisation action: sets all variables to starting values

@action def step(c: Context[S]): alts = iter(c.alternatives("up", "down")) with next(alts): ... # up branch with next(alts): ... # down branch

Step action: c.alternatives() declares mutually exclusive branches

c.state.x = Val(1)

Assign a variable (at most once per transition)

c.assume(condition)

Guard: prune this branch if condition is false

with c.one_of(S, "x") as x: c.state.y = x

Non-deterministic choice from set S

(then_, else_) = c.split(condition)

Branch on a boolean condition

cached = c.cache(expr, name="cached")

Bind a reused symbolic subexpression to a fresh local name

c.state.pc[q] = Val(PC.CS) c.state.req[p][q] = Val(0) c.state.chan.val = d

Direct nested update — map entry, nested path, or record field (immediate)

old_x, old_y = s.x, s.y s.x = old_y s.y = old_x

Read the pre-update ("old") state under direct assignment: snapshot it in a local before assigning (state expressions are immutable). This swaps correctly; s.x = s.y; s.y = s.x would not.

with c.state.editing() as upd: upd.x = c.state.y upd.y = c.state.x

Escape hatch that snapshots automatically across a block (reads see the pre-update state). Equivalent to capturing locals; reach for it only when several interdependent fields make manual snapshots awkward.

balances: StateVar[dict[str, int], UPSERT]

Mark a map field so keyed assignment inserts missing keys (insert-or-update)

Explicit program counters

Recommended replacement for the old process-builder style

@state class S(MachineStateBase): N: Param[int] pc: StateVar[dict[int, PC]] flag: StateVar[dict[int, bool]] num: StateVar[dict[int, int]]

For multiple symmetric processes, store the program counter and process-local variables as map-valued StateVar fields indexed by process id. This is the style used by examples/bakery.py.

def procs(s): return Set(Val(1), ..., s.N) @action(init=True) def Init(c): s, ps = c.state, procs(c.state) s.pc = ps.map_to(lambda _: Val(PC.NCS)) s.flag = ps.map_to(lambda _: Val(False))

Initialize every process id explicitly with map_to. There is no generated process initialization helper.

@action(inline=True) def p(c, q): s = c.state alts = iter(c.alternatives(*(pc.name for pc in PROCESS_PCS)))

Write one ordinary action helper for a single process id. Choose among its possible program-counter cases with c.alternatives(...).

alts = iter(c.alternatives("NCS", "E1")) with next(alts): c.assume(s.pc[q] == PC.NCS) s.pc[q] = Val(PC.E1)

Each branch starts with a guard on the current program counter. Assign the next counter explicitly instead of using goto or fallthrough.

with c.one_of(s.unchecked[q], "i") as i: s.nxt[q] = i c.assume(~s.flag[s.nxt[q]]) s.pc[q] = Val(PC.W2)

Use c.one_of(...) directly in the branch for nondeterministic values. Parameters and variables are read from c.state.

with next(alts): c.assume(s.pc[q] == PC.W2) s.unchecked[q] -= Set(s.nxt[q]) s.pc[q] = Val(PC.W1)

A branch can contain multiple assignments. Set pc[q] to the next control state when the atomic step is done.

@action def Next(c): with c.one_of(procs(c.state), "q") as q: p(c, q)

Choose the process id explicitly in the top-level step. The worker step is just another action helper.

@action(inline=True) def kv_step(c): s = c.state c.assume(s.pc == PC.Wait) s.pc = Val(PC.Sync)

For a single-process machine, use the same pattern with a scalar pc: StateVar[PC].

examples/bakery.py examples/dekker.py

Reference examples for explicit program-counter state machines. Prefer this pattern over generator-based process specifications.

Sub-machines

from wunderspec.submachine import SubMachine

InChan = SubMachine(Data="Message", chan="cin")

Map sub-spec fields to parent fields (TLA⁺ INSTANCE … WITH)

channel.init(InChan(c))

Wrap context c so sub-spec actions read/write parent fields

channel.type_inv(InChan.view(s))

Read-only view of state s with mapped field names (for invariants)

Imperative flow builder

from wunderspec.flow import flow, with_flow

@with_flow def f(x: Expr, y: Expr) -> Expr: with flow.if_(x > y): flow.return_(x) flow.return_(y) return flow.end()

Builds an Ite expression imperatively; sequential with flow.if_ blocks chain as if/elif/else

flow.if_(cond)

Open a conditional branch (use as with flow.if_(cond):)

flow.else_()

Explicit else branch after a completed if_ block (optional)

flow.return_(expr)

Set the value for the current branch

return flow.end()

Assemble the final expression (always the last statement)