API reference

Everything below is re-exported from the top-level musil package, e.g. from musil import Model, check.

Core — check, Model, Action

The model definition and the safety/deadlock/reachability checker.

musil.core

The explicit-state core: a model is (initial states, guarded actions, invariants), and check does a breadth-first sweep of every reachable state, reporting the first invariant violation or deadlock with the shortest trace that reaches it.

States are plain immutable, hashable Python values — a frozen dataclass is the idiomatic choice. Actions are pure: enabled(state) decides whether the action can fire, apply(state) returns the resulting NEW state. Because the sweep fires every enabled action from every state, modelling several concurrent actors is just handing it all their actions — the interleaving is explored for free. That's the whole point: the bugs that matter are races, and BFS over interleavings finds them.

This module is dependency-free and knows nothing about any particular domain.

Action dataclass

A guarded, atomic transition. enabled decides if it can fire in a state; apply returns the resulting new state (states are immutable, so it never mutates). Keep both pure — the checker calls them many times and assumes no side effects.

Source code in src/musil/core.py

@dataclass(frozen=True, slots=True)
class Action[S: Hashable]:
    """A guarded, atomic transition. ``enabled`` decides if it can fire in a state; ``apply`` returns
    the resulting new state (states are immutable, so it never mutates). Keep both pure — the checker
    calls them many times and assumes no side effects."""

    name: str
    enabled: Callable[[S], bool]
    apply: Callable[[S], S]

Step dataclass

One link in a counterexample: the action that fired and the state it produced.

Source code in src/musil/core.py

@dataclass(frozen=True, slots=True)
class Step[S: Hashable]:
    """One link in a counterexample: the action that fired and the state it produced."""

    action: str
    state: S

Model dataclass

What to check: where to start, what can happen, and what must always hold.

init is one state or several. actions are the (possibly multi-actor) transitions. invariants map a name → predicate that must hold in every reachable state. terminal marks states allowed to have no enabled action (an absorbing sink like deleted); any other dead end is reported as a deadlock.

Source code in src/musil/core.py

@dataclass(slots=True)
class Model[S: Hashable]:
    """What to check: where to start, what can happen, and what must always hold.

    ``init`` is one state or several. ``actions`` are the (possibly multi-actor) transitions.
    ``invariants`` map a name → predicate that must hold in every reachable state. ``terminal``
    marks states allowed to have no enabled action (an absorbing sink like ``deleted``); any other
    dead end is reported as a deadlock."""

    init: S | Iterable[S]
    actions: Iterable[Action[S]]
    invariants: Mapping[str, Invariant[S]] = field(default_factory=dict[str, Invariant[S]])
    terminal: Callable[[S], bool] = _never

    def initial_states(self) -> tuple[S, ...]:
        init = self.init
        if _is_state_collection(init):
            return tuple(init)  # type: ignore[arg-type]
        return (init,)  # type: ignore[return-value]

Result dataclass

The outcome. ok is True when the whole reachable space satisfied every invariant with no deadlock. Otherwise kind is "invariant" or "deadlock", invariant names the broken one, and trace is the shortest path from an initial state to the offending state.

Source code in src/musil/core.py

@dataclass(frozen=True, slots=True)
class Result[S: Hashable]:
    """The outcome. ``ok`` is True when the whole reachable space satisfied every invariant with no
    deadlock. Otherwise ``kind`` is ``"invariant"`` or ``"deadlock"``, ``invariant`` names the broken
    one, and ``trace`` is the shortest path from an initial state to the offending state."""

    ok: bool
    states_explored: int
    kind: str | None = None
    invariant: str | None = None
    reason: str | None = None  # detail from a string-returning invariant (which sub-rule/entity)
    trace: tuple[Step[S], ...] = ()
    truncated: bool = False  # True if the max_states cap was hit before the space was exhausted

    def __bool__(self) -> bool:
        return self.ok

    def __str__(self) -> str:
        if self.ok:
            cap = " (TRUNCATED at cap)" if self.truncated else ""
            return f"OK — {self.states_explored} states, no violations{cap}"
        if self.kind == "invariant":
            head = f"INVARIANT VIOLATED: {self.invariant}"
            if self.reason:
                head += f" — {self.reason}"
        else:
            head = "DEADLOCK (a non-terminal state with no enabled action)"
        return f"{head}\n{format_trace(self.trace)}\n({self.states_explored} states explored)"

Verdict

Bases: Protocol

What assert_ok needs from a result: a boolean verdict and a renderer. Every musil result type satisfies it structurally -- Result, LivenessResult, ConformanceResult, RefinementResult, OpenResult, SimReport.

Source code in src/musil/core.py

class Verdict(Protocol):
    """What ``assert_ok`` needs from a result: a boolean verdict and a renderer. Every musil result
    type satisfies it structurally -- ``Result``, ``LivenessResult``, ``ConformanceResult``,
    ``RefinementResult``, ``OpenResult``, ``SimReport``."""

    @property
    def ok(self) -> bool: ...

    def __str__(self) -> str: ...

Graph dataclass

The full reachable state graph: initial states, every reachable state, and the labelled transitions (edges[s] = the (action_name, next_state) pairs out of s). Built by explore and consumed by liveness checking and graph export. truncated flags hitting the max_states cap.

Source code in src/musil/core.py

@dataclass(frozen=True, slots=True)
class Graph[S: Hashable]:
    """The full reachable state graph: initial states, every reachable state, and the labelled
    transitions (``edges[s]`` = the ``(action_name, next_state)`` pairs out of ``s``). Built by
    ``explore`` and consumed by liveness checking and graph export. ``truncated`` flags hitting the
    ``max_states`` cap."""

    initial: tuple[S, ...]
    states: tuple[S, ...]
    edges: Mapping[S, tuple[tuple[str, S], ...]]
    truncated: bool = False

FixedPoints dataclass

Where the system settles: for each initial state, the distinct terminal states (no outgoing edge) reachable from it. Built by fixed_points; confluent is the headline question.

Source code in src/musil/core.py

@dataclass(frozen=True, slots=True)
class FixedPoints[S: Hashable]:
    """Where the system settles: for each initial state, the distinct terminal states (no outgoing
    edge) reachable from it. Built by ``fixed_points``; ``confluent`` is the headline question."""

    by_init: Mapping[S, frozenset[S]]
    truncated: bool = False

    @property
    def confluent(self) -> bool:
        """True when every single initial state settles in at most one place — the fixed point does
        not depend on the order actions fired. A start with an *empty* set never settles at all (it
        can only loop); that is a liveness question, not a confluence one, and does not count
        against confluence. False whenever ``truncated`` (inconclusive)."""
        return not self.truncated and all(len(t) <= 1 for t in self.by_init.values())

    def __str__(self) -> str:
        if self.truncated:
            return "FIXED POINTS INCONCLUSIVE (truncated at max_states)"
        lines = [
            f"  {init} settles in {len(terms)} place(s)" for init, terms in self.by_init.items()
        ]
        head = "CONFLUENT" if self.confluent else "NOT CONFLUENT (an initial state has a choice)"
        return "\n".join([head, *lines])

confluent property

confluent: bool

True when every single initial state settles in at most one place — the fixed point does not depend on the order actions fired. A start with an empty set never settles at all (it can only loop); that is a liveness question, not a confluence one, and does not count against confluence. False whenever truncated (inconclusive).

StateViolation dataclass

One reachable state that breaks one invariant. reason carries the detail from a string-returning invariant, if any.

Source code in src/musil/core.py

@dataclass(frozen=True, slots=True)
class StateViolation[S: Hashable]:
    """One reachable state that breaks one invariant. ``reason`` carries the detail from a
    string-returning invariant, if any."""

    state: S
    invariant: str
    reason: str | None = None

invariant_from_violations

invariant_from_violations(
    checker: Callable[[S], Sequence[object]],
    *,
    describe: Callable[[object], str] = str,
) -> Invariant[S]

Adapt an external checker into a musil invariant. checker(state) returns a sequence of violation objects — empty means the state is fine. The resulting invariant returns True when that sequence is empty, else the first violation's describe(...) string (a reason surfaced in the counterexample). Use this to drive a real world -> [Violation] checker as an invariant without the not checker(s) footgun (a truthy reason string would otherwise read as 'holds').

Source code in src/musil/core.py

def invariant_from_violations[S: Hashable](
    checker: Callable[[S], Sequence[object]],
    *,
    describe: Callable[[object], str] = str,
) -> Invariant[S]:
    """Adapt an external checker into a musil invariant. ``checker(state)`` returns a sequence of
    violation objects — empty means the state is fine. The resulting invariant returns ``True`` when
    that sequence is empty, else the first violation's ``describe(...)`` string (a reason surfaced in
    the counterexample). Use this to drive a real ``world -> [Violation]`` checker as an invariant
    without the ``not checker(s)`` footgun (a truthy reason string would otherwise read as 'holds')."""

    def inv(state: S) -> bool | str:
        result = checker(state)
        return True if not result else describe(result[0])

    return inv

format_trace

format_trace(trace: Iterable[Step[S]]) -> str

Render a counterexample as Init: <state> then → <action> <state> per step.

Source code in src/musil/core.py

def format_trace[S: Hashable](trace: Iterable[Step[S]]) -> str:
    """Render a counterexample as ``Init: <state>`` then ``→ <action>  <state>`` per step."""
    lines: list[str] = []
    for i, step in enumerate(trace):
        if i == 0:
            lines.append(f"  Init: {step.state}")
        else:
            lines.append(f"  → {step.action}: {step.state}")
    return "\n".join(lines)

assert_ok

assert_ok(
    result: Verdict, message: str | None = None
) -> None

Assert a result is OK, failing with its rendered form -- the counterexample, not the repr.

assert check(model) inside a test runner reports the result's repr, and pytest shortens a long one by eliding its middle: the trace comes out as trace=(Step(action='<init>', s...Step(action='inc', state=S(n=3))), with the initial state cut, followed by the whole Model repr (lambda addresses included) as assertion context. The counterexample is not merely ugly there, it is incomplete. This raises AssertionError(str(result)) instead, so the failure message is the Init: ... -> action: ... rendering __str__ already produces. message is prefixed when the same assertion runs over several models.

Works on any musil result (they all carry ok and a meaningful __str__), and needs no test framework -- it is a plain AssertionError, so unittest and bare assert get it too.

Source code in src/musil/core.py

def assert_ok(result: Verdict, message: str | None = None) -> None:
    """Assert a result is OK, failing with its *rendered* form -- the counterexample, not the repr.

    ``assert check(model)`` inside a test runner reports the result's **repr**, and pytest shortens a
    long one by eliding its middle: the trace comes out as
    ``trace=(Step(action='<init>', s...Step(action='inc', state=S(n=3)))``, with the initial state
    cut, followed by the whole ``Model`` repr (lambda addresses included) as assertion context. The
    counterexample is not merely ugly there, it is incomplete. This raises ``AssertionError(str(result))``
    instead, so the failure message is the ``Init: ... -> action: ...`` rendering ``__str__`` already
    produces. ``message`` is prefixed when the same assertion runs over several models.

    Works on any musil result (they all carry ``ok`` and a meaningful ``__str__``), and needs no test
    framework -- it is a plain ``AssertionError``, so ``unittest`` and bare ``assert`` get it too."""
    # pytest hides a frame that sets this, which is the difference between a clean failure and one
    # that reprints the elided repr as the helper's arguments. It is an ordinary local otherwise, so
    # nothing here depends on pytest being installed.
    __tracebackhide__ = True
    if result.ok:
        return
    rendered = str(result)
    raise AssertionError(f"{message}\n{rendered}" if message else rendered)

check

check(
    model: Model[S],
    *,
    max_states: int = 1000000,
    canonicalize: Callable[[S], S] | None = None,
) -> Result[S]

Breadth-first sweep of every reachable state. Returns the first invariant violation or deadlock found (shortest trace, because BFS visits by increasing depth), else OK once the whole reachable space is exhausted. max_states is a runaway guard for unbounded models.

canonicalize enables symmetry reduction (OPT-IN, soundness is the caller's obligation): map each state to a canonical representative of its symmetry-equivalence class and only that representative is explored — collapsing interchangeable components (e.g. sort the per-worker slices) so the space stops blowing up combinatorially. SOUND only if it is a genuine symmetry of the model: every action commutes with it and every invariant is constant across each class. Getting that wrong silently drops reachable states (misses bugs), so VALIDATE it on small instances with symmetry_reduction_sound before trusting it on large ones. Counterexample traces are then witnesses up to symmetry (each step a real action followed by a renaming).

Source code in src/musil/core.py

def check[S: Hashable](
    model: Model[S],
    *,
    max_states: int = 1_000_000,
    canonicalize: Callable[[S], S] | None = None,
) -> Result[S]:
    """Breadth-first sweep of every reachable state. Returns the first invariant violation or
    deadlock found (shortest trace, because BFS visits by increasing depth), else OK once the whole
    reachable space is exhausted. ``max_states`` is a runaway guard for unbounded models.

    ``canonicalize`` enables **symmetry reduction** (OPT-IN, soundness is the caller's obligation):
    map each state to a canonical representative of its symmetry-equivalence class and only that
    representative is explored — collapsing interchangeable components (e.g. sort the per-worker
    slices) so the space stops blowing up combinatorially. SOUND only if it is a genuine symmetry of
    the model: every action commutes with it and every invariant is constant across each class.
    Getting that wrong silently drops reachable states (misses bugs), so VALIDATE it on small
    instances with ``symmetry_reduction_sound`` before trusting it on large ones. Counterexample
    traces are then witnesses *up to symmetry* (each step a real action followed by a renaming)."""
    starts = model.initial_states()
    actions = tuple(model.actions)
    canon: Callable[[S], S] = canonicalize if canonicalize is not None else (lambda s: s)

    # parent[state] = (previous_state, action_name) — for reconstructing the shortest trace.
    parent: dict[S, tuple[S, str] | None] = {}
    seen: set[S] = set()
    frontier: list[S] = []
    for s0 in starts:
        s = canon(s0)
        if s not in seen:
            seen.add(s)
            parent[s] = None
            frontier.append(s)

    head = 0  # index into frontier instead of pop(0), so it behaves as a real FIFO queue
    while head < len(frontier):
        state = frontier[head]
        head += 1

        for name, inv in model.invariants.items():
            verdict = inv(state)
            # A string return is a violation carrying a reason; otherwise legacy truthiness:
            # falsy (False/None/"") is a violation, truthy holds.
            if isinstance(verdict, str):
                return Result(
                    ok=False, states_explored=len(seen), kind="invariant", invariant=name,
                    reason=verdict or None, trace=_trace_to(parent, state),
                )
            if not verdict:
                return Result(
                    ok=False, states_explored=len(seen), kind="invariant", invariant=name,
                    trace=_trace_to(parent, state),
                )

        fired = False
        for action in actions:
            if not action.enabled(state):
                continue
            fired = True
            nxt = canon(action.apply(state))
            if nxt not in seen:
                seen.add(nxt)
                parent[nxt] = (state, action.name)
                frontier.append(nxt)
                if len(seen) >= max_states:
                    return Result(ok=True, states_explored=len(seen), truncated=True)

        if not fired and not model.terminal(state):
            return Result(
                ok=False,
                states_explored=len(seen),
                kind="deadlock",
                trace=_trace_to(parent, state),
            )

    return Result(ok=True, states_explored=len(seen))

explore

explore(
    model: Model[S],
    *,
    max_states: int = 1000000,
    canonicalize: Callable[[S], S] | None = None,
) -> Graph[S]

Build the whole reachable state graph (no invariant checks). The shared exploration primitive behind liveness and graph export; check stays separate so it can early-exit with the shortest counterexample. canonicalize enables symmetry reduction (see check); validate it with symmetry_reduction_sound before trusting it.

Source code in src/musil/core.py

def explore[S: Hashable](
    model: Model[S],
    *,
    max_states: int = 1_000_000,
    canonicalize: Callable[[S], S] | None = None,
) -> Graph[S]:
    """Build the whole reachable state graph (no invariant checks). The shared exploration primitive
    behind liveness and graph export; ``check`` stays separate so it can early-exit with the shortest
    counterexample. ``canonicalize`` enables symmetry reduction (see ``check``); validate it with
    ``symmetry_reduction_sound`` before trusting it."""
    starts = model.initial_states()
    actions = tuple(model.actions)
    canon: Callable[[S], S] = canonicalize if canonicalize is not None else (lambda s: s)
    seen: dict[S, list[tuple[str, S]]] = {}
    frontier: list[S] = []
    for s0 in starts:
        s = canon(s0)
        if s not in seen:
            seen[s] = []
            frontier.append(s)

    head = 0
    truncated = False
    while head < len(frontier):
        state = frontier[head]
        head += 1
        out: list[tuple[str, S]] = []
        for action in actions:
            if not action.enabled(state):
                continue
            nxt = canon(action.apply(state))
            out.append((action.name, nxt))
            if nxt not in seen:
                seen[nxt] = []
                frontier.append(nxt)
                if len(seen) >= max_states:
                    truncated = True
                    break
        seen[state] = out
        if truncated:
            break

    return Graph(
        initial=tuple(canon(s) for s in starts),
        states=tuple(seen),
        edges={s: tuple(out) for s, out in seen.items()},
        truncated=truncated,
    )

reaches

reaches(
    graph: Graph[S], predicate: Callable[[S], bool]
) -> frozenset[S]

The states of graph from which a predicate-satisfying state is reachable (states already satisfying it included) — plain reachability, by BFS over the reversed edges.

This is the weaker cousin of liveness: "can it still get there?", not "must it get there?". reaches(g, p) == set(g.states) says the goal stays available from every reachable state (in CTL terms: AG EF p) — e.g. "fmi3FreeInstance is legal in every mode" — whereas check_liveness(..., everywhere=True) demands every run actually reach it (AG AF p): a cycle that forever declines an available exit toward p violates the latter, never the former. The complement set(graph.states) - reaches(graph, p) is the trap — reachable states from which p is already lost.

Source code in src/musil/core.py

def reaches[S: Hashable](graph: Graph[S], predicate: Callable[[S], bool]) -> frozenset[S]:
    """The states of ``graph`` from which a ``predicate``-satisfying state is reachable (states
    already satisfying it included) — plain reachability, by BFS over the reversed edges.

    This is the *weaker* cousin of liveness: "can it still get there?", not "must it get there?".
    ``reaches(g, p) == set(g.states)`` says the goal stays available from every reachable state
    (in CTL terms: AG EF p) — e.g. "``fmi3FreeInstance`` is legal in every mode" — whereas
    ``check_liveness(..., everywhere=True)`` demands every run *actually* reach it (AG AF p): a
    cycle that forever declines an available exit toward p violates the latter, never the former.
    The complement ``set(graph.states) - reaches(graph, p)`` is the trap — reachable states from
    which p is already lost."""
    rev: dict[S, list[S]] = {s: [] for s in graph.states}
    for s in graph.states:
        for _a, t in graph.edges[s]:
            rev[t].append(s)
    seen = {s for s in graph.states if predicate(s)}
    frontier = list(seen)
    head = 0
    while head < len(frontier):
        s = frontier[head]
        head += 1
        for p in rev[s]:
            if p not in seen:
                seen.add(p)
                frontier.append(p)
    return frozenset(seen)

fixed_points

fixed_points(
    model: Model[S],
    *,
    max_states: int = 1000000,
    canonicalize: Callable[[S], S] | None = None,
) -> FixedPoints[S]

From each initial state separately: in which distinct terminal states can the runs end? One start that can settle in two different places means the outcome depends on the order actions fired — the bug this exists to catch (e.g. event iteration whose fixed point depends on which pending event is applied first). Per initial state on purpose: different starts are allowed to settle differently (that is what a latch is for); confluent only demands each single start settle in at most one place.

Terminal here is structural — no outgoing edge — independent of model.terminal: the question is where the dynamics stop, not which stops are declared legitimate. A start mapped to an empty set never settles (every path loops); ask check_liveness about that instead. canonicalize applies symmetry reduction (see check). If any exploration hits max_states the result is inconclusive: truncated is set and confluent is False.

Source code in src/musil/core.py

def fixed_points[S: Hashable](
    model: Model[S],
    *,
    max_states: int = 1_000_000,
    canonicalize: Callable[[S], S] | None = None,
) -> FixedPoints[S]:
    """From each initial state separately: in which distinct terminal states can the runs end?
    One start that can settle in two different places means the outcome depends on the order
    actions fired — the bug this exists to catch (e.g. event iteration whose fixed point depends
    on which pending event is applied first). Per initial state on purpose: different starts are
    *allowed* to settle differently (that is what a latch is for); ``confluent`` only demands each
    single start settle in at most one place.

    Terminal here is structural — no outgoing edge — independent of ``model.terminal``: the
    question is where the dynamics stop, not which stops are declared legitimate. A start mapped
    to an empty set never settles (every path loops); ask ``check_liveness`` about that instead.
    ``canonicalize`` applies symmetry reduction (see ``check``). If any exploration hits
    ``max_states`` the result is inconclusive: ``truncated`` is set and ``confluent`` is False."""
    by_init: dict[S, frozenset[S]] = {}
    truncated = False
    for s0 in model.initial_states():
        g = explore(
            Model(init=(s0,), actions=model.actions),
            max_states=max_states,
            canonicalize=canonicalize,
        )
        truncated = truncated or g.truncated
        by_init[g.initial[0]] = frozenset(s for s in g.states if not g.edges[s])
    return FixedPoints(by_init=by_init, truncated=truncated)

symmetry_reduction_sound

symmetry_reduction_sound(
    model: Model[S],
    canonicalize: Callable[[S], S],
    *,
    max_states: int = 1000000,
) -> bool

Validate a symmetry-reduction canonicalize for model on a (bounded) instance — the soundness gate that should pass before check(..., canonicalize=...) is trusted on larger instances.

It explores the model fully and reduced and confirms two things agree: (1) the reduced run visited exactly the canonical quotient of the full reachable set ({canonicalize(s) for s in full} == set(reduced)) — reachability coverage; and (2) the SAME set of invariants is violable in both runs — which additionally certifies that every invariant is constant across each symmetry class (a state and its representative agree).

Both matter: (1) alone passes for a lossy projection that merges non-equivalent states, but (2) then catches it because an invariant violated on a dropped state is no longer violated on its (information-losing) representative. Because the full run is the trusted reference, any unsound canonicalize — not a real symmetry, or a real symmetry against a non-class-constant invariant — makes one of these disagree and returns False. (Deadlock-freedom under reduction is NOT covered; validate that separately.) Returns False if either run truncated at max_states (inconclusive).

Source code in src/musil/core.py

def symmetry_reduction_sound[S: Hashable](
    model: Model[S],
    canonicalize: Callable[[S], S],
    *,
    max_states: int = 1_000_000,
) -> bool:
    """Validate a symmetry-reduction ``canonicalize`` for ``model`` on a (bounded) instance — the
    soundness gate that should pass before ``check(..., canonicalize=...)`` is trusted on larger
    instances.

    It explores the model fully and reduced and confirms two things agree:
    (1) the reduced run visited exactly the canonical quotient of the full reachable set
        (``{canonicalize(s) for s in full} == set(reduced)``) — reachability coverage; and
    (2) the SAME set of invariants is violable in both runs — which additionally certifies that every
        invariant is constant across each symmetry class (a state and its representative agree).

    Both matter: (1) alone passes for a lossy projection that merges non-equivalent states, but (2)
    then catches it because an invariant violated on a dropped state is no longer violated on its
    (information-losing) representative. Because the full run is the trusted reference, any unsound
    ``canonicalize`` — not a real symmetry, or a real symmetry against a non-class-constant invariant —
    makes one of these disagree and returns False. (Deadlock-freedom under reduction is NOT covered;
    validate that separately.) Returns False if either run truncated at ``max_states`` (inconclusive)."""
    full = explore(model, max_states=max_states)
    reduced = explore(model, max_states=max_states, canonicalize=canonicalize)
    if full.truncated or reduced.truncated:
        return False
    coverage = {canonicalize(s) for s in full.states} == set(reduced.states)
    full_violated = {v.invariant for v in reachable_violations(model, max_states=max_states)}
    reduced_violated = {
        v.invariant
        for v in reachable_violations(model, max_states=max_states, canonicalize=canonicalize)
    }
    return coverage and full_violated == reduced_violated

reachable_violations

reachable_violations(
    model: Model[S],
    *,
    max_states: int = 1000000,
    canonicalize: Callable[[S], S] | None = None,
) -> list[StateViolation[S]]

Every (reachable state, broken invariant) pair — an exhaustive audit, where check stops at the first. Use it to enumerate all the invalid states a model can reach — e.g. to derive the exact set of cross-resource combinations that guards must exclude — instead of one shortest counterexample. Deadlock is not considered (use check for that). Each invariant is evaluated with the same semantics as check: a string return is a violation carrying that reason. canonicalize applies symmetry reduction (see check).

Source code in src/musil/core.py

def reachable_violations[S: Hashable](
    model: Model[S], *, max_states: int = 1_000_000, canonicalize: Callable[[S], S] | None = None
) -> list[StateViolation[S]]:
    """Every ``(reachable state, broken invariant)`` pair — an exhaustive audit, where ``check`` stops
    at the first. Use it to enumerate *all* the invalid states a model can reach — e.g. to derive the
    exact set of cross-resource combinations that guards must exclude — instead of one shortest
    counterexample. Deadlock is not considered (use ``check`` for that). Each invariant is evaluated
    with the same semantics as ``check``: a string return is a violation carrying that reason.
    ``canonicalize`` applies symmetry reduction (see ``check``)."""
    g = explore(model, max_states=max_states, canonicalize=canonicalize)
    out: list[StateViolation[S]] = []
    for s in g.states:
        for name, inv in model.invariants.items():
            verdict = inv(s)
            if isinstance(verdict, str):
                out.append(StateViolation(state=s, invariant=name, reason=verdict or None))
            elif not verdict:
                out.append(StateViolation(state=s, invariant=name))
    return out

Symmetry — canonical_by_sorting, canonical_by_permutations

Build the canonicalize that check/explore take, instead of hand-writing it. Sort the slices when the components are anonymous; relabel the whole state under each permutation when something names one. Either way, symmetry_reduction_sound still has to pass on a small instance first.

musil.symmetry

Building the canonicalize that check/explore take -- symmetry reduction you can derive rather than hand-write.

check(model, canonicalize=...) has always accepted a symmetry reduction, and symmetry_reduction_sound has always been able to grade one. What was missing is the step between: turning "these N components are interchangeable" into the function. Hand-writing it is where the mistakes live, and an unsound canonicalizer silently drops reachable states -- a checker returning green for a question nobody asked.

Two constructions, and which one is correct depends on one question: does anything else in the state name a component?

• canonical_by_sorting -- the components are anonymous. Their slices are sorted into a canonical order. O(N log N), so it is the one that scales, and the one to reach for by default. Sound only while no other field refers to a component by position or id: sorting the slices destroys that correspondence, and the state that said "worker 2 holds the lock" now says something else.
• canonical_by_permutations -- something does name a component (a leader, an owner, a route id in a message). The whole state is relabelled under each permutation of the identities and the smallest result wins, so every reference moves with the component it names. Correct where sorting is not, and factorial: usable for a handful of identities, not for 24.

Both still have to pass symmetry_reduction_sound on a small instance before being trusted on a large one -- that gate is the point, and neither construction removes the obligation. The sorting case is exactly where it earns its keep: a model that grows a leader field next year makes a previously-sound sort unsound, and nothing about the sort itself changes to say so.

canonical_by_sorting

canonical_by_sorting(
    *,
    get: Callable[[S], Sequence[T]],
    put: Callable[[S, tuple[T, ...]], S],
    key: Callable[[T], Any] | None = None,
) -> Callable[[S], S]

Canonicalize by sorting the interchangeable components' slices into a fixed order.

get(state) reads the per-component slices (one entry per component, in component order) and put(state, slices) writes a reordered tuple back -- the same accessor pair the channel kit uses. key orders slices that are not comparable with < (a frozen dataclass per component, say); without it the slices are sorted directly.

Collapses the N! orderings of N anonymous components into one representative, so a space that was V ** N becomes the multisets of size N over V. For 5 components over 3 values: 243 states down to 21.

SOUND ONLY IF nothing outside these slices names a component. A leader: int field indexing into them, an id inside a message, a per-component key elsewhere in the state -- any of those and sorting breaks the reference while leaving the state well-formed, which is the quiet kind of wrong. Use canonical_by_permutations there, and either way run symmetry_reduction_sound on a small instance first.

Source code in src/musil/symmetry.py

def canonical_by_sorting[S: Hashable, T](
    *,
    get: Callable[[S], Sequence[T]],
    put: Callable[[S, tuple[T, ...]], S],
    key: Callable[[T], Any] | None = None,
) -> Callable[[S], S]:
    """Canonicalize by sorting the interchangeable components' slices into a fixed order.

    ``get(state)`` reads the per-component slices (one entry per component, in component order) and
    ``put(state, slices)`` writes a reordered tuple back -- the same accessor pair the channel kit
    uses. ``key`` orders slices that are not comparable with ``<`` (a frozen dataclass per component,
    say); without it the slices are sorted directly.

    Collapses the N! orderings of N anonymous components into one representative, so a space that was
    ``V ** N`` becomes the multisets of size N over V. For 5 components over 3 values: 243 states
    down to 21.

    SOUND ONLY IF nothing outside these slices names a component. A ``leader: int`` field indexing
    into them, an id inside a message, a per-component key elsewhere in the state -- any of those and
    sorting breaks the reference while leaving the state well-formed, which is the quiet kind of
    wrong. Use ``canonical_by_permutations`` there, and either way run ``symmetry_reduction_sound``
    on a small instance first."""

    def canonicalize(state: S) -> S:
        return put(state, tuple(sorted(get(state), key=key)))  # type: ignore[arg-type, type-var]

    return canonicalize

canonical_by_permutations

canonical_by_permutations(
    identities: Sequence[I],
    relabel: Callable[[S, Mapping[I, I]], S],
    *,
    order: Order[S] = repr,
    max_permutations: int = 5040,
) -> Callable[[S], S]

Canonicalize by taking the smallest state in its orbit under every permutation of identities -- the general construction, for when the identities are referenced.

relabel(state, mapping) applies one permutation to the WHOLE state: every place an identity appears must be rewritten, not just the per-component slices. Missing one is precisely the bug this construction exists to avoid, and it shows up as symmetry_reduction_sound returning False rather than as a wrong answer.

order is the total order the minimum is taken under; any fixed one yields a canonical representative, and the default repr works for arbitrary states without requiring <. It must be stable for the duration of a check -- true for the frozen dataclasses states are normally written as. (Ordering by repr of a frozenset field is stable within a process but varies between runs under hash randomization: still canonical, just not reproducible across runs.)

The orbit has len(identities)! elements and all of them are built for every state, so this is for a handful of identities. max_permutations (5040 = 7!) makes that a loud error instead of a hang; raising it is a decision, not a default. For many anonymous components, canonical_by_sorting is the construction that scales.

Source code in src/musil/symmetry.py

def canonical_by_permutations[S: Hashable, I](
    identities: Sequence[I],
    relabel: Callable[[S, Mapping[I, I]], S],
    *,
    order: Order[S] = repr,
    max_permutations: int = 5_040,
) -> Callable[[S], S]:
    """Canonicalize by taking the smallest state in its orbit under every permutation of
    ``identities`` -- the general construction, for when the identities are referenced.

    ``relabel(state, mapping)`` applies one permutation to the WHOLE state: every place an identity
    appears must be rewritten, not just the per-component slices. Missing one is precisely the bug
    this construction exists to avoid, and it shows up as ``symmetry_reduction_sound`` returning
    False rather than as a wrong answer.

    ``order`` is the total order the minimum is taken under; any fixed one yields a canonical
    representative, and the default ``repr`` works for arbitrary states without requiring ``<``. It
    must be stable for the duration of a check -- true for the frozen dataclasses states are normally
    written as. (Ordering by ``repr`` of a ``frozenset`` field is stable within a process but varies
    between runs under hash randomization: still canonical, just not reproducible across runs.)

    The orbit has ``len(identities)!`` elements and all of them are built for every state, so this is
    for a handful of identities. ``max_permutations`` (5040 = 7!) makes that a loud error instead of a
    hang; raising it is a decision, not a default. For many anonymous components, ``canonical_by_sorting``
    is the construction that scales."""
    n = len(identities)
    if factorial(n) > max_permutations:
        raise ValueError(
            f"{n} identities means {factorial(n)} permutations per state, over the "
            f"max_permutations={max_permutations} cap. Sorting the per-component slices with "
            "canonical_by_sorting is the construction that scales; raise the cap only if the "
            "identities really are referenced elsewhere and N is small."
        )
    perms = tuple(
        dict(zip(identities, p, strict=True)) for p in permutations(identities)
    )

    def canonicalize(state: S) -> S:
        return min((relabel(state, p) for p in perms), key=order)

    return canonicalize

Liveness — check_liveness

Eventually (<>P) and always-eventually ([]<>P) under weak and strong fairness.

musil.liveness

Liveness: does the system eventually reach a goal, or can it loop forever short of it?

check_liveness(model, goal=P) proves the temporal property "eventually P" (<>P): on every run from an initial state, P holds at some point. It fails in exactly two ways, and we report whichever we find with a lasso counterexample (a stem into a recurring set, plus the loop):

• P unreachable — a reachable state from which P can never be reached (a dead end or a trap SCC with no exit to P). No scheduling fixes this.
• fair cycle — a reachable cycle of ¬P states that the system can follow forever. Whether this counts depends on FAIRNESS: under weak fairness of an action set, an action that is continuously enabled along the cycle must eventually be taken — so a cycle that starves a continuously-enabled fair action is not a real counterexample (the fair scheduler would break it). Pass fair=[...] to rule those out; with fair=() you get strict (no-fairness) liveness.

Method: explore the reachable graph, restrict to ¬P states reachable through ¬P from a ¬P initial state, then (1) flag any from which P is unreachable, (2) Tarjan-SCC the rest and report the first weakly-fair non-trivial SCC. Dependency-free; for finite (bounded) models.

LivenessResult dataclass

Outcome of an "eventually P" check. ok True means every run reaches P. Otherwise kind is "p-unreachable" or "fair-cycle"; stem is the shortest path from an initial state to the offending state, and cycle is the recurring loop (empty for a P-unreachable sink).

Source code in src/musil/liveness.py

@dataclass(frozen=True, slots=True)
class LivenessResult[S: Hashable]:
    """Outcome of an "eventually P" check. ``ok`` True means every run reaches P. Otherwise ``kind``
    is ``"p-unreachable"`` or ``"fair-cycle"``; ``stem`` is the shortest path from an initial state
    to the offending state, and ``cycle`` is the recurring loop (empty for a P-unreachable sink)."""

    ok: bool
    states_explored: int
    goal: str = "goal"
    kind: str | None = None
    stem: tuple[Step[S], ...] = ()
    cycle: tuple[Step[S], ...] = ()
    truncated: bool = False

    def __bool__(self) -> bool:
        return self.ok

    def __str__(self) -> str:
        if self.ok:
            cap = " (TRUNCATED at cap)" if self.truncated else ""
            return f"OK — eventually '{self.goal}' on every run ({self.states_explored} states){cap}"
        if self.kind == "p-unreachable":
            return (
                f"LIVENESS VIOLATED: '{self.goal}' is unreachable from a reachable state\n"
                f"{format_trace(self.stem)}\n({self.states_explored} states explored)"
            )
        loop = "\n".join(f"  ↺ {s.action}: {s.state}" for s in self.cycle)
        return (
            f"LIVENESS VIOLATED: a fair cycle never reaches '{self.goal}'\n"
            f"{format_trace(self.stem)}\n  --- loops forever ---\n{loop}\n"
            f"({self.states_explored} states explored)"
        )

check_liveness

check_liveness(
    model: Model[S],
    *,
    goal: Callable[[S], bool],
    fair: Iterable[str] = (),
    fair_strong: Iterable[str] = (),
    everywhere: bool = False,
    leadsto_from: Callable[[S], bool] | None = None,
    goal_name: str = "goal",
    max_states: int = 1000000,
) -> LivenessResult[S]

Prove a liveness property and return a lasso counterexample if it fails.

Default (everywhere=False) checks <>P — eventually P on every run from an initial state. everywhere=True checks []<>P — on every run, always eventually P: past any point of any run, P must actually be reached again. This is NOT reachability: a reachable cycle that forever declines to move toward P is a violation even when P stays reachable from every state on it — declare the exit in fair=[...] if a fair scheduler would eventually take it. For the weaker "P merely remains reachable from every state" (AG EF, no state is ever trapped), use reaches: reaches(g, goal) == set(g.states) with g = explore(model). Use everywhere=True for convergence/recurrence properties whose model may start already satisfying P.

leadsto_from=Q checks the response property [](Q -> <>P) — whenever Q holds, P eventually follows. This is the untimed "no resource gets stuck" property: e.g. every service that enters placed eventually reaches running or failed. It is weaker than everywhere=True — only states triggered by Q must reach P, so a ¬P cycle that no Q-state can reach is not a violation. (Implemented by seeding the same analysis from every reachable Q∧¬P state; everywhere is ignored when leadsto_from is given.)

Fairness prunes cycles a fair scheduler would break. fair = weak fairness (an action continuously enabled along a cycle must eventually be taken). fair_strong = strong fairness (an action enabled infinitely often — i.e. in any state of the cycle — must be taken); strong is the stronger assumption and rules out more cycles, so it can force a "flickering" action that weak fairness cannot.

Source code in src/musil/liveness.py

def check_liveness[S: Hashable](
    model: Model[S],
    *,
    goal: Callable[[S], bool],
    fair: Iterable[str] = (),
    fair_strong: Iterable[str] = (),
    everywhere: bool = False,
    leadsto_from: Callable[[S], bool] | None = None,
    goal_name: str = "goal",
    max_states: int = 1_000_000,
) -> LivenessResult[S]:
    """Prove a liveness property and return a lasso counterexample if it fails.

    Default (``everywhere=False``) checks ``<>P`` — *eventually* P on every run from an initial
    state. ``everywhere=True`` checks ``[]<>P`` — on every run, *always eventually* P: past any
    point of any run, P must actually be reached again. This is NOT reachability: a reachable
    cycle that forever declines to move toward P is a violation even when P stays reachable from
    every state on it — declare the exit in ``fair=[...]`` if a fair scheduler would eventually
    take it. For the weaker "P merely remains reachable from every state" (AG EF, no state is ever
    *trapped*), use ``reaches``: ``reaches(g, goal) == set(g.states)`` with ``g = explore(model)``.
    Use ``everywhere=True`` for convergence/recurrence properties whose model may start already
    satisfying P.

    ``leadsto_from=Q`` checks the **response** property ``[](Q -> <>P)`` — *whenever Q holds, P
    eventually follows*. This is the untimed "no resource gets stuck" property: e.g. every service
    that enters ``placed`` eventually reaches ``running`` or ``failed``. It is weaker than
    ``everywhere=True`` — only states triggered by Q must reach P, so a ¬P cycle that no Q-state can
    reach is not a violation. (Implemented by seeding the same analysis from every reachable Q∧¬P
    state; ``everywhere`` is ignored when ``leadsto_from`` is given.)

    Fairness prunes cycles a fair scheduler would break. ``fair`` = weak fairness (an action
    *continuously* enabled along a cycle must eventually be taken). ``fair_strong`` = strong fairness
    (an action enabled *infinitely often* — i.e. in any state of the cycle — must be taken); strong
    is the stronger assumption and rules out more cycles, so it can force a "flickering" action that
    weak fairness cannot."""
    g = explore(model, max_states=max_states)
    fairset = frozenset(fair)
    strongset = frozenset(fair_strong)

    bad = {s for s in g.states if not goal(s)}
    if not bad:
        return LivenessResult(ok=True, states_explored=len(g.states), goal=goal_name, truncated=g.truncated)

    if leadsto_from is not None:
        # [](Q -> <>P): a violating run holds Q at some point, then stays ¬P forever. Every such run
        # has a witnessing reachable Q∧¬P state; seed from all of them and stay within ¬P (once P is
        # hit the obligation is discharged). Same ¬P-subgraph analysis as <>P, different seed.
        triggers = [s for s in g.states if s in bad and leadsto_from(s)]
        if not triggers:
            return LivenessResult(ok=True, states_explored=len(g.states), goal=goal_name, truncated=g.truncated)
        parent, order = _bfs(triggers, lambda s: [(a, t) for (a, t) in g.edges[s] if t in bad])
        reachable_bad = set(order)
    elif everywhere:
        # []<>P: a violating run may start anywhere, so seed from EVERY reachable ¬P state and take
        # stems over the full graph (the prefix may pass through P states before entering a stale cycle).
        parent, order_all = _bfs(g.initial, lambda s: g.edges[s])
        order = [s for s in order_all if s in bad]
        reachable_bad = set(bad)
    else:
        # <>P: a violating run is ¬P throughout, so seed from ¬P inits and stay within ¬P.
        bad_inits = [s for s in g.initial if s in bad]
        if not bad_inits:
            return LivenessResult(ok=True, states_explored=len(g.states), goal=goal_name, truncated=g.truncated)
        parent, order = _bfs(bad_inits, lambda s: [(a, t) for (a, t) in g.edges[s] if t in bad])
        reachable_bad = set(order)

    bad_succ: dict[S, list[tuple[str, S]]] = {
        s: [(a, t) for (a, t) in g.edges[s] if t in reachable_bad] for s in reachable_bad
    }
    can_reach_goal = reaches(g, goal)

    # (1) a reachable ¬P state from which P can never be reached — unconditional violation.
    for s in order:  # BFS order → shortest stem
        if s not in can_reach_goal:
            return LivenessResult(
                ok=False, states_explored=len(g.states), goal=goal_name,
                kind="p-unreachable", stem=_stem(parent, s),
            )

    # (2) a weakly-fair recurring cycle of ¬P states.
    for scc in _tarjan(reachable_bad, bad_succ):
        if not _has_internal_edge(scc, bad_succ):
            continue
        if _fair_admissible(scc, g.edges, fairset, strongset):
            entry = next(s for s in order if s in scc)  # shortest-stem entry
            return LivenessResult(
                ok=False, states_explored=len(g.states), goal=goal_name, kind="fair-cycle",
                stem=_stem(parent, entry), cycle=_simple_cycle(scc, entry, bad_succ),
            )

    return LivenessResult(ok=True, states_explored=len(g.states), goal=goal_name, truncated=g.truncated)

Composition — compose

The interleaved product of independent component models.

musil.compose

Parallel composition of component models (I/O-automata style).

Specify each component independently as a Model (states + named actions), then compose them into one Model whose state is the tuple of component states and whose actions are the interleaving of all components' actions, each lifted to act on its own slice. This is the "specify per component, compose" discipline from the methodology note (docs/specs/formal-modeling-distributed-systems.md): the joint behaviour is every interleaving of the parts, which is exactly where concurrency bugs live.

Component actions are name-qualified ("<component>:<action>") and component invariants are lifted (checked against that component's slice). Pass extra invariants to compose for joint properties that span components. The composite is terminal only when every component is terminal.

Composite dataclass

A joint state: one component state per name, kept sorted by name so it hashes deterministically. Read a component's slice with state["name"].

Source code in src/musil/compose.py

@dataclass(frozen=True, slots=True)
class Composite:
    """A joint state: one component state per name, kept sorted by name so it hashes deterministically.
    Read a component's slice with ``state["name"]``."""

    parts: tuple[tuple[str, Hashable], ...]

    def __getitem__(self, name: str) -> Hashable:
        for n, s in self.parts:
            if n == name:
                return s
        raise KeyError(name)

    def with_part(self, name: str, new: Hashable) -> Composite:
        return Composite(tuple((n, new if n == name else s) for n, s in self.parts))

compose

compose(
    components: Mapping[str, Model[Any]],
    *,
    invariants: Mapping[str, Invariant[Composite]]
    | None = None,
) -> Model[Composite]

Interleaving product of components (name -> Model). The result's reachable states are every interleaving of the parts; component invariants are lifted ("<name>:<inv>") and invariants adds joint properties over the whole composite.

Source code in src/musil/compose.py

def compose(
    components: Mapping[str, Model[Any]],
    *,
    invariants: Mapping[str, Invariant[Composite]] | None = None,
) -> Model[Composite]:
    """Interleaving product of ``components`` (name -> Model). The result's reachable states are every
    interleaving of the parts; component invariants are lifted (``"<name>:<inv>"``) and ``invariants``
    adds joint properties over the whole composite."""
    names = sorted(components)
    init_slices = [[(n, s) for s in components[n].initial_states()] for n in names]
    inits = [Composite(tuple(combo)) for combo in product(*init_slices)]

    actions: list[Action[Composite]] = []
    for n in names:
        for action in components[n].actions:
            actions.append(
                Action(
                    name=f"{n}:{action.name}",
                    enabled=lambda s, _n=n, _a=action: _a.enabled(s[_n]),
                    apply=lambda s, _n=n, _a=action: s.with_part(_n, _a.apply(s[_n])),
                )
            )

    inv: dict[str, Invariant[Composite]] = {}
    for n in names:
        for iname, ipred in components[n].invariants.items():
            inv[f"{n}:{iname}"] = lambda s, _n=n, _p=ipred: _p(s[_n])
    if invariants:
        inv.update(invariants)

    def terminal(s: Composite) -> bool:
        return all(components[n].terminal(s[n]) for n in names)

    return Model(init=inits, actions=actions, invariants=inv, terminal=terminal)

Channels — channel_actions, send

Model a message channel (reliable / lossy / duplicating; unordered, so reordering is free).

musil.channels

Generic message-channel modeling kit -- the "model the network explicitly" principle as a reusable building block (methodology doc section 3).

A channel's in-flight messages are held as a frozenset somewhere in your model's state. This kit builds the channel's actions over that state: deliver a message (apply it to the receiver), and -- depending on the channel's reliability -- drop it (loss) or keep it after delivery (duplication). Because the in-flight set is unordered and deliver may fire for ANY in-flight message, reordering is modelled for free; lossy adds message loss; duplicating allows redelivery. (A set dedups a re-send of the same message -- which matches an idempotent re-push. FIFO/ordered channels would need a sequence instead; the realistic network is unordered, which is what this models.)

Domain-free: messages are any hashable values the caller supplies. Use send inside a sender's action to inject a message, and channel_actions to add the channel's own steps to the model.

send

send(
    get: Callable[[S], frozenset[Message]],
    put: Callable[[S, frozenset[Message]], S],
) -> Callable[[S, Message], S]

Return a helper send(state, msg) that injects msg into the channel's in-flight set (idempotent: re-sending an in-flight message is a no-op, like an idempotent re-push).

Source code in src/musil/channels.py

def send[S: Hashable](
    get: Callable[[S], frozenset[Message]], put: Callable[[S, frozenset[Message]], S]
) -> Callable[[S, Message], S]:
    """Return a helper ``send(state, msg)`` that injects ``msg`` into the channel's in-flight set
    (idempotent: re-sending an in-flight message is a no-op, like an idempotent re-push)."""
    return lambda s, msg: put(s, get(s) | frozenset({msg}))

fifo_send

fifo_send(
    get: Callable[[S], tuple[Message, ...]],
    put: Callable[[S, tuple[Message, ...]], S],
) -> Callable[[S, Message], S]

Return a helper send(state, msg) that appends msg to the tail of an ORDERED channel (a tuple, not a set). Unlike the unordered send, this is not idempotent — each call enqueues another copy — so the caller must bound the queue (e.g. only send when it's empty, or guard on a capacity) to keep the model finite.

Source code in src/musil/channels.py

def fifo_send[S: Hashable](
    get: Callable[[S], tuple[Message, ...]], put: Callable[[S, tuple[Message, ...]], S]
) -> Callable[[S, Message], S]:
    """Return a helper ``send(state, msg)`` that appends ``msg`` to the tail of an ORDERED channel
    (a tuple, not a set). Unlike the unordered ``send``, this is not idempotent — each call enqueues
    another copy — so the caller must bound the queue (e.g. only send when it's empty, or guard on a
    capacity) to keep the model finite."""
    return lambda s, msg: put(s, (*get(s), msg))

fifo_channel_actions

fifo_channel_actions(
    *,
    get: Callable[[S], tuple[Message, ...]],
    put: Callable[[S, tuple[Message, ...]], S],
    deliver: Callable[[S, Message], S],
    lossy: bool = False,
    name: str = "ch",
) -> list[Action[S]]

An ordered (FIFO) channel's steps over the caller's state, the order-preserving counterpart to channel_actions. The in-flight messages are an ordered tuple (get/put read/write it); only the head is deliverable, so messages arrive in send order — no reordering. lossy adds a head-drop (loss). deliver(state, msg) applies the delivered head to the receiver; the head is then removed. Use for sequence-number protocols (alternating-bit, sliding-window) and any link whose correctness assumes order — where the unordered channel_actions would wrongly explore reorderings. (Bound the queue from the sender side; see fifo_send.)

Source code in src/musil/channels.py

def fifo_channel_actions[S: Hashable](
    *,
    get: Callable[[S], tuple[Message, ...]],
    put: Callable[[S, tuple[Message, ...]], S],
    deliver: Callable[[S, Message], S],
    lossy: bool = False,
    name: str = "ch",
) -> list[Action[S]]:
    """An **ordered (FIFO)** channel's steps over the caller's state, the order-preserving counterpart
    to ``channel_actions``. The in-flight messages are an ordered tuple (``get``/``put`` read/write
    it); only the **head** is deliverable, so messages arrive in send order — no reordering. ``lossy``
    adds a head-drop (loss). ``deliver(state, msg)`` applies the delivered head to the receiver; the
    head is then removed. Use for sequence-number protocols (alternating-bit, sliding-window) and any
    link whose correctness assumes order — where the unordered ``channel_actions`` would wrongly
    explore reorderings. (Bound the queue from the sender side; see ``fifo_send``.)"""
    actions: list[Action[S]] = [
        Action(
            name=f"{name}:deliver",
            enabled=lambda s: bool(get(s)),
            apply=lambda s: put(deliver(s, get(s)[0]), get(s)[1:]),  # deliver head, then drop it
        )
    ]
    if lossy:
        actions.append(
            Action(name=f"{name}:drop", enabled=lambda s: bool(get(s)),
                   apply=lambda s: put(s, get(s)[1:]))  # lose the head
        )
    return actions

channel_actions

channel_actions(
    *,
    messages: Iterable[Message],
    get: Callable[[S], frozenset[Message]],
    put: Callable[[S, frozenset[Message]], S],
    deliver: Callable[[S, Message], S],
    lossy: bool = False,
    duplicating: bool = False,
    name: str = "ch",
) -> list[Action[S]]

The channel's steps over the caller's state. get/put read/write the in-flight set; deliver(state, msg) applies a delivered message to the receiver. messages is the (bounded) universe of possible messages. lossy adds drop steps; duplicating lets a message be delivered more than once. Delivery order is unconstrained (reordering).

Source code in src/musil/channels.py

def channel_actions[S: Hashable](
    *,
    messages: Iterable[Message],
    get: Callable[[S], frozenset[Message]],
    put: Callable[[S, frozenset[Message]], S],
    deliver: Callable[[S, Message], S],
    lossy: bool = False,
    duplicating: bool = False,
    name: str = "ch",
) -> list[Action[S]]:
    """The channel's steps over the caller's state. ``get``/``put`` read/write the in-flight set;
    ``deliver(state, msg)`` applies a delivered message to the receiver. ``messages`` is the (bounded)
    universe of possible messages. ``lossy`` adds drop steps; ``duplicating`` lets a message be
    delivered more than once. Delivery order is unconstrained (reordering)."""
    actions: list[Action[S]] = []
    for m in messages:
        actions.append(_deliver_action(f"{name}:deliver:{m!r}", get, put, deliver, m, duplicating))
        if lossy:
            actions.append(_drop_action(f"{name}:drop:{m!r}", get, put, m))
    return actions

Clock — bounded_clock

A discrete clock bounded at a declared horizon, so a timeout model stays finite by construction instead of by hitting max_states — and binding() says whether the bound was reached, i.e. whether a green result covers the whole model or only the first horizon ticks.

musil.clock

A bounded discrete clock for the explicit-state layer -- timeouts without an unbounded space.

sim has a virtual clock; Model/Action has none, so every timeout model (a wait with a deadline, a retry ladder that gives up) hand-rolls the same two things: a tick action, and a bound to stop it running forever. The bound is the part that is easy to get wrong, and getting it wrong is quiet. An unbounded clock does not fail -- check hits max_states and returns Result(ok=True, truncated=True), which reads as success to if result: and prints as OK -- 1000000 states, no violations (TRUNCATED at cap). Nothing was proved and it looks like something was.

bounded_clock makes the bound a declared number instead of an open-coded guard, and makes the question it raises answerable: was the horizon binding? A green result under a binding horizon means "no violation within horizon ticks", which is a weaker claim than "no violation" and should not be quoted as the stronger one.

Domain-free, like the rest of the kit: the clock lives in a field of the caller's own state, read by get and written by put.

clock = bounded_clock(get=lambda s: s.now, put=lambda s, t: replace(s, now=t), horizon=8)
model = Model(
    init=World(),
    actions=[*my_actions, clock.action],
    invariants={...},
    terminal=clock.at_horizon,   # see below -- running out of time is a legitimate sink
)
graph = explore(model)
assert not clock.binding(graph)   # or say so when reporting the result

terminal matters: at the horizon the tick is disabled, so a state whose only remaining move was the tick has no enabled action and check calls it a deadlock. That is a false alarm -- the model did not deadlock, time ran out -- so mark those states terminal (terminal=clock.at_horizon, or lambda s: clock.at_horizon(s) or mine(s)). Forgetting it fails loudly, with a trace ending at the horizon, which is the failure mode to prefer over a silent one.

BoundedClock dataclass

A clock that advances one tick at a time and stops at horizon. Built by bounded_clock.

action is the tick, to be added to the model's actions. now reads the clock out of a state, at_horizon says whether a state sits at the bound, and binding answers the question the bound raises.

Source code in src/musil/clock.py

@dataclass(frozen=True, slots=True)
class BoundedClock[S: Hashable]:
    """A clock that advances one tick at a time and stops at ``horizon``. Built by ``bounded_clock``.

    ``action`` is the tick, to be added to the model's actions. ``now`` reads the clock out of a
    state, ``at_horizon`` says whether a state sits at the bound, and ``binding`` answers the
    question the bound raises."""

    horizon: int
    action: Action[S]
    now: Callable[[S], int]
    at_horizon: Callable[[S], bool]

    def horizon_states(self, graph: Graph[S]) -> tuple[S, ...]:
        """The reachable states sitting at the horizon -- where exploration stopped because the
        clock was not allowed to advance, not because the system had nowhere left to go."""
        return tuple(s for s in graph.states if self.at_horizon(s))

    def binding(self, graph: Graph[S]) -> bool:
        """Did the horizon cut the exploration? False means the model finished on its own with time
        to spare, so the bound cost nothing and a green result is a result about the whole model.
        True means the bound was reached, and a green result only says "no violation within
        ``horizon`` ticks" -- raise the horizon until this goes False, or quote the weaker claim."""
        return bool(self.horizon_states(graph))

horizon_states

horizon_states(graph: Graph[S]) -> tuple[S, ...]

The reachable states sitting at the horizon -- where exploration stopped because the clock was not allowed to advance, not because the system had nowhere left to go.

Source code in src/musil/clock.py

def horizon_states(self, graph: Graph[S]) -> tuple[S, ...]:
    """The reachable states sitting at the horizon -- where exploration stopped because the
    clock was not allowed to advance, not because the system had nowhere left to go."""
    return tuple(s for s in graph.states if self.at_horizon(s))

binding

binding(graph: Graph[S]) -> bool

Did the horizon cut the exploration? False means the model finished on its own with time to spare, so the bound cost nothing and a green result is a result about the whole model. True means the bound was reached, and a green result only says "no violation within horizon ticks" -- raise the horizon until this goes False, or quote the weaker claim.

Source code in src/musil/clock.py

def binding(self, graph: Graph[S]) -> bool:
    """Did the horizon cut the exploration? False means the model finished on its own with time
    to spare, so the bound cost nothing and a green result is a result about the whole model.
    True means the bound was reached, and a green result only says "no violation within
    ``horizon`` ticks" -- raise the horizon until this goes False, or quote the weaker claim."""
    return bool(self.horizon_states(graph))

bounded_clock

bounded_clock(
    *,
    get: Callable[[S], int],
    put: Callable[[S, int], S],
    horizon: int,
    name: str = "clock",
) -> BoundedClock[S]

A tick action over the caller's clock field, bounded at horizon.

get(state) reads the current tick and put(state, t) returns a state with the clock at t. The action is named f"{name}:tick" and is enabled exactly while get(state) < horizon, so the clock contributes at most horizon + 1 values to the state space -- linear in the horizon, and finite by construction rather than by hitting max_states.

horizon must be non-negative. horizon=0 is a clock that never ticks, which is a valid (if pointless) model rather than an error -- the value is then always the initial one.

Deadlines are ordinary guards over get: a timeout fires when get(s) >= armed_at + delay, a retry is enabled while get(s) < give_up_at. This helper does not model those, it only supplies the time they are read from and the bound that keeps them finite.

Source code in src/musil/clock.py

def bounded_clock[S: Hashable](
    *,
    get: Callable[[S], int],
    put: Callable[[S, int], S],
    horizon: int,
    name: str = "clock",
) -> BoundedClock[S]:
    """A tick action over the caller's clock field, bounded at ``horizon``.

    ``get(state)`` reads the current tick and ``put(state, t)`` returns a state with the clock at
    ``t``. The action is named ``f"{name}:tick"`` and is enabled exactly while ``get(state) <
    horizon``, so the clock contributes at most ``horizon + 1`` values to the state space -- linear
    in the horizon, and finite by construction rather than by hitting ``max_states``.

    ``horizon`` must be non-negative. ``horizon=0`` is a clock that never ticks, which is a valid
    (if pointless) model rather than an error -- the value is then always the initial one.

    Deadlines are ordinary guards over ``get``: a timeout fires when ``get(s) >= armed_at + delay``,
    a retry is enabled while ``get(s) < give_up_at``. This helper does not model those, it only
    supplies the time they are read from and the bound that keeps them finite."""
    if horizon < 0:
        raise ValueError(f"horizon must be >= 0, got {horizon}")

    def at_horizon(state: S) -> bool:
        return get(state) >= horizon

    return BoundedClock(
        horizon=horizon,
        action=Action(
            name=f"{name}:tick",
            enabled=lambda s: get(s) < horizon,
            apply=lambda s: put(s, get(s) + 1),
        ),
        now=get,
        at_horizon=at_horizon,
    )

FSM bridge — transition_actions, status_field_actions

Build a model straight from an allowed-transitions table, so it can't drift from the code.

musil.fsm

Build a model straight from a transition table — {from_state: {to_states}} — instead of hand-writing one Action per edge. This is the zero-drift bridge: point it at the very table your code enforces (e.g. a state machine's allowed-transitions map) and the model can never disagree with the code, because it is the code's table. No separate spec file to keep in sync.

Generic over the state type: you supply get (read the status out of a state) and with_status (return a new state with the status replaced), so the table can drive any state shape — typically a frozen dataclass with a status field plus whatever other variables your invariants need.

transition_actions

transition_actions(
    table: Table,
    *,
    get: Callable[[S], str],
    with_status: Callable[[S, str], S],
    name: Callable[[str, str], str] = lambda src, dst: (
        f"{src}->{dst}"
    ),
) -> list[Action[S]]

One Action per declared src -> dst edge. get reads the current status from a state; with_status returns a new state with the status set to dst (states are immutable).

Source code in src/musil/fsm.py

def transition_actions[S: Hashable](
    table: Table,
    *,
    get: Callable[[S], str],
    with_status: Callable[[S, str], S],
    name: Callable[[str, str], str] = lambda src, dst: f"{src}->{dst}",
) -> list[Action[S]]:
    """One Action per declared ``src -> dst`` edge. ``get`` reads the current status from a state;
    ``with_status`` returns a new state with the status set to ``dst`` (states are immutable)."""
    actions: list[Action[S]] = []
    for src, dsts in table.items():
        for dst in dsts:
            actions.append(
                Action(
                    name=name(src, dst),
                    # default-arg capture pins src/dst per loop iteration (avoid late-binding).
                    enabled=lambda s, _src=src: get(s) == _src,
                    apply=lambda s, _dst=dst: with_status(s, _dst),
                )
            )
    return actions

status_field_actions

status_field_actions(
    table: Table, *, field: str = "status"
) -> list[Action[Any]]

Convenience for the common case: states are frozen dataclasses whose status lives in a single field (default status). Builds actions that read/replace that field via dataclasses.replace.

Source code in src/musil/fsm.py

def status_field_actions(table: Table, *, field: str = "status") -> list[Action[Any]]:
    """Convenience for the common case: states are frozen dataclasses whose status lives in a single
    field (default ``status``). Builds actions that read/replace that field via dataclasses.replace."""
    return transition_actions(
        table,
        get=lambda s: getattr(s, field),
        with_status=lambda s, dst: replace(s, **{field: dst}),
    )

multi_status_field_actions

multi_status_field_actions(
    tables: Mapping[str, Table],
) -> list[Action[Any]]

For a dataclass with SEVERAL status fields, each driven by its own table (e.g. a control-plane World whose node/service/cert fields each follow a different ALLOWED[...] map): build every field's edge-actions at once. tables maps a field name to its table; action names are namespaced "<field>:<src>-><dst>" so they stay distinct across fields. musil fires every enabled action from every state, so the interleaved product of the per-field machines — every way the resources can advance concurrently — is explored for free.

Source code in src/musil/fsm.py

def multi_status_field_actions(tables: Mapping[str, Table]) -> list[Action[Any]]:
    """For a dataclass with SEVERAL status fields, each driven by its own table (e.g. a control-plane
    ``World`` whose ``node``/``service``/``cert`` fields each follow a different ``ALLOWED[...]`` map):
    build every field's edge-actions at once. ``tables`` maps a field name to its table; action names
    are namespaced ``"<field>:<src>-><dst>"`` so they stay distinct across fields. musil fires every
    enabled action from every state, so the interleaved product of the per-field machines — every way
    the resources can advance concurrently — is explored for free."""
    actions: list[Action[Any]] = []
    for field, table in tables.items():
        actions.extend(
            transition_actions(
                table,
                get=lambda s, f=field: getattr(s, f),
                with_status=lambda s, dst, f=field: replace(s, **{f: dst}),
                name=lambda src, dst, f=field: f"{f}:{src}->{dst}",
            )
        )
    return actions

protocol_actions

protocol_actions(
    table: Protocol[T],
    *,
    get: Callable[[S], T],
    with_state: Callable[[S, T], S],
    name: Callable[
        [T, str], str
    ] = lambda state, operation: f"{operation} in {state}",
) -> list[Action[S]]

One Action per (state, operation) that moves, from a table of legal operations.

The zero-drift argument, one level up from :func:transition_actions: a protocol table is what a guard needs anyway — "is this call legal where I am, and where does it leave me" — so pointing the checker at the same table means the sequence being verified is the sequence being enforced, not a second description of it.

Operations mapped to None produce no action. They are legal and they stay put, so they add nothing to the reachable state space; a guard still has to know about them, which is why they belong in the table rather than being left out of it.

get reads the protocol state out of your state; with_state returns a new one with it replaced. For a model whose state is the protocol state, those are lambda s: s and lambda _s, t: t.

Source code in src/musil/fsm.py

def protocol_actions[S: Hashable, T](
    table: Protocol[T],
    *,
    get: Callable[[S], T],
    with_state: Callable[[S, T], S],
    name: Callable[[T, str], str] = lambda state, operation: f"{operation} in {state}",
) -> list[Action[S]]:
    """One Action per ``(state, operation)`` that moves, from a table of legal operations.

    The zero-drift argument, one level up from :func:`transition_actions`: a protocol table is
    what a *guard* needs anyway — "is this call legal where I am, and where does it leave me" —
    so pointing the checker at the same table means the sequence being verified is the sequence
    being enforced, not a second description of it.

    Operations mapped to ``None`` produce no action. They are legal and they stay put, so they
    add nothing to the reachable state space; a guard still has to know about them, which is why
    they belong in the table rather than being left out of it.

    ``get`` reads the protocol state out of your state; ``with_state`` returns a new one with it
    replaced. For a model whose state *is* the protocol state, those are ``lambda s: s`` and
    ``lambda _s, t: t``.
    """
    actions: list[Action[S]] = []
    for state, operations in table.items():
        for operation, target in operations.items():
            if target is None or target == state:
                continue
            actions.append(
                Action(
                    # default-arg capture pins state/target per loop iteration (late-binding).
                    name=name(state, operation),
                    enabled=lambda s, _f=state: get(s) == _f,
                    apply=lambda s, _t=target: with_state(s, _t),
                )
            )
    return actions

protocol_operations

protocol_operations(
    table: Protocol[T], state: T
) -> frozenset[str]

Every operation legal in state, moving or not — what a guard checks against.

Source code in src/musil/fsm.py

def protocol_operations[T](table: Protocol[T], state: T) -> frozenset[str]:
    """Every operation legal in ``state``, moving or not — what a guard checks against."""
    return frozenset(table.get(state, {}))

declared_states

declared_states(table: Table) -> set[str]

Every state named in the table — sources (keys) plus every target.

Source code in src/musil/fsm.py

def declared_states(table: Table) -> set[str]:
    """Every state named in the table — sources (keys) plus every target."""
    return set(table) | {dst for dsts in table.values() for dst in dsts}

terminal_states

terminal_states(table: Table) -> set[str]

States with no outgoing transitions — absorbing sinks (a key mapped to an empty set, or a target that's never a source). Use to build a model's terminal predicate so a legitimate resting/sink state isn't reported as a deadlock.

Source code in src/musil/fsm.py

def terminal_states(table: Table) -> set[str]:
    """States with no outgoing transitions — absorbing sinks (a key mapped to an empty set, or a
    target that's never a source). Use to build a model's ``terminal`` predicate so a legitimate
    resting/sink state isn't reported as a deadlock."""
    return {s for s in declared_states(table) if not table.get(s)}

Conformance — generate_traces, replay

Generate traces from the model and replay them against the real implementation.

musil.conformance

Conformance: does the real implementation behave like the model?

Model checking proves the design is sound. Conformance testing checks the code matches the design: generate action traces from the model (sequences that cover the reachable graph), then replay each against the real system through a thin adapter, asserting the implementation lands in the state the model predicted after every step. It does not prove refinement (that's theorem-proving territory), but it checks it exhaustively over the model's traces against the running code — the practical "implementation faithful to spec" bridge (see docs/specs/RESEARCH.md: P, Stateright, ioco).

Two pieces

• generate_traces(model) -> action sequences (lists of Steps, <init> first) that together cover every reachable edge (or state).
• replay(model, traces, step_fn, project, start) -> ConformanceResult. step_fn(real, action) drives the real system; project(real) maps it back to a model state to compare. The first divergence (real projected state != model-predicted state) is reported.

Divergence dataclass

The implementation departed from the model: after action (step step_index of trace trace_index), the model predicted expected but the implementation projected to actual.

Source code in src/musil/conformance.py

@dataclass(frozen=True, slots=True)
class Divergence[S: Hashable]:
    """The implementation departed from the model: after ``action`` (step ``step_index`` of trace
    ``trace_index``), the model predicted ``expected`` but the implementation projected to ``actual``."""

    trace_index: int
    step_index: int
    action: str
    expected: S
    actual: S

generate_traces

generate_traces(
    model: Model[S],
    *,
    cover: str = "edges",
    max_traces: int = 100000,
    max_states: int = 1000000,
) -> list[tuple[Step[S], ...]]

Action sequences from the reachable graph (each a tuple of Steps, <init> first). With cover='edges' (default) the set exercises every reachable transition at least once; with cover='states', a shortest path to each reachable state. Each trace is a shortest stem, so replays stay short.

Source code in src/musil/conformance.py

def generate_traces[S: Hashable](
    model: Model[S], *, cover: str = "edges", max_traces: int = 100_000, max_states: int = 1_000_000
) -> list[tuple[Step[S], ...]]:
    """Action sequences from the reachable graph (each a tuple of Steps, ``<init>`` first). With
    ``cover='edges'`` (default) the set exercises every reachable transition at least once; with
    ``cover='states'``, a shortest path to each reachable state. Each trace is a shortest stem, so
    replays stay short."""
    g = explore(model, max_states=max_states)
    parent, order = _paths(g)
    traces: list[tuple[Step[S], ...]] = []
    if cover == "states":
        for s in order:
            traces.append(_path_to(parent, s))
            if len(traces) >= max_traces:
                break
        return traces
    for s in order:
        base = _path_to(parent, s)
        for action, t in g.edges[s]:
            traces.append((*base, Step(action=action, state=t)))
            if len(traces) >= max_traces:
                return traces
    return traces

replay

replay(
    traces: Iterable[Sequence[Step[S]]],
    *,
    step_fn: Callable[[R, str], R],
    project: Callable[[R], S],
    start: R,
) -> ConformanceResult[S]

Replay each trace (from generate_traces) against the real system. start is the real initial state (its projection must equal the model's initial state); step_fn(real, action) applies an action to the real system; project maps a real state back to a model state for comparison. Single-initial-state models; for multiple inits, project(start) must match each trace's initial state. Use areplay when the real system is async.

Source code in src/musil/conformance.py

def replay[R, S: Hashable](
    traces: Iterable[Sequence[Step[S]]],
    *,
    step_fn: Callable[[R, str], R],
    project: Callable[[R], S],
    start: R,
) -> ConformanceResult[S]:
    """Replay each trace (from ``generate_traces``) against the real system. ``start`` is the real
    initial state (its projection must equal the model's initial state); ``step_fn(real, action)``
    applies an action to the real system; ``project`` maps a real state back to a model state for
    comparison. Single-initial-state models; for multiple inits, project(start) must match each
    trace's initial state. Use ``areplay`` when the real system is async."""
    driver = _drive(traces)
    real = start
    try:
        restart, action, _expected = next(driver)
        while True:
            real = start if restart else step_fn(real, action)
            restart, action, _expected = driver.send(project(real))
    except StopIteration as stop:
        result: ConformanceResult[S] = stop.value
        return result

areplay async

areplay(
    traces: Iterable[Sequence[Step[S]]],
    *,
    step_fn: Callable[[R, str], Awaitable[R]],
    project: Callable[[R], S] | Callable[[R], Awaitable[S]],
    start: R,
) -> ConformanceResult[S]

replay for an async implementation: step_fn is awaited, and project may be either sync or a coroutine function (awaited when it returns an awaitable). Same arguments, same ConformanceResult, same divergence indices -- it drives the same loop as replay.

This exists because the sync replay forces one asyncio.run per step, which cannot be called from a running loop (so no async def test can use it) and, worse, gives every step its own event loop: a connection pool or an open transaction created in one step is dead in the next. Awaiting inside one loop keeps that loop-scoped state alive across a whole trace.

Source code in src/musil/conformance.py

async def areplay[R, S: Hashable](
    traces: Iterable[Sequence[Step[S]]],
    *,
    step_fn: Callable[[R, str], Awaitable[R]],
    project: Callable[[R], S] | Callable[[R], Awaitable[S]],
    start: R,
) -> ConformanceResult[S]:
    """``replay`` for an async implementation: ``step_fn`` is awaited, and ``project`` may be either
    sync or a coroutine function (awaited when it returns an awaitable). Same arguments, same
    ``ConformanceResult``, same divergence indices -- it drives the same loop as ``replay``.

    This exists because the sync ``replay`` forces one ``asyncio.run`` per step, which cannot be
    called from a running loop (so no ``async def`` test can use it) and, worse, gives every step its
    own event loop: a connection pool or an open transaction created in one step is dead in the next.
    Awaiting inside one loop keeps that loop-scoped state alive across a whole trace."""
    driver = _drive(traces)
    real = start
    try:
        restart, action, _expected = next(driver)
        while True:
            real = start if restart else await step_fn(real, action)
            # project may be sync or async; awaiting loses S through isawaitable's TypeGuard
            # (Awaitable[Any]), so the cast restores what the signature already promised.
            candidate = project(real)
            projected = cast("S", await candidate) if isawaitable(candidate) else candidate
            restart, action, _expected = driver.send(projected)
    except StopIteration as stop:
        result: ConformanceResult[S] = stop.value
        return result

Refinement — check_refinement, RefinementMonitor

Check that an observed run of the real system refines the model (each transition is a model edge or a stutter) -- the oracle simulate uses to hold real code to a model-checked spec.

musil.refine

Refinement: hold real code to a model by checking its observed states, not its actions.

A run of the real system produces a sequence of concrete states (snapshots). An abstraction maps each concrete state to a model state. The run refines the model iff every consecutive pair maps to either the same abstract state (a stutter -- the real code took an internal step the model doesn't care about) or to an abstract transition the model actually allows. The first observation must map to a model initial state (unless require_init=False, for checking a mid-run fragment).

This is the runtime analog of refinement in the seL4/TLA+ sense (concrete behaviors are a subset of abstract behaviors), done by checking against the explored model rather than proving -- the practical bridge from a model-checked design to the running implementation. musil.sim feeds a RefinementMonitor a snapshot after every step so the simulated real code is held to the spec.

Complement of conformance (which drives the real code with model-generated action labels); here the real code drives itself and we watch where it goes.

RefinementViolation dataclass

The real run left the model. reason is "bad-init" (first state isn't a model initial state), "unknown-state" (mapped to a state the model can't reach), or "no-model-step" (the abstract transition isn't a model edge). index is the offending observation's position.

Source code in src/musil/refine.py

@dataclass(frozen=True, slots=True)
class RefinementViolation[C, S: Hashable]:
    """The real run left the model. ``reason`` is ``"bad-init"`` (first state isn't a model initial
    state), ``"unknown-state"`` (mapped to a state the model can't reach), or ``"no-model-step"`` (the
    abstract transition isn't a model edge). ``index`` is the offending observation's position."""

    index: int
    reason: str
    concrete_from: C | None
    concrete_to: C
    abstract_from: S | None
    abstract_to: S

    def __str__(self) -> str:
        if self.reason == "bad-init":
            return f"REFINEMENT: observation 0 maps to {self.abstract_to!r}, not a model initial state"
        if self.reason == "unknown-state":
            return f"REFINEMENT: observation {self.index} maps to {self.abstract_to!r}, unreachable in the model"
        return (
            f"REFINEMENT: observation {self.index} steps {self.abstract_from!r} -> {self.abstract_to!r}, "
            f"which the model does not allow"
        )

RefinementMonitor

Streaming refinement check. Construct with a model and an abstraction; feed observed concrete states one at a time to observe, which returns a RefinementViolation the first time the run leaves the model (and None otherwise). The model is explored once, up front, so this is for bounded models (the explicit-state regime musil lives in).

abstraction may be async (it has to read the real system rather than a captured snapshot); in that case feed observations through aobserve, and observe raises TypeError rather than silently abstracting everything to a coroutine object.

Source code in src/musil/refine.py

class RefinementMonitor[C, S: Hashable]:
    """Streaming refinement check. Construct with a model and an abstraction; feed observed concrete
    states one at a time to ``observe``, which returns a ``RefinementViolation`` the first time the
    run leaves the model (and ``None`` otherwise). The model is explored once, up front, so this is
    for bounded models (the explicit-state regime musil lives in).

    ``abstraction`` may be async (it has to read the real system rather than a captured snapshot); in
    that case feed observations through ``aobserve``, and ``observe`` raises ``TypeError`` rather than
    silently abstracting everything to a coroutine object."""

    __slots__ = ("_abs", "_edges", "_index", "_inits", "_prev_abstract", "_prev_concrete", "_require_init", "_started", "_states")

    def __init__(
        self,
        model: Model[S],
        abstraction: Callable[[C], S] | Callable[[C], Awaitable[S]],
        *,
        require_init: bool = True,
        max_states: int = 1_000_000,
    ) -> None:
        g = explore(model, max_states=max_states)
        self._states: frozenset[S] = frozenset(g.states)
        self._inits: frozenset[S] = frozenset(g.initial)
        self._edges: dict[S, frozenset[S]] = {s: frozenset(t for _, t in outs) for s, outs in g.edges.items()}
        self._abs = abstraction
        self._require_init = require_init
        self._started = False
        self._index = -1
        self._prev_concrete: C | None = None
        self._prev_abstract: S | None = None

    def observe(self, concrete: C) -> RefinementViolation[C, S] | None:
        a = self._abs(concrete)
        if isawaitable(a):
            if isinstance(a, Coroutine):
                a.close()  # never awaited; closing keeps Python from warning about it
            raise TypeError(
                "abstraction returned an awaitable: it is async, so use 'await monitor.aobserve(...)' "
                "(or 'acheck_refinement'). Calling observe here would abstract every observation to a "
                "coroutine object, which is in no model, and report a bogus 'unknown-state' violation."
            )
        return self._verdict(concrete, a)

    async def aobserve(self, concrete: C) -> RefinementViolation[C, S] | None:
        """``observe`` for an async abstraction -- one whose mapping to a model state has to await
        (it reads the real system rather than a captured snapshot). A sync abstraction works here
        too, so this is always the safe call inside a coroutine."""
        a = self._abs(concrete)
        resolved = cast("S", await a) if isawaitable(a) else a
        return self._verdict(concrete, resolved)

    def _verdict(self, concrete: C, a: S) -> RefinementViolation[C, S] | None:
        """The refinement decision itself, shared by ``observe`` and ``aobserve``: only how the
        abstract state is obtained differs between them, never what counts as a violation."""
        self._index += 1
        if not self._started:
            self._started = True
            self._prev_concrete, self._prev_abstract = concrete, a
            if a not in self._states:
                return RefinementViolation(self._index, "unknown-state", None, concrete, None, a)
            if self._require_init and a not in self._inits:
                return RefinementViolation(self._index, "bad-init", None, concrete, None, a)
            return None
        prev_a = self._prev_abstract
        if a not in self._states:
            return RefinementViolation(self._index, "unknown-state", self._prev_concrete, concrete, prev_a, a)
        if a != prev_a and a not in self._edges.get(prev_a, frozenset()):  # type: ignore[arg-type]
            return RefinementViolation(self._index, "no-model-step", self._prev_concrete, concrete, prev_a, a)
        self._prev_concrete, self._prev_abstract = concrete, a
        return None

aobserve async

aobserve(concrete: C) -> RefinementViolation[C, S] | None

observe for an async abstraction -- one whose mapping to a model state has to await (it reads the real system rather than a captured snapshot). A sync abstraction works here too, so this is always the safe call inside a coroutine.

Source code in src/musil/refine.py

async def aobserve(self, concrete: C) -> RefinementViolation[C, S] | None:
    """``observe`` for an async abstraction -- one whose mapping to a model state has to await
    (it reads the real system rather than a captured snapshot). A sync abstraction works here
    too, so this is always the safe call inside a coroutine."""
    a = self._abs(concrete)
    resolved = cast("S", await a) if isawaitable(a) else a
    return self._verdict(concrete, resolved)

check_refinement

check_refinement(
    model: Model[S],
    observations: Iterable[C],
    abstraction: Callable[[C], S],
    *,
    require_init: bool = True,
    max_states: int = 1000000,
) -> RefinementResult[C, S]

Check a whole observed run against the model. Returns at the first divergence (with the offending observation), else OK once the run is exhausted. abstraction maps each observed concrete state to a model state; require_init demands the first observation be a model initial state (set False to check a mid-run fragment).

Source code in src/musil/refine.py

def check_refinement[C, S: Hashable](
    model: Model[S],
    observations: Iterable[C],
    abstraction: Callable[[C], S],
    *,
    require_init: bool = True,
    max_states: int = 1_000_000,
) -> RefinementResult[C, S]:
    """Check a whole observed run against the model. Returns at the first divergence (with the
    offending observation), else OK once the run is exhausted. ``abstraction`` maps each observed
    concrete state to a model state; ``require_init`` demands the first observation be a model initial
    state (set False to check a mid-run fragment)."""
    monitor = RefinementMonitor(model, abstraction, require_init=require_init, max_states=max_states)
    checked = 0
    for obs in observations:
        checked += 1
        violation = monitor.observe(obs)
        if violation is not None:
            return RefinementResult(ok=False, steps_checked=checked, violation=violation)
    return RefinementResult(ok=True, steps_checked=checked)

acheck_refinement async

acheck_refinement(
    model: Model[S],
    observations: AsyncIterable[C],
    abstraction: Callable[[C], S]
    | Callable[[C], Awaitable[S]],
    *,
    require_init: bool = True,
    max_states: int = 1000000,
) -> RefinementResult[C, S]

check_refinement over an async source of observations, with an optionally async abstraction. Same result, same first-divergence semantics.

The point is that the observations can be produced by awaiting -- polling a live service, reading a queue, draining an async generator -- without collecting the whole run into a list first. Collecting is the workaround it replaces, and it is not equivalent: a long or unbounded run has to be checked as it happens, and a violation should stop the polling rather than be found afterwards.

Source code in src/musil/refine.py

async def acheck_refinement[C, S: Hashable](
    model: Model[S],
    observations: AsyncIterable[C],
    abstraction: Callable[[C], S] | Callable[[C], Awaitable[S]],
    *,
    require_init: bool = True,
    max_states: int = 1_000_000,
) -> RefinementResult[C, S]:
    """``check_refinement`` over an async source of observations, with an optionally async
    ``abstraction``. Same result, same first-divergence semantics.

    The point is that the observations can be *produced* by awaiting -- polling a live service,
    reading a queue, draining an async generator -- without collecting the whole run into a list
    first. Collecting is the workaround it replaces, and it is not equivalent: a long or unbounded
    run has to be checked as it happens, and a violation should stop the polling rather than be found
    afterwards."""
    monitor = RefinementMonitor(model, abstraction, require_init=require_init, max_states=max_states)
    checked = 0
    async for obs in observations:
        checked += 1
        violation = await monitor.aobserve(obs)
        if violation is not None:
            return RefinementResult(ok=False, steps_checked=checked, violation=violation)
    return RefinementResult(ok=True, steps_checked=checked)

Simulation — simulate, Simulator, NetworkModel, FaultSchedule

Deterministic simulation testing: run real event-driven node code under a controlled clock and a fault-injecting network, reproducible from a seed, with invariants + refinement + a convergence goal as the oracle. Beyond NetworkModel's independent per-message loss, a FaultSchedule of sustained, heal-able faults — network Partitions and node Crashes (with optional cold restart) — drives VOPR-style liveness testing: after the faults heal, the system must converge to its goal (a failure to do so is reported as kind="liveness").

musil.sim

Deterministic simulation: run real, event-driven node code under a controlled world.

The expensive distributed bugs hide in timing -- message loss, reordering, a timeout racing a reply, a partition healing at the wrong moment. This runs your actual node logic against a virtual network and clock where all of that is controlled and seeded, so every run is reproducible from its seed (a found bug replays exactly). It's the FoundationDB/TigerBeetle "deterministic simulation testing" technique as a pure-Python library -- bug finding, not proof.

Model: nodes are event-driven actors (on_start / on_message / on_timer) that act through a Context (send a message, set a timer, read the virtual clock, draw seeded randomness). The Simulator owns a discrete-event queue ordered by (virtual time, sequence); the NetworkModel decides per-message loss / duplication / latency, so reordering and delay emerge from latency spread. A FaultSchedule layers on sustained faults -- network partitions and node crashes that hold for a window then heal -- so simulate can check liveness: does the system converge once they lift? All randomness comes from one seeded RNG. musil.refine + invariants are the oracle (next slice: simulate).

Envelope dataclass

A message in flight: a unique sequence number, source, destination, and payload.

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class Envelope:
    """A message in flight: a unique sequence number, source, destination, and payload."""

    seq: int
    src: str
    dst: str
    payload: object

NetworkModel dataclass

How the virtual network treats each message. Defaults are a reliable, in-order channel; set loss/duplicate (probabilities in [0, 1]) and a min_latency/max_latency spread (virtual-time units) to inject drops, duplicates, and reordering. All draws use the sim's seed.

mutate is a Byzantine fault hook: mutate(src, dst, payload) -> object | None. The returned value replaces the payload before delivery; returning None drops the message. Applied after loss/duplicate decisions, so it targets messages that survive the network model. Models wire-level corruption, wrong-answer injection, or selective drops (Castro & Liskov, OSDI 1999).

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class NetworkModel:
    """How the virtual network treats each message. Defaults are a reliable, in-order channel; set
    ``loss``/``duplicate`` (probabilities in [0, 1]) and a ``min_latency``/``max_latency`` spread
    (virtual-time units) to inject drops, duplicates, and reordering. All draws use the sim's seed.

    ``mutate`` is a Byzantine fault hook: ``mutate(src, dst, payload) -> object | None``. The
    returned value replaces the payload before delivery; returning ``None`` drops the message.
    Applied after loss/duplicate decisions, so it targets messages that survive the network model.
    Models wire-level corruption, wrong-answer injection, or selective drops (Castro & Liskov, OSDI
    1999)."""

    loss: float = 0.0
    duplicate: float = 0.0
    min_latency: int = 1
    max_latency: int = 1
    mutate: Callable[[str, str, object], object | None] | None = None

Partition dataclass

A sustained network fault: every directed link in links ((src, dst) pairs) drops ALL traffic while active -- the window [start, end) in virtual time -- then heals. Unlike NetworkModel.loss (independent per message, which a retrying sender papers over), a partition cuts the link wholesale, so the system can only recover once it lifts. This is what makes a real liveness test: does it converge after the partition heals?

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class Partition:
    """A sustained network fault: every directed link in ``links`` (``(src, dst)`` pairs) drops ALL
    traffic while active -- the window ``[start, end)`` in virtual time -- then heals. Unlike
    ``NetworkModel.loss`` (independent per message, which a retrying sender papers over), a partition
    cuts the link wholesale, so the system can only recover once it lifts. This is what makes a real
    *liveness* test: does it converge after the partition heals?"""

    links: frozenset[tuple[str, str]]
    start: int
    end: int

    def cuts(self, src: str, dst: str, t: int) -> bool:
        return self.start <= t < self.end and (src, dst) in self.links

Crash dataclass

A sustained process fault: node is down over [start, end) -- inbound messages are dropped, its timers don't fire, it runs no handlers -- then restarts. lose_state=True is a cold restart (a fresh node instance that re-runs on_start); the default is a warm resume that keeps the instance's state (it simply missed everything while down).

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class Crash:
    """A sustained process fault: ``node`` is down over ``[start, end)`` -- inbound messages are
    dropped, its timers don't fire, it runs no handlers -- then restarts. ``lose_state=True`` is a cold
    restart (a fresh node instance that re-runs ``on_start``); the default is a warm resume that keeps
    the instance's state (it simply missed everything while down)."""

    node: str
    start: int
    end: int
    lose_state: bool = False

    def down(self, node: str, t: int) -> bool:
        return node == self.node and self.start <= t < self.end

FaultSchedule dataclass

A fixed set of sustained faults for one run. heal_time() is when the last fault lifts (the world is clean again) -- the point after which a live system must converge to its goal. Build one explicitly, or per-seed via a Callable[[Random], FaultSchedule] so randomized schedules stay seed-reproducible.

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class FaultSchedule:
    """A fixed set of sustained faults for one run. ``heal_time()`` is when the last fault lifts (the
    world is clean again) -- the point after which a live system must converge to its goal. Build one
    explicitly, or per-seed via a ``Callable[[Random], FaultSchedule]`` so randomized schedules stay
    seed-reproducible."""

    partitions: tuple[Partition, ...] = ()
    crashes: tuple[Crash, ...] = ()

    def heal_time(self) -> int:
        return max([p.end for p in self.partitions] + [c.end for c in self.crashes], default=0)

    def link_cut(self, src: str, dst: str, t: int) -> bool:
        return any(p.cuts(src, dst, t) for p in self.partitions)

    def node_down(self, node: str, t: int) -> bool:
        return any(c.down(node, t) for c in self.crashes)

    def crash_ending_at(self, node: str, t: int) -> Crash | None:
        for c in self.crashes:
            if c.node == node and c.end == t:
                return c
        return None

RunResult dataclass

Outcome of a run. reason is "quiescent" (queue drained), "until" (goal predicate met), or "max-steps" (step cap hit).

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class RunResult:
    """Outcome of a run. ``reason`` is ``"quiescent"`` (queue drained), ``"until"`` (goal predicate
    met), or ``"max-steps"`` (step cap hit)."""

    steps: int
    time: int
    delivered: int
    dropped: int
    reason: str

Context

A node's handle on the world, passed to every handler. Lets the node send messages, arm/cancel timers, read the virtual clock (now), and draw seeded randomness -- never use real time or the global random module in a node, or runs stop being reproducible.

Source code in src/musil/sim.py

class Context:
    """A node's handle on the world, passed to every handler. Lets the node send messages, arm/cancel
    timers, read the virtual clock (``now``), and draw seeded randomness -- never use real time or the
    global ``random`` module in a node, or runs stop being reproducible."""

    __slots__ = ("_node", "_sim")

    def __init__(self, sim: Simulator, node: str) -> None:
        self._sim = sim
        self._node = node

    @property
    def node(self) -> str:
        return self._node

    @property
    def now(self) -> int:
        return self._sim.now

    def send(self, dst: str, payload: object) -> None:
        self._sim.inject(self._node, dst, payload)

    def set_timer(self, name: str, delay: int = 1) -> None:
        """Arm (or re-arm) a timer; it fires ``delay`` virtual-time units from now via ``on_timer``.
        Re-arming a live timer of the same name replaces it."""
        self._sim.schedule_timer(self._node, name, delay)

    def cancel_timer(self, name: str) -> None:
        self._sim.cancel_timer(self._node, name)

    def random(self) -> float:
        return self._sim.random()

    def randint(self, a: int, b: int) -> int:
        return self._sim.randint(a, b)

set_timer

set_timer(name: str, delay: int = 1) -> None

Arm (or re-arm) a timer; it fires delay virtual-time units from now via on_timer. Re-arming a live timer of the same name replaces it.

Source code in src/musil/sim.py

def set_timer(self, name: str, delay: int = 1) -> None:
    """Arm (or re-arm) a timer; it fires ``delay`` virtual-time units from now via ``on_timer``.
    Re-arming a live timer of the same name replaces it."""
    self._sim.schedule_timer(self._node, name, delay)

Node

Bases: Protocol

What the simulator needs from a node: three event handlers. Subclass BaseNode and override only the ones you need.

Source code in src/musil/sim.py

class Node(Protocol):
    """What the simulator needs from a node: three event handlers. Subclass ``BaseNode`` and override
    only the ones you need."""

    def on_start(self, ctx: Context) -> None: ...
    def on_message(self, ctx: Context, src: str, payload: object) -> None: ...
    def on_timer(self, ctx: Context, name: str) -> None: ...

BaseNode

A node with no-op handlers; override what you need. Keep all mutable state on the instance.

Source code in src/musil/sim.py

class BaseNode:
    """A node with no-op handlers; override what you need. Keep all mutable state on the instance."""

    def on_start(self, ctx: Context) -> None:
        pass

    def on_message(self, ctx: Context, src: str, payload: object) -> None:
        pass

    def on_timer(self, ctx: Context, name: str) -> None:
        pass

AdversarialNode

Bases: BaseNode

A simulation node that injects adversarial behavior via a prioritized behavior list.

Models an external component (a K8s node, an AWS service, a Byzantine peer) that sometimes misbehaves. Each behavior is a (guard, response) pair:

• guard(src, payload) -> bool — returns True when this behavior should fire.
• response(ctx, src, payload) -> None — the adversarial action: send wrong data, send nothing, send duplicate responses, violate the protocol.

On each on_message, behaviors are tried in order; the first matching guard fires and no further behaviors are tried. If no guard matches, the message is silently dropped — the default adversarial posture (an external service that ignores requests).

All randomness in behavior callbacks must go through ctx.random() / ctx.randint() so runs stay reproducible across seeds. This is the simulation counterpart of EnvironmentSpec: same contract theory, different execution model.

Source code in src/musil/sim.py

class AdversarialNode(BaseNode):
    """A simulation node that injects adversarial behavior via a prioritized behavior list.

    Models an external component (a K8s node, an AWS service, a Byzantine peer) that sometimes
    misbehaves. Each behavior is a ``(guard, response)`` pair:

    - ``guard(src, payload) -> bool``  — returns True when this behavior should fire.
    - ``response(ctx, src, payload) -> None``  — the adversarial action: send wrong data,
      send nothing, send duplicate responses, violate the protocol.

    On each ``on_message``, behaviors are tried in order; the first matching guard fires and
    no further behaviors are tried. If no guard matches, the message is silently dropped —
    the default adversarial posture (an external service that ignores requests).

    All randomness in behavior callbacks must go through ``ctx.random()`` / ``ctx.randint()``
    so runs stay reproducible across seeds. This is the simulation counterpart of
    ``EnvironmentSpec``: same contract theory, different execution model.
    """

    def __init__(
        self,
        behaviors: list[
            tuple[
                Callable[[str, object], bool],         # guard(src, payload)
                Callable[[Context, str, object], None], # response(ctx, src, payload)
            ]
        ],
    ) -> None:
        self._behaviors = behaviors

    def on_message(self, ctx: Context, src: str, payload: object) -> None:
        for guard, response in self._behaviors:
            if guard(src, payload):
                response(ctx, src, payload)
                return

Simulator

Runs a fixed set of named nodes under a virtual clock and network. Construct with the node map, a seed, and an optional NetworkModel; call run. Determinism: identical (nodes, seed, network) always produce the identical run.

Source code in src/musil/sim.py

class Simulator:
    """Runs a fixed set of named nodes under a virtual clock and network. Construct with the node
    map, a ``seed``, and an optional ``NetworkModel``; call ``run``. Determinism: identical
    (nodes, seed, network) always produce the identical run."""

    __slots__ = (
        "_active_timers", "_counter", "_ctxs", "_delivered", "_dropped", "_events",
        "_faults", "_net", "_nodes", "_now", "_queue", "_rebuild", "_rng", "_started",
    )

    def __init__(
        self,
        nodes: Mapping[str, Node],
        *,
        seed: int = 0,
        network: NetworkModel | None = None,
        faults: FaultSchedule | None = None,
        rebuild: Callable[[], Mapping[str, Node]] | None = None,
    ) -> None:
        self._nodes: dict[str, Node] = dict(nodes)
        self._net = network if network is not None else NetworkModel()
        self._faults = faults
        # how to build a fresh node for a cold restart (Crash.lose_state); the per-node fresh
        # instance is taken from a fresh full map, so the same factory simulate already uses works.
        self._rebuild = rebuild
        self._rng = Random(seed)
        self._now: int = 0
        self._counter: int = 0
        self._delivered: int = 0
        self._dropped: int = 0
        self._queue: list[tuple[int, int]] = []  # (due_time, seq) min-heap
        self._events: dict[int, _Event] = {}
        self._active_timers: dict[tuple[str, str], int] = {}
        self._started: set[str] = set()
        self._ctxs: dict[str, Context] = {name: Context(self, name) for name in self._nodes}

    @property
    def nodes(self) -> Mapping[str, Node]:
        """The live node map -- the authoritative instances, which a cold restart swaps. The oracle
        snapshots through this so it always reads post-restart state."""
        return self._nodes

    @property
    def now(self) -> int:
        return self._now

    def _schedule(self, due: int, event: _Event) -> int:
        self._counter += 1
        seq = self._counter
        self._events[seq] = event
        heapq.heappush(self._queue, (due, seq))
        return seq

    def random(self) -> float:
        """A seeded random float in [0, 1) -- the only randomness nodes may use (keeps runs reproducible)."""
        return self._rng.random()

    def randint(self, a: int, b: int) -> int:
        """A seeded random integer in [a, b]."""
        return self._rng.randint(a, b)

    def inject(self, src: str, dst: str, payload: object) -> None:
        """Send a message (from a node's ``ctx.send``, or externally to kick the system). Subject to
        the network model: it may be dropped, duplicated, or delayed."""
        if dst not in self._nodes:
            raise ValueError(f"send to unknown node {dst!r} (known: {sorted(self._nodes)})")
        if self._faults is not None and self._faults.link_cut(src, dst, self._now):
            self._dropped += 1  # link partitioned right now -> the message never makes it on
            return
        net = self._net
        if net.loss > 0.0 and self._rng.random() < net.loss:
            self._dropped += 1
            return
        if net.mutate is not None:
            payload = net.mutate(src, dst, payload)
            if payload is None:
                self._dropped += 1
                return
        latency = self._rng.randint(net.min_latency, net.max_latency)
        env = Envelope(seq=self._counter + 1, src=src, dst=dst, payload=payload)
        self._schedule(self._now + latency, _Deliver(env))
        if net.duplicate > 0.0 and self._rng.random() < net.duplicate:
            dup_latency = self._rng.randint(net.min_latency, net.max_latency)
            self._schedule(self._now + dup_latency, _Deliver(env))

    def schedule_timer(self, node: str, name: str, delay: int) -> None:
        """Arm a timer for ``node`` that fires ``delay`` units from now (re-arming replaces it)."""
        seq = self._schedule(self._now + max(delay, 1), _Timer(node, name))
        self._active_timers[(node, name)] = seq

    def cancel_timer(self, node: str, name: str) -> None:
        self._active_timers.pop((node, name), None)

    def run(
        self,
        *,
        max_steps: int = 10_000,
        until: Callable[[Simulator], bool] | None = None,
        on_step: Callable[[Simulator], None] | None = None,
        after_start: Callable[[Simulator], None] | None = None,
    ) -> RunResult:
        """Drive the system to quiescence (queue empty), until ``until(sim)`` is true, or until
        ``max_steps`` events have fired. ``after_start(sim)`` runs once after every node's
        ``on_start`` (the initial world); ``on_step(sim)`` runs after each fired event -- the hooks
        the oracle uses to snapshot and check the world."""
        for name in sorted(self._nodes):
            if self._faults is not None and self._faults.node_down(name, 0):
                continue  # crashed at t=0 -> it boots when its window ends (its _Restart event)
            self._nodes[name].on_start(self._ctxs[name])
            self._started.add(name)
        if self._faults is not None:
            for crash in self._faults.crashes:
                # a restart event both reboots the node AND keeps the queue alive until heal, so a
                # run that would otherwise go quiescent mid-crash still waits for recovery.
                self._schedule(crash.end, _Restart(crash.node))
        if after_start is not None:
            after_start(self)
        if until is not None and until(self):
            return RunResult(0, self._now, self._delivered, self._dropped, "until")

        steps = 0
        reason = "quiescent"
        while self._queue:
            if steps >= max_steps:
                reason = "max-steps"
                break
            due, seq = heapq.heappop(self._queue)
            event = self._events.pop(seq)
            if isinstance(event, _Timer):
                if self._active_timers.get((event.node, event.name)) != seq:
                    continue  # cancelled or superseded by a re-arm
                self._active_timers.pop((event.node, event.name), None)
                if self._faults is not None and self._faults.node_down(event.node, due):
                    continue  # node is down -> its timer is lost
                self._now = due
                self._nodes[event.node].on_timer(self._ctxs[event.node], event.name)
            elif isinstance(event, _Restart):
                self._now = due
                self._restart(event.node, due)
            else:
                env = event.env
                if self._faults is not None and self._faults.node_down(env.dst, due):
                    self._dropped += 1
                    continue  # destination is down -> message dropped on arrival
                self._now = due
                self._delivered += 1
                self._nodes[env.dst].on_message(self._ctxs[env.dst], env.src, env.payload)
            steps += 1
            if on_step is not None:
                on_step(self)
            if until is not None and until(self):
                reason = "until"
                break

        return RunResult(steps, self._now, self._delivered, self._dropped, reason)

    def _restart(self, node: str, t: int) -> None:
        """Bring ``node`` back up at the end of its crash window. A cold restart (``lose_state``) swaps
        in a fresh instance from the factory and re-runs ``on_start``; a warm one resumes the existing
        instance (and runs ``on_start`` only if the node was down from t=0 and never booted)."""
        crash = self._faults.crash_ending_at(node, t) if self._faults is not None else None
        if crash is not None and crash.lose_state and self._rebuild is not None:
            self._nodes[node] = self._rebuild()[node]
            self._started.discard(node)
        if node not in self._started:
            self._nodes[node].on_start(self._ctxs[node])
            self._started.add(node)

nodes property

nodes: Mapping[str, Node]

The live node map -- the authoritative instances, which a cold restart swaps. The oracle snapshots through this so it always reads post-restart state.

random

random() -> float

A seeded random float in [0, 1) -- the only randomness nodes may use (keeps runs reproducible).

Source code in src/musil/sim.py

def random(self) -> float:
    """A seeded random float in [0, 1) -- the only randomness nodes may use (keeps runs reproducible)."""
    return self._rng.random()

randint

randint(a: int, b: int) -> int

A seeded random integer in [a, b].

Source code in src/musil/sim.py

def randint(self, a: int, b: int) -> int:
    """A seeded random integer in [a, b]."""
    return self._rng.randint(a, b)

inject

inject(src: str, dst: str, payload: object) -> None

Send a message (from a node's ctx.send, or externally to kick the system). Subject to the network model: it may be dropped, duplicated, or delayed.

Source code in src/musil/sim.py

def inject(self, src: str, dst: str, payload: object) -> None:
    """Send a message (from a node's ``ctx.send``, or externally to kick the system). Subject to
    the network model: it may be dropped, duplicated, or delayed."""
    if dst not in self._nodes:
        raise ValueError(f"send to unknown node {dst!r} (known: {sorted(self._nodes)})")
    if self._faults is not None and self._faults.link_cut(src, dst, self._now):
        self._dropped += 1  # link partitioned right now -> the message never makes it on
        return
    net = self._net
    if net.loss > 0.0 and self._rng.random() < net.loss:
        self._dropped += 1
        return
    if net.mutate is not None:
        payload = net.mutate(src, dst, payload)
        if payload is None:
            self._dropped += 1
            return
    latency = self._rng.randint(net.min_latency, net.max_latency)
    env = Envelope(seq=self._counter + 1, src=src, dst=dst, payload=payload)
    self._schedule(self._now + latency, _Deliver(env))
    if net.duplicate > 0.0 and self._rng.random() < net.duplicate:
        dup_latency = self._rng.randint(net.min_latency, net.max_latency)
        self._schedule(self._now + dup_latency, _Deliver(env))

schedule_timer

schedule_timer(node: str, name: str, delay: int) -> None

Arm a timer for node that fires delay units from now (re-arming replaces it).

Source code in src/musil/sim.py

def schedule_timer(self, node: str, name: str, delay: int) -> None:
    """Arm a timer for ``node`` that fires ``delay`` units from now (re-arming replaces it)."""
    seq = self._schedule(self._now + max(delay, 1), _Timer(node, name))
    self._active_timers[(node, name)] = seq

run

run(
    *,
    max_steps: int = 10000,
    until: Callable[[Simulator], bool] | None = None,
    on_step: Callable[[Simulator], None] | None = None,
    after_start: Callable[[Simulator], None] | None = None,
) -> RunResult

Drive the system to quiescence (queue empty), until until(sim) is true, or until max_steps events have fired. after_start(sim) runs once after every node's on_start (the initial world); on_step(sim) runs after each fired event -- the hooks the oracle uses to snapshot and check the world.

Source code in src/musil/sim.py

def run(
    self,
    *,
    max_steps: int = 10_000,
    until: Callable[[Simulator], bool] | None = None,
    on_step: Callable[[Simulator], None] | None = None,
    after_start: Callable[[Simulator], None] | None = None,
) -> RunResult:
    """Drive the system to quiescence (queue empty), until ``until(sim)`` is true, or until
    ``max_steps`` events have fired. ``after_start(sim)`` runs once after every node's
    ``on_start`` (the initial world); ``on_step(sim)`` runs after each fired event -- the hooks
    the oracle uses to snapshot and check the world."""
    for name in sorted(self._nodes):
        if self._faults is not None and self._faults.node_down(name, 0):
            continue  # crashed at t=0 -> it boots when its window ends (its _Restart event)
        self._nodes[name].on_start(self._ctxs[name])
        self._started.add(name)
    if self._faults is not None:
        for crash in self._faults.crashes:
            # a restart event both reboots the node AND keeps the queue alive until heal, so a
            # run that would otherwise go quiescent mid-crash still waits for recovery.
            self._schedule(crash.end, _Restart(crash.node))
    if after_start is not None:
        after_start(self)
    if until is not None and until(self):
        return RunResult(0, self._now, self._delivered, self._dropped, "until")

    steps = 0
    reason = "quiescent"
    while self._queue:
        if steps >= max_steps:
            reason = "max-steps"
            break
        due, seq = heapq.heappop(self._queue)
        event = self._events.pop(seq)
        if isinstance(event, _Timer):
            if self._active_timers.get((event.node, event.name)) != seq:
                continue  # cancelled or superseded by a re-arm
            self._active_timers.pop((event.node, event.name), None)
            if self._faults is not None and self._faults.node_down(event.node, due):
                continue  # node is down -> its timer is lost
            self._now = due
            self._nodes[event.node].on_timer(self._ctxs[event.node], event.name)
        elif isinstance(event, _Restart):
            self._now = due
            self._restart(event.node, due)
        else:
            env = event.env
            if self._faults is not None and self._faults.node_down(env.dst, due):
                self._dropped += 1
                continue  # destination is down -> message dropped on arrival
            self._now = due
            self._delivered += 1
            self._nodes[env.dst].on_message(self._ctxs[env.dst], env.src, env.payload)
        steps += 1
        if on_step is not None:
            on_step(self)
        if until is not None and until(self):
            reason = "until"
            break

    return RunResult(steps, self._now, self._delivered, self._dropped, reason)

SimFailure dataclass

A reproducible failure from simulate. kind is "invariant", "refinement", "goal" (settled-but-goal-false with no faults), or "liveness" (under a FaultSchedule, the system did not converge to the goal after the faults healed). detail names the broken invariant / refinement violation / goal. world is the offending snapshot and history the snapshots leading to it (a trace). Re-run with seeds=[seed] to reproduce exactly.

Source code in src/musil/sim.py

@dataclass(frozen=True, slots=True)
class SimFailure[W]:
    """A reproducible failure from ``simulate``. ``kind`` is ``"invariant"``, ``"refinement"``,
    ``"goal"`` (settled-but-goal-false with no faults), or ``"liveness"`` (under a ``FaultSchedule``,
    the system did not converge to the goal after the faults healed). ``detail`` names the broken
    invariant / refinement violation / goal. ``world`` is the offending snapshot and ``history`` the
    snapshots leading to it (a trace). Re-run with ``seeds=[seed]`` to reproduce exactly."""

    seed: int
    kind: str
    detail: str
    step: int
    world: W
    history: tuple[W, ...]

    def __str__(self) -> str:
        return (
            f"{self.kind.upper()} on seed {self.seed} at step {self.step}: {self.detail}\n"
            f"  world: {self.world}\n"
            f"  reproduce with seeds=[{self.seed}]"
        )

simulate

simulate(
    factory: Callable[[], Mapping[str, Node]],
    *,
    seeds: Iterable[int],
    snapshot: Callable[[Mapping[str, Node]], W],
    network: NetworkModel | None = None,
    faults: FaultSchedule
    | Callable[[Random], FaultSchedule]
    | None = None,
    invariants: Mapping[str, Callable[[W], bool]]
    | None = None,
    model: Model[S] | None = None,
    abstraction: Callable[[W], S] | None = None,
    goal: Callable[[W], bool] | None = None,
    goal_name: str = "goal",
    converge_within: int | None = None,
    max_steps: int = 10000,
    require_init: bool = True,
) -> SimReport[W]

Deterministic simulation testing of the real nodes. Runs the system once per seed under network (independent per-message faults) and an optional faults schedule (sustained, heal-able partitions / crashes -- a FaultSchedule or a per-seed Random -> FaultSchedule). After every step it checks each invariant and (if a model + abstraction are given) that the run refines the model -- safety holds during the chaos too. If a goal is given, the settled final world must satisfy it: with a non-trivial faults schedule this is the liveness check (did the system converge once the faults healed? -> kind="liveness"); with no faults it is the plain convergence check (kind="goal"). converge_within (steps after the schedule's heal_time) catches livelock early. Returns at the FIRST failing seed -- a reproducible repro (report.failure.seed). factory builds a fresh node map per run (also reused to cold-restart a crashed node); snapshot maps the live nodes to an immutable world value used by the oracle.

Source code in src/musil/sim.py

def simulate[W, S: Hashable](
    factory: Callable[[], Mapping[str, Node]],
    *,
    seeds: Iterable[int],
    snapshot: Callable[[Mapping[str, Node]], W],
    network: NetworkModel | None = None,
    faults: FaultSchedule | Callable[[Random], FaultSchedule] | None = None,
    invariants: Mapping[str, Callable[[W], bool]] | None = None,
    model: Model[S] | None = None,
    abstraction: Callable[[W], S] | None = None,
    goal: Callable[[W], bool] | None = None,
    goal_name: str = "goal",
    converge_within: int | None = None,
    max_steps: int = 10_000,
    require_init: bool = True,
) -> SimReport[W]:
    """Deterministic simulation testing of the real nodes. Runs the system once per seed under
    ``network`` (independent per-message faults) and an optional ``faults`` schedule (sustained,
    heal-able partitions / crashes -- a ``FaultSchedule`` or a per-seed ``Random -> FaultSchedule``).
    After every step it checks each invariant and (if a ``model`` + ``abstraction`` are given) that the
    run refines the model -- safety holds *during* the chaos too. If a ``goal`` is given, the settled
    final world must satisfy it: with a non-trivial ``faults`` schedule this is the **liveness** check
    (did the system converge once the faults healed? -> ``kind="liveness"``); with no faults it is the
    plain convergence check (``kind="goal"``). ``converge_within`` (steps after the schedule's
    ``heal_time``) catches livelock early. Returns at the FIRST failing seed -- a reproducible repro
    (``report.failure.seed``). ``factory`` builds a fresh node map per run (also reused to cold-restart
    a crashed node); ``snapshot`` maps the live nodes to an immutable world value used by the oracle."""
    if model is not None and abstraction is None:
        raise ValueError("model given without abstraction: pass abstraction=... to map worlds to model states")
    invs: Mapping[str, Callable[[W], bool]] = invariants if invariants is not None else {}
    runs = 0
    for seed in seeds:
        runs += 1
        schedule = faults(Random(seed)) if callable(faults) else faults
        heal_time = schedule.heal_time() if schedule is not None else 0
        nodes = factory()
        sim = Simulator(nodes, seed=seed, network=network, faults=schedule, rebuild=factory)
        monitor: RefinementMonitor[W, S] | None = (
            RefinementMonitor(model, abstraction, require_init=require_init)
            if model is not None and abstraction is not None
            else None
        )
        oracle = _Oracle(
            seed, sim, snapshot, invs, monitor,
            goal=goal, goal_name=goal_name, heal_time=heal_time, converge_within=converge_within,
        )
        try:
            sim.run(max_steps=max_steps, after_start=oracle, on_step=oracle)
        except _OracleStop:
            return SimReport(ok=False, runs=runs, failure=oracle.failure)
        if goal is not None:
            final = oracle.history[-1] if oracle.history else snapshot(sim.nodes)
            if not goal(final):
                kind = "liveness" if heal_time > 0 else "goal"
                step = len(oracle.history) - 1
                return SimReport(
                    ok=False,
                    runs=runs,
                    failure=SimFailure(seed, kind, goal_name, step, final, tuple(oracle.history)),
                )
    return SimReport(ok=True, runs=runs)

Open systems — check_open, EnvironmentSpec, Assumption

Verify a system against adversarial external environments. Each EnvironmentSpec is a contract: the non-deterministic behaviors the environment can exhibit (eviction, crash, wrong answer), the guarantees it commits to, and the assumptions those guarantees rest on (following seL4's two-layer assurance model: Klein et al., SOSP 2009). check_open composes by merging environment behaviors into the system's action set (so the BFS explores every adversarial move) and environment guarantees into the invariants. The result surface all Assumption objects that are not yet "verified" as residual proof obligations.

musil.env

Open-system verification: check a system against adversarial external environments.

Real distributed systems do not operate in isolation. A service runs on Linux, inside K8s, talking to AWS -- components that can crash, evict, throttle, or return wrong answers at any moment. musil's closed-system check assumes the entire state space is under the model's control. check_open lifts that restriction: each external component is an EnvironmentSpec -- a contract consisting of the non-deterministic behaviors it can exhibit, the guarantees it commits to, and the assumptions those guarantees rest on (following seL4's two-layer assurance model: Klein et al., SOSP 2009).

check_open composes by merging: env behaviors become additional actions (the BFS explores every thing the environment can do), env guarantees become additional invariants. This is Interface Automata composition (de Alfaro & Henzinger, ESEC/FSE 2001) -- the environment acts adversarially within its contract; the system must handle all of it.

Assumptions are first-class: they are printed in the result, never silently swallowed. A guarantee that rests on an unverified assumption is weaker than one that rests on a verified proof -- the caller knows exactly what the verification is worth.

Assumption dataclass

A named proof obligation taken on faith or verified elsewhere.

status tracks where the assumption stands: - "axiom" — taken as a foundation; not expected to be proved (e.g., hardware correctness) - "verified" — machine-checked or formally proved (e.g., seL4 kernel proof) - "unverified" — stated but not yet validated (the default; should be addressed)

source cites where the assumption comes from (paper, spec, URL). predicate is an optional machine-checkable version of the assumption; None for assumptions that cannot be expressed as a state predicate (e.g., "network eventually delivers").

Source code in src/musil/env.py

@dataclass(frozen=True, slots=True)
class Assumption:
    """A named proof obligation taken on faith or verified elsewhere.

    ``status`` tracks where the assumption stands:
    - ``"axiom"``      — taken as a foundation; not expected to be proved (e.g., hardware correctness)
    - ``"verified"``   — machine-checked or formally proved (e.g., seL4 kernel proof)
    - ``"unverified"`` — stated but not yet validated (the default; should be addressed)

    ``source`` cites where the assumption comes from (paper, spec, URL).
    ``predicate`` is an optional machine-checkable version of the assumption; ``None`` for
    assumptions that cannot be expressed as a state predicate (e.g., "network eventually delivers").
    """

    name: str
    description: str
    status: Literal["axiom", "verified", "unverified"] = "unverified"
    source: str = ""
    predicate: Callable[[Any], bool] | None = None

EnvironmentSpec dataclass

A contract for one external component: what it can do, what it promises, and what it assumes.

behaviors — the non-deterministic actions the environment can take. Each is an Action[S] that fires from the same global state the system uses. The BFS in check_open explores every possible environment action from every state, so the system is checked against the worst possible environment (within the contract). Model each distinct failure mode -- crash, evict, wrong-answer, timeout -- as a separate behavior.

guarantees — invariants the environment commits to hold in every reachable state. If the environment breaks one, check_open reports it as "<env.name>:<guarantee_name>".

assumptions — named Assumption objects the guarantees rest on. Unverified assumptions are surfaced in OpenResult.unverified_assumptions; they are never silently dropped.

Source code in src/musil/env.py

@dataclass(slots=True)
class EnvironmentSpec[S: Hashable]:
    """A contract for one external component: what it can do, what it promises, and what it assumes.

    ``behaviors`` — the non-deterministic actions the environment can take. Each is an ``Action[S]``
    that fires from the *same* global state the system uses. The BFS in ``check_open`` explores
    every possible environment action from every state, so the system is checked against the worst
    possible environment (within the contract). Model each distinct failure mode -- crash, evict,
    wrong-answer, timeout -- as a separate behavior.

    ``guarantees`` — invariants the environment commits to hold in every reachable state. If the
    environment breaks one, ``check_open`` reports it as ``"<env.name>:<guarantee_name>"``.

    ``assumptions`` — named ``Assumption`` objects the guarantees rest on. Unverified assumptions
    are surfaced in ``OpenResult.unverified_assumptions``; they are never silently dropped.
    """

    name: str
    behaviors: list[Action[S]]
    guarantees: Mapping[str, Invariant[S]] = field(default_factory=dict[str, Invariant[S]])
    assumptions: Mapping[str, Assumption] = field(default_factory=dict[str, Assumption])

OpenResult dataclass

The outcome of check_open. Wraps the underlying Result and surfaces every Assumption that is not yet "verified" -- these are the residual proof obligations the caller must discharge for the verification to be unconditional.

bool(open_result) delegates to result.ok so it works in assertions.

Source code in src/musil/env.py

@dataclass(frozen=True, slots=True)
class OpenResult[S: Hashable]:
    """The outcome of ``check_open``. Wraps the underlying ``Result`` and surfaces every
    ``Assumption`` that is not yet ``"verified"`` -- these are the residual proof obligations
    the caller must discharge for the verification to be unconditional.

    ``bool(open_result)`` delegates to ``result.ok`` so it works in assertions."""

    result: Result[S]
    unverified_assumptions: tuple[Assumption, ...]

    @property
    def ok(self) -> bool:
        return self.result.ok

    def __bool__(self) -> bool:
        return self.ok

    def __str__(self) -> str:
        lines = [str(self.result)]
        if self.unverified_assumptions:
            lines.append("Unverified assumptions (residual proof obligations):")
            for a in self.unverified_assumptions:
                src = f"  [{a.source}]" if a.source else ""
                lines.append(f"  [{a.status}] {a.name}: {a.description}{src}")
        return "\n".join(lines)

check_open

check_open(
    system: Model[S],
    *envs: EnvironmentSpec[S],
    max_states: int = 1000000,
) -> OpenResult[S]

Check system against adversarial external environments.

Composes the system and every environment by merging: - all env.behaviors into the system's actions (the BFS explores every env action) - all env.guarantees into the system's invariants (qualified as "<env.name>:<k>")

Then calls the standard check on the composed model.

check_open(system) with no environments is identical to check(system).

Unverified assumptions from all environments are collected and returned in OpenResult.unverified_assumptions -- they are the residual proof obligations.

Source code in src/musil/env.py

def check_open[S: Hashable](
    system: Model[S],
    *envs: EnvironmentSpec[S],
    max_states: int = 1_000_000,
) -> OpenResult[S]:
    """Check ``system`` against adversarial external environments.

    Composes the system and every environment by merging:
    - all ``env.behaviors`` into the system's actions (the BFS explores every env action)
    - all ``env.guarantees`` into the system's invariants (qualified as ``"<env.name>:<k>"``)

    Then calls the standard ``check`` on the composed model.

    ``check_open(system)`` with no environments is identical to ``check(system)``.

    Unverified assumptions from all environments are collected and returned in
    ``OpenResult.unverified_assumptions`` -- they are the residual proof obligations.
    """
    all_actions = list(system.actions)
    for env in envs:
        all_actions.extend(env.behaviors)

    all_invariants: dict[str, Invariant[S]] = dict(system.invariants)
    for env in envs:
        for k, inv in env.guarantees.items():
            all_invariants[f"{env.name}:{k}"] = inv

    composed = Model(
        init=system.init,
        actions=all_actions,
        invariants=all_invariants,
        terminal=system.terminal,
    )
    result = check(composed, max_states=max_states)

    unverified: list[Assumption] = []
    for env in envs:
        for assumption in env.assumptions.values():
            if assumption.status != "verified":
                unverified.append(assumption)

    return OpenResult(result=result, unverified_assumptions=tuple(unverified))

Metrics & coverage — state_space_metrics, coverage, solution_cardinality

How big and how branchy a design's reachable behaviour is, how many reachable states satisfy a goal, and — because a green check does not say the behaviour you wrote was ever explored — which actions never fired, were always enabled, or only ever self-looped.

musil.metrics

State-space metrics — how big and how branchy a design's reachable behaviour is.

Inspired by "expression cardinality" (given a semantic intent, how many valid programs express it?): for a design we don't have to guess a per-language score — musil already enumerates the whole reachable state graph, so we can MEASURE the exact analog on the behaviour itself.

• state_space_metrics reports the size and branching of a model: reachable states, transitions, terminals, and per-state branching (distinct successors, self-loops excluded — an Idle action shouldn't read as a choice). Lower branching + a tighter state space = a more deterministic, more reviewable design. A blow-up is the MDD "design for modelability" smell: behaviour you can't reason about, to reshape before it hides a bug.
• solution_cardinality answers the cardinality question directly: how many reachable states satisfy a goal. Exact over the explored space, not a guessed 1-10.

Deliberately NOT a single fabricated "determinism score" — these are honest counts on your model.

StateSpaceMetrics dataclass

The shape of a model's reachable behaviour. mean_branching/max_branching count DISTINCT successor states (excluding self-loops) — the real fan-out of choices; terminal_states have none. truncated means the max_states cap was hit, so the counts are lower bounds.

Source code in src/musil/metrics.py

@dataclass(frozen=True, slots=True)
class StateSpaceMetrics:
    """The shape of a model's reachable behaviour. ``mean_branching``/``max_branching`` count DISTINCT
    successor states (excluding self-loops) — the real fan-out of choices; ``terminal_states`` have
    none. ``truncated`` means the ``max_states`` cap was hit, so the counts are lower bounds."""

    reachable_states: int
    transitions: int
    initial_states: int
    terminal_states: int
    max_branching: int
    mean_branching: float
    truncated: bool

    def __str__(self) -> str:
        cap = " (TRUNCATED — counts are lower bounds)" if self.truncated else ""
        return (
            f"{self.reachable_states} states, {self.transitions} transitions{cap}\n"
            f"  branching: mean {self.mean_branching:.2f}, max {self.max_branching} "
            f"(distinct successors, self-loops excluded)\n"
            f"  {self.initial_states} initial, {self.terminal_states} terminal"
        )

Coverage dataclass

Which of a model's actions actually did anything over the reachable space.

A green check says no invariant broke; it does not say the behaviour you wrote was ever explored. These are the three ways that goes wrong, as sets of names rather than accusations — each has legitimate cases, and only the author knows which.

• dead_actions: enabled in NO reachable state. The sharpest of the three: an action that can never fire is a guard bug or dead code, and it silently removes an interleaving the author believed was covered. A run that never explored it still reports OK.
• always_enabled_actions: enabled in EVERY reachable state, i.e. the guard never guards. Often intentional — an Idle step, or an adversarial crash that may happen at any moment — so this is a smell to confirm, not a defect. Empty whenever the model has a terminal state, since nothing is enabled there.
• stutter_only_actions: enabled somewhere, but every firing lands back on the state it came from. The action fires and changes nothing.

truncated carries over from the graph and changes what dead_actions means: under a cap, an action is only not yet seen enabled, which is a weaker claim than dead.

Source code in src/musil/metrics.py

@dataclass(frozen=True, slots=True)
class Coverage:
    """Which of a model's actions actually did anything over the reachable space.

    A green ``check`` says no invariant broke; it does not say the behaviour you wrote was ever
    explored. These are the three ways that goes wrong, as *sets of names* rather than accusations —
    each has legitimate cases, and only the author knows which.

    - ``dead_actions``: enabled in NO reachable state. The sharpest of the three: an action that can
      never fire is a guard bug or dead code, and it silently removes an interleaving the author
      believed was covered. A run that never explored it still reports OK.
    - ``always_enabled_actions``: enabled in EVERY reachable state, i.e. the guard never guards.
      Often intentional — an ``Idle`` step, or an adversarial ``crash`` that may happen at any
      moment — so this is a smell to confirm, not a defect. Empty whenever the model has a terminal
      state, since nothing is enabled there.
    - ``stutter_only_actions``: enabled somewhere, but every firing lands back on the state it came
      from. The action fires and changes nothing.

    ``truncated`` carries over from the graph and changes what ``dead_actions`` means: under a cap,
    an action is only *not yet seen enabled*, which is a weaker claim than dead."""

    dead_actions: tuple[str, ...]
    always_enabled_actions: tuple[str, ...]
    stutter_only_actions: tuple[str, ...]
    actions_total: int
    truncated: bool

    @property
    def ok(self) -> bool:
        """True when nothing was flagged. Not a verdict on the model — only the author can say
        whether an always-enabled action is intended — but a useful gate in a test suite."""
        return not (self.dead_actions or self.always_enabled_actions or self.stutter_only_actions)

    def __str__(self) -> str:
        if self.truncated:
            head = "COVERAGE INCONCLUSIVE (truncated at max_states — 'dead' only means not yet seen enabled)"
        elif self.ok:
            head = f"COVERAGE — all {self.actions_total} actions fire and move"
        else:
            head = f"COVERAGE — of {self.actions_total} actions:"
        lines = [head]
        for label, names in (
            ("never enabled (dead)", self.dead_actions),
            ("always enabled (the guard never guards)", self.always_enabled_actions),
            ("only ever self-loops (fires, changes nothing)", self.stutter_only_actions),
        ):
            if names:
                lines.append(f"  {label}: {', '.join(names)}")
        return "\n".join(lines)

ok property

ok: bool

True when nothing was flagged. Not a verdict on the model — only the author can say whether an always-enabled action is intended — but a useful gate in a test suite.

metrics_of

metrics_of(graph: Graph[S]) -> StateSpaceMetrics

Compute metrics from an already-explored :class:~musil.core.Graph.

Source code in src/musil/metrics.py

def metrics_of[S: Hashable](graph: Graph[S]) -> StateSpaceMetrics:
    """Compute metrics from an already-explored :class:`~musil.core.Graph`."""
    states = graph.states
    n = len(states)
    successors = [
        {nxt for _, nxt in graph.edges.get(s, ()) if nxt != s} for s in states
    ]
    degrees = [len(succ) for succ in successors]
    return StateSpaceMetrics(
        reachable_states=n,
        transitions=sum(len(graph.edges.get(s, ())) for s in states),
        initial_states=len(graph.initial),
        terminal_states=sum(1 for d in degrees if d == 0),
        max_branching=max(degrees, default=0),
        mean_branching=(sum(degrees) / n if n else 0.0),
        truncated=graph.truncated,
    )

state_space_metrics

state_space_metrics(
    model: Model[S], *, max_states: int = 1000000
) -> StateSpaceMetrics

Explore model and report the size + branching of its reachable behaviour (see :class:StateSpaceMetrics).

Source code in src/musil/metrics.py

def state_space_metrics[S: Hashable](
    model: Model[S], *, max_states: int = 1_000_000
) -> StateSpaceMetrics:
    """Explore ``model`` and report the size + branching of its reachable behaviour (see
    :class:`StateSpaceMetrics`)."""
    return metrics_of(explore(model, max_states=max_states))

coverage_of

coverage_of(graph: Graph[S], model: Model[S]) -> Coverage

Compute coverage from an already-explored :class:~musil.core.Graph.

Reads the graph's recorded edges only — enabled is never called again, so this is O(edges) and, unlike re-applying the actions, it stays correct under symmetry reduction (the edges are already canonical). model is needed for one thing: the full list of action names, since an action that never fired leaves no trace in the graph to be counted.

Source code in src/musil/metrics.py

def coverage_of[S: Hashable](graph: Graph[S], model: Model[S]) -> Coverage:
    """Compute coverage from an already-explored :class:`~musil.core.Graph`.

    Reads the graph's recorded edges only — ``enabled`` is never called again, so this is O(edges)
    and, unlike re-applying the actions, it stays correct under symmetry reduction (the edges are
    already canonical). ``model`` is needed for one thing: the full list of action names, since an
    action that never fired leaves no trace in the graph to be counted."""
    names = [a.name for a in model.actions]
    fired: set[str] = set()
    moved: set[str] = set()
    enabled_in: dict[str, int] = dict.fromkeys(names, 0)
    for s in graph.states:
        seen_here: set[str] = set()
        for action, nxt in graph.edges.get(s, ()):
            fired.add(action)
            if nxt != s:
                moved.add(action)
            seen_here.add(action)
        for action in seen_here:
            enabled_in[action] = enabled_in.get(action, 0) + 1
    total_states = len(graph.states)
    return Coverage(
        dead_actions=tuple(n for n in names if n not in fired),
        always_enabled_actions=tuple(
            n for n in names if total_states and enabled_in.get(n, 0) == total_states
        ),
        stutter_only_actions=tuple(n for n in names if n in fired and n not in moved),
        actions_total=len(names),
        truncated=graph.truncated,
    )

coverage

coverage(
    model: Model[S], *, max_states: int = 1000000
) -> Coverage

Explore model and report which of its actions ever fired, always could, or never moved (see :class:Coverage).

Deliberately separate from check rather than a field on Result: check early-exits at the first violation, so the graph it saw is partial and a coverage answer computed from it would be wrong in the direction that matters — actions look dead because the sweep stopped, not because they cannot fire. Run this on the whole space, next to check, not inside it.

Source code in src/musil/metrics.py

def coverage[S: Hashable](model: Model[S], *, max_states: int = 1_000_000) -> Coverage:
    """Explore ``model`` and report which of its actions ever fired, always could, or never moved
    (see :class:`Coverage`).

    Deliberately separate from ``check`` rather than a field on ``Result``: ``check`` early-exits at
    the first violation, so the graph it saw is partial and a coverage answer computed from it would
    be wrong in the direction that matters — actions look dead because the sweep stopped, not because
    they cannot fire. Run this on the whole space, next to ``check``, not inside it."""
    return coverage_of(explore(model, max_states=max_states), model)

solution_cardinality

solution_cardinality(
    model: Model[S],
    predicate: Callable[[S], bool],
    *,
    max_states: int = 1000000,
) -> int

How many REACHABLE states satisfy predicate — the design's "expression cardinality" for that intent, counted exactly over the explored state space (not a guessed score). A large number for a goal that should be reached one way is a redundancy/ambiguity smell; for a recovery goal, more satisfying states can mean more resilience. Interpret in context.

Source code in src/musil/metrics.py

def solution_cardinality[S: Hashable](
    model: Model[S], predicate: Callable[[S], bool], *, max_states: int = 1_000_000
) -> int:
    """How many REACHABLE states satisfy ``predicate`` — the design's "expression cardinality" for
    that intent, counted exactly over the explored state space (not a guessed score). A large number
    for a goal that *should* be reached one way is a redundancy/ambiguity smell; for a recovery goal,
    more satisfying states can mean more resilience. Interpret in context."""
    return sum(1 for s in explore(model, max_states=max_states).states if predicate(s))

Graph export — to_dot, explore

The reachable state graph and its Graphviz DOT rendering.

musil.graph

Render a model's reachable state graph as Graphviz DOT — the same artifact FizzBee emits, but from the Python model. to_dot(model) explores the model and returns a DOT string you can pipe to dot -Tsvg. Pass highlight (a sequence of states, e.g. the states of a counterexample trace) to colour that path — handy for visualising why a check failed.

to_dot

to_dot(
    source: Model[S] | Graph[S],
    *,
    highlight: Iterable[S] = (),
    name: str = "musil",
    max_states: int = 1000000,
) -> str

Graphviz DOT for the reachable state graph. source is a Model (explored here) or a pre-built Graph. Initial states are drawn bold; highlight states are filled and edges between two highlighted states are red (e.g. highlight=[s.state for s in result.trace]).

Source code in src/musil/graph.py

def to_dot[S: Hashable](
    source: Model[S] | Graph[S],
    *,
    highlight: Iterable[S] = (),
    name: str = "musil",
    max_states: int = 1_000_000,
) -> str:
    """Graphviz DOT for the reachable state graph. ``source`` is a Model (explored here) or a
    pre-built Graph. Initial states are drawn bold; ``highlight`` states are filled and edges between
    two highlighted states are red (e.g. ``highlight=[s.state for s in result.trace]``)."""
    graph = source if isinstance(source, Graph) else explore(source, max_states=max_states)
    idx = {s: i for i, s in enumerate(graph.states)}
    hi = set(highlight)
    inits = set(graph.initial)

    lines = [
        f"digraph {name} {{",
        "  rankdir=LR;",
        '  node [shape=box, fontname="monospace"];',
        '  edge [fontname="monospace"];',
    ]
    for s in graph.states:
        attrs = [f'label="{_escape(_label(s))}"']
        if s in inits:
            attrs.append("penwidth=2")
        if s in hi:
            attrs += ["style=filled", 'fillcolor="#ffe0e0"']
        lines.append(f"  n{idx[s]} [{', '.join(attrs)}];")
    for s in graph.states:
        for action, t in graph.edges[s]:
            edge_attrs = [f'label="{_escape(action)}"']
            if s in hi and t in hi:
                edge_attrs.append('color="red"')
                edge_attrs.append("penwidth=2")
            lines.append(f"  n{idx[s]} -> n{idx[t]} [{', '.join(edge_attrs)}];")
    lines.append("}")
    return "\n".join(lines) + "\n"