Skip to content

Create, Compile, and Run a Phased Reactive System

Recap

By this point the video player has three layers of declarations:

  • node instances such as Network, QualityPolicy, and MediaSession;
  • phases such as measure, decide, drop_quality, and play;
  • transitions that say how one phase hands control to the next.

The running example: an adaptive-bitrate video player

The player samples network bandwidth, decides whether the current bitrate is sustainable, optionally lowers quality, and then plays the next chunk. The buffer and bitrate persist across ticks, so the next tick starts from the state left by the previous one.

flowchart LR
    init([init]) --> measure
    measure --> decide[decide]
    decide -->|healthy| play
    decide -->|stalling| drop_quality
    drop_quality --> play
    play --> done([⊥])

    classDef measure fill:#2f6fed22,stroke:#2f6fed;
    classDef decide fill:#7c3aed22,stroke:#7c3aed;
    classDef dropQuality fill:#d9770622,stroke:#d97706;
    classDef play fill:#15803d22,stroke:#15803d;

    class measure measure;
    class decide decide;
    class drop_quality dropQuality;
    class play play;
Phase Nodes What happens
measure Network Sample the current bandwidth from the system clock.
decide QualityPolicy Compare projected drain rate against the buffer; set stalling.
drop_quality BitrateController Drop the target bitrate by one rung.
play Decoder, MediaSession, Logger Compute downloaded seconds, integrate the buffer, log.

At node level, the same model looks like this. Solid arrows show that one node reads another node's state variable; dashed arrows from state show self-reads, where a node reads its own state variable from the previous tick. The node colors correspond to the phase colors in the table above:

flowchart LR
    network["Network"]
    policy["QualityPolicy"]
    controller["BitrateController"]
    decoder["Decoder"]
    session["MediaSession"]
    logger["Logger"]
    controller_state(("state"))
    session_state(("state"))
    logger_state(("state"))

    network --> policy
    network --> decoder
    network --> logger
    controller --> policy
    controller --> decoder
    controller --> logger
    decoder --> session
    session --> logger
    session --> policy
    policy --> logger
    controller_state -.-> controller
    session_state -.-> session
    logger_state -.-> logger

    classDef measure fill:#2f6fed22,stroke:#2f6fed;
    classDef decide fill:#7c3aed22,stroke:#7c3aed;
    classDef dropQuality fill:#d9770622,stroke:#d97706;
    classDef play fill:#15803d22,stroke:#15803d;
    classDef state fill:#94a3b822,stroke:#94a3b8,stroke-dasharray:3 3;

    class network measure;
    class policy decide;
    class controller dropQuality;
    class decoder,session,logger play;
    class controller_state,session_state,logger_state state;
Full code listing: examples/video_player/video_player.py 
"""Adaptive-bitrate video player as a feedback control loop.

Each tick the player asks one question: do I have enough buffered video to
keep playing at the current quality?  If yes, the short branch just plays the
next chunk.  If no, the long branch first lowers the target bitrate, then
plays.  Buffer and bitrate persist across ticks, so the branching pattern is
driven by the closed loop between the network, the buffer, and the policy.

Phase graph::

    measure (init) -> decide -+-[healthy]--> play -> bottom
                              |
                              +-[stalling]-> drop_quality -> play -> bottom
"""

from __future__ import annotations

import regelum as rg

TICK_DT_SECONDS = 1.0
BITRATE_LADDER_KBPS = (240, 480, 720, 1080, 2160)
TOP_BITRATE_KBPS = BITRATE_LADDER_KBPS[-1]
STALL_HORIZON_SECONDS = 4.0


class Network(rg.Node):
    """Stochastic-looking but deterministic bandwidth model.

    Drops to a slow link in the middle of the run so the policy has to react,
    then recovers so the buffer can refill.  The schedule is fixed to keep the
    example reproducible.
    """

    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)

    class State(rg.NodeState):
        bandwidth_kbps: float = rg.var(init=float(TOP_BITRATE_KBPS))

    def update(self, inputs: Inputs) -> State:
        if inputs.tick < 6:
            value = 2400.0
        elif inputs.tick < 14:
            value = 600.0
        elif inputs.tick < 22:
            value = 1100.0
        else:
            value = 2400.0
        return self.State(bandwidth_kbps=value)


class QualityPolicy(rg.Node):
    """Decides whether the player is about to stall.

    The estimated drain rate is ``1 - bandwidth / bitrate`` seconds of video
    lost per wall-second.  If buffered seconds will not survive
    ``STALL_HORIZON_SECONDS`` at that drain rate, mark the tick as stalling.
    """

    class Inputs(rg.NodeInputs):
        buffer_seconds: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)

    class State(rg.NodeState):
        stalling: bool = rg.var(init=False)

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        drain = max(0.0, 1.0 - inputs.bandwidth_kbps / bitrate)
        if drain <= 0.0:
            return self.State(stalling=False)
        time_to_empty = inputs.buffer_seconds / drain
        return self.State(stalling=time_to_empty < STALL_HORIZON_SECONDS)


class BitrateController(rg.Node):
    """Owns the current target bitrate.  Drops one rung when invoked."""

    class Inputs(rg.NodeInputs):
        current: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        value: int = rg.var(init=TOP_BITRATE_KBPS)

    def update(self, inputs: Inputs) -> State:
        try:
            index = BITRATE_LADDER_KBPS.index(inputs.current)
        except ValueError:
            index = len(BITRATE_LADDER_KBPS) - 1
        next_index = max(0, index - 1)
        return self.State(value=BITRATE_LADDER_KBPS[next_index])


class Decoder(rg.Node):
    """Models how many seconds of video can be downloaded in one tick.

    With ``bandwidth_kbps`` of throughput and a video encoded at
    ``bitrate_kbps``, one wall-second of downloading produces
    ``bandwidth / bitrate`` seconds of playable content.
    """

    class Inputs(rg.NodeInputs):
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)

    class State(rg.NodeState):
        fetched_seconds: float

    def update(self, inputs: Inputs) -> State:
        bitrate = max(inputs.bitrate_kbps, 1)
        return self.State(fetched_seconds=inputs.bandwidth_kbps / bitrate * TICK_DT_SECONDS)


class MediaSession(rg.Node):
    """The plant.  Buffer fills with newly fetched video, drains with playback."""

    class Inputs(rg.NodeInputs):
        previous: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        fetched: float = rg.src(Decoder.State.fetched_seconds)

    class State(rg.NodeState):
        buffer_seconds: float = rg.var(init=10.0)

    def update(self, inputs: Inputs) -> State:
        next_buffer = inputs.previous + inputs.fetched - TICK_DT_SECONDS
        return self.State(buffer_seconds=max(0.0, next_buffer))


class Logger(rg.Node):
    """Appends a per-tick record so the trajectory is visible after the run."""

    Sample = tuple[int, float, int, float, bool]

    class Inputs(rg.NodeInputs):
        tick: int = rg.src(rg.Clock.tick)
        bandwidth_kbps: float = rg.src(Network.State.bandwidth_kbps)
        bitrate_kbps: int = rg.src(lambda: BitrateController.State.value)
        buffer_seconds: float = rg.src(lambda: MediaSession.State.buffer_seconds)
        stalling: bool = rg.src(QualityPolicy.State.stalling)
        history: list["Logger.Sample"] = rg.src(lambda: Logger.State.history)

    class State(rg.NodeState):
        history: list["Logger.Sample"] = rg.var(init=lambda: [])

    def update(self, inputs: Inputs) -> State:
        record: Logger.Sample = (
            inputs.tick,
            inputs.bandwidth_kbps,
            inputs.bitrate_kbps,
            inputs.buffer_seconds,
            inputs.stalling,
        )
        inputs.history.append(record)
        return self.State(history=inputs.history)


def build_system() -> rg.PhasedReactiveSystem:
    network = Network()
    policy = QualityPolicy()
    controller = BitrateController()
    decoder = Decoder()
    session = MediaSession()
    logger = Logger()

    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "measure",
                nodes=(network,),
                transitions=(rg.Goto("decide"),),
                is_initial=True,
            ),
            rg.Phase(
                "decide",
                nodes=(policy,),
                transitions=(
                    rg.If(rg.V(policy.State.stalling), "drop_quality", name="stalling"),
                    rg.Else("play", name="healthy"),
                ),
            ),
            rg.Phase(
                "drop_quality",
                nodes=(controller,),
                transitions=(rg.Goto("play"),),
            ),
            rg.Phase(
                "play",
                nodes=(decoder, session, logger),
                transitions=(rg.Goto(rg.terminate),),
            ),
        ],
    )


def main() -> None:
    system = build_system()
    print(f"compile_ok = {system.compile_report.ok}")
    print(
        "phase schedules: "
        + " | ".join(
            f"{name}={schedule}" for name, schedule in system.compile_report.phase_schedules.items()
        )
    )
    print()
    print("tick | bw(kbps) | bitrate | buffer(s) | stall? | path")
    print("-----+----------+---------+-----------+--------+----------------------")
    for _ in range(30):
        records = system.step()
        path_phases: list[str] = []
        for record in records:
            if not path_phases or path_phases[-1] != record.phase:
                path_phases.append(record.phase)
        path = " -> ".join(path_phases)
        snapshot = system.snapshot()
        print(
            f"{system.read(rg.Clock.tick):4d} | "
            f"{snapshot['Network.bandwidth_kbps']:8.0f} | "
            f"{snapshot['BitrateController.value']:7d} | "
            f"{snapshot['MediaSession.buffer_seconds']:9.2f} | "
            f"{str(snapshot['QualityPolicy.stalling']):>6} | "
            f"{path}"
        )


if __name__ == "__main__":
    main()

What Regelum Checks

The PRS declaration is intentionally flexible: a model author can create overlapping conditional transitions, omit a transition case, place mutually dependent nodes in the same phase, or create a phase graph whose control flow never reaches terminate. Regelum therefore compiles and checks the declarative PRS structure before the first runtime step.

The formal conditions are:

Condition Meaning Runtime problem it prevents
C1 Every phase-local node dependency graph is acyclic. Nodes in one phase cannot require each other's current output in a circular order.
C2 The phase-transition graph is acyclic. A tick cannot loop forever by following phase edges.
C3 For every phase and state, exactly one outgoing transition is enabled. A phase cannot have either no next step or multiple possible next steps.
C2* A semantic cycle check: a phase cycle is allowed only when the guards cannot be satisfied for enough consecutive traversals. Dead or self-blocking cycles can be accepted while live infinite loops are rejected.

C1, C2, and C3 are syntactic sufficient conditions. They are intentionally conservative: if all three hold, the compiled system has an executable node order, a terminating tick graph, and deterministic branch selection. C2* is a less conservative termination check for cyclic phase graphs. Instead of rejecting every phase cycle, it asks whether a cycle can actually be followed repeatedly under the symbolic guards that lead around the cycle. The finite bound is computed from R_C, the guard-relevant cycle-state set: state variables that are both read by cycle guards and written by phases in the same cycle.

Current Regelum checks these items concretely:

  • Node and phase consistency. Every node state path must be unique; every node name must be unique; every input source must be connected, known, and unambiguous; every instance-bound input or guard source must belong to a node assigned to a phase; every phase target must resolve to another declared phase or to terminate.
  • Transition structure. Goto cannot be mixed with If / ElseIf / Else chains. ElseIf and Else must follow an open If chain, and a chain cannot continue after Else.
  • C1 phase schedulability. For each phase, Regelum builds edges from each active node to active nodes that read its state in the same phase. The phase is accepted only if those edges admit a topological order. The resolved order is stored in compile_report.phase_schedules.
  • C3 guarded-transition determinism. Regelum converts If / ElseIf / Else chains into effective predicates. For symbolic predicates built with rg.V(...), it uses Z3 to prove that no two effective predicates overlap and that their disjunction covers the relevant state space. For callable guards, it samples finite state domains when possible. A failure reports either an overlapping transition pair or a state where no transition is enabled.
  • C2 and C2* tick termination. If the phase graph has no cycles, it satisfies C2 directly. If the graph has cycles, Regelum extracts each simple cycle and tries C2*: every edge around that cycle must have symbolic guards, and all variables in R_C must have finite domains. The checker asks Z3 whether the cycle can be traversed for |D^{R_C}| consecutive traversals. If such a witness exists, the cycle is rejected; if the formula is unsatisfiable, the cycle is accepted as dead or self-blocking.
  • Continuous-phase contract. When a system has a continuous phase, Regelum also checks that a tick reaches exactly one continuous phase on every feasible path. This avoids silently choosing semantics for repeated continuous integration inside one logical tick.

Build The Runtime System

At this point the declarations are still just Python objects: node instances, phase objects, and transition objects. They describe the intended feedback loop, but nothing has been scheduled or executed yet.

rg.PhasedReactiveSystem is the boundary between declaration and runtime. When you construct it, Regelum compiles the phase list into a concrete runtime model: it resolves references, builds node schedules, checks transition targets, computes initial-state requirements, and stores the result in compile_report.

import regelum as rg


def build_system() -> rg.PhasedReactiveSystem:
    network = Network()
    policy = QualityPolicy()
    controller = BitrateController()
    decoder = Decoder()
    session = MediaSession()
    logger = Logger()

    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "measure",
                nodes=(network,),
                transitions=(rg.Goto("decide"),),
                is_initial=True,
            ),
            rg.Phase(
                "decide",
                nodes=(policy,),
                transitions=(
                    rg.If(
                        rg.V(policy.State.stalling),
                        "drop_quality",
                        name="stalling",
                    ),
                    rg.Else("play", name="healthy"),
                ),
            ),
            rg.Phase(
                "drop_quality",
                nodes=(controller,),
                transitions=(rg.Goto("play"),),
            ),
            rg.Phase(
                "play",
                nodes=(decoder, session, logger),
                transitions=(rg.Goto(rg.terminate),),
            ),
        ],
    )

rg.PhasedReactiveSystem is the object you keep and run. It owns the compiled phase graph, the current runtime state, and the compile report.

Constructor Behavior

The constructor receives the phases that define the system.

system = rg.PhasedReactiveSystem(phases=phases)

During construction, regelum compiles the system. If compilation succeeds, the returned object is ready for step(), update(), snapshot(), read(...), and reset(...). If compilation fails, the default behavior is to raise CompileError.

Use strict=False when you want to inspect a broken system instead of raising immediately:

system = rg.PhasedReactiveSystem(
    phases=phases,
    strict=False,
)

print(system.compile_report.issues)

What compilation resolves

Compilation resolves:

  • node names;
  • input sources;
  • instance connections;
  • state paths;
  • phase targets;
  • guard references;
  • phase schedules;
  • dependency edges;
  • required initial state variables.

For the video player, the report's phase_schedules shows the topologically ordered nodes per phase. minimal_initial_state_vars lists the state variables that must exist before their first read: BitrateController.value, Logger.history, and MediaSession.buffer_seconds.

Compile report

Every rg.PhasedReactiveSystem stores a compile_report. Read it before debugging runtime behavior; it tells you what the constructor resolved and what it rejected.

system = build_system()
report = system.compile_report

print(report.ok)
print(report.issues)
print(report.warnings)
print(report.phase_schedules)
print(report.minimal_initial_state_vars)
print(report.required_initial_state_vars)

For the video player this prints:

True
()
()
{'measure': ('Network',), 'decide': ('QualityPolicy',), 'drop_quality': ('BitrateController',), 'play': ('Decoder', 'MediaSession', 'Logger')}
('BitrateController.value', 'Logger.history', 'MediaSession.buffer_seconds')
{'BitrateController.value': ('BitrateController.current', 'Decoder.bitrate_kbps', 'Logger.bitrate_kbps', 'QualityPolicy.bitrate_kbps'), 'Logger.history': ('Logger.history',), 'MediaSession.buffer_seconds': ('MediaSession.previous', 'QualityPolicy.buffer_seconds')}

Read this output as follows:

  • ok=True means compilation produced no errors.
  • issues=() means nothing was rejected. If a C1/C2/C3 check fails, this tuple contains CompileIssue objects naming the phase and condition.
  • warnings=() means there are no non-fatal structural warnings.
  • phase_schedules is the per-phase topological order. The play phase runs Decoder first, then MediaSession, then Logger, because the session reads Decoder.State.fetched_seconds and the logger reads the session state.
  • minimal_initial_state_vars is the smallest set of state variables that must have a tick-zero value for this compiled graph.
  • required_initial_state_vars maps each such state variable to the inputs that may read it before it is written in the current tick. The map is not an error by itself: in this example all three variables have rg.var(init=...), so the constructor can initialize them automatically.

Use format() for the same report plus dependency edges:

print(system.compile_report.format())

ok = True
issues = ()
warnings = ()
minimal_initial_state_vars = ('BitrateController.value', 'Logger.history', 'MediaSession.buffer_seconds')
state_vars_without_initial = ('Decoder.fetched_seconds',)
required_initial_state_vars = {'BitrateController.value': ('BitrateController.current', 'Decoder.bitrate_kbps', 'Logger.bitrate_kbps', 'QualityPolicy.bitrate_kbps'), 'Logger.history': ('Logger.history',), 'MediaSession.buffer_seconds': ('MediaSession.previous', 'QualityPolicy.buffer_seconds')}
phase_schedules = {'measure': ('Network',), 'decide': ('QualityPolicy',), 'drop_quality': ('BitrateController',), 'play': ('Decoder', 'MediaSession', 'Logger')}
phase_dependency_edges = {'measure': (), 'decide': (), 'drop_quality': (), 'play': (('Decoder', 'MediaSession'), ('MediaSession', 'Logger'))}

state_vars_without_initial can contain variables that are still safe. Here Decoder.fetched_seconds has no init, but it is written in play before MediaSession reads it in the same phase. Therefore it is not listed in required_initial_state_vars and does not produce an issue.

minimal_initial_state_vars is the smallest set of state variables that must have a tick-zero value for this compiled graph. Those values may come from rg.var(init=...), from a callable initializer, or from a runtime initial_state override.

required_initial_state_vars tells you which state variables are read before they are guaranteed to be written by the compiled phase schedule. If any of those variables has no declared init and no runtime override, Regelum reports an issue. You can use that list to build an initial_state mapping:

missing = system.compile_report.required_initial_state_vars
print(missing)

system.reset(
    initial_state={
        MediaSession.State.buffer_seconds: 5.0,
        BitrateController.State.value: 720,
    }
)

In the video player, all required tick-zero state variables have either a static initial value or a callable initializer.

Common compile issues

Typical issues include:

  • input source is not connected;
  • input source is unknown;
  • class-level reference is ambiguous;
  • state path is duplicated;
  • explicit node names are duplicated;
  • state variable without initial value is read too early;
  • phase graph is incomplete;
  • transition target is unknown;
  • transition chain is malformed.

C1, C2, C3, and C2* Examples

Compilation rejects cyclic dependency graphs inside a phase. This is the C1 check. For the video player, the only non-trivial intra-phase dependency chain is Decoder → MediaSession → Logger in play, which is acyclic.

For finite state variable domains, compilation also checks C3 by requiring exactly one enabled transition per sampled state. The branching in decide is If(V(QualityPolicy.State.stalling), "drop_quality") plus Else("play"), with stalling: bool — boolean has a finite domain, so C3 is verified statically.

The video-player phase graph is acyclic, so it satisfies C2 directly and does not need a C2* cycle proof.

You can see all three facts in the compile report:

from examples.video_player.video_player import build_system


system = build_system()
print(system.compile_report.format())

ok = True
issues = ()
warnings = ()
minimal_initial_state_vars = ('BitrateController.value', 'Logger.history', 'MediaSession.buffer_seconds')
state_vars_without_initial = ('Decoder.fetched_seconds',)
required_initial_state_vars = {'BitrateController.value': ('BitrateController.current', 'Decoder.bitrate_kbps', 'Logger.bitrate_kbps', 'QualityPolicy.bitrate_kbps'), 'Logger.history': ('Logger.history',), 'MediaSession.buffer_seconds': ('MediaSession.previous', 'QualityPolicy.buffer_seconds')}
phase_schedules = {'measure': ('Network',), 'decide': ('QualityPolicy',), 'drop_quality': ('BitrateController',), 'play': ('Decoder', 'MediaSession', 'Logger')}
phase_dependency_edges = {'measure': (), 'decide': (), 'drop_quality': (), 'play': (('Decoder', 'MediaSession'), ('MediaSession', 'Logger'))}

For C1, inspect phase_dependency_edges: only play has intra-phase edges, and they form a chain, not a cycle. For C3, issues=() means the symbolic branch from decide has neither overlap nor a missing case. For C2, the phase graph has no edge returning to an earlier phase: measure -> decide -> drop_quality -> play -> terminate and measure -> decide -> play -> terminate are both finite paths.

C1 Violation: One Phase Has A Node Cycle

C1 is local to a single phase. The phase is invalid if the active nodes cannot be topologically ordered. This example puts two mutually dependent nodes in one phase:

flowchart LR
    subgraph coupled["phase: coupled"]
        first["First<br/>reads Second.b<br/>writes First.a"]
        second["Second<br/>reads First.a<br/>writes Second.b"]
        first -->|"First.a"| second
        second -->|"Second.b"| first
    end
    coupled --> done([⊥])

    classDef bad fill:#dc262622,stroke:#dc2626;
    class first,second bad;

The corresponding script is examples/compile_checks/c1_violation.py. It prints:

Full code listing: examples/compile_checks/c1_violation.py 
from __future__ import annotations

import regelum as rg


class First(rg.Node):
    class Inputs(rg.NodeInputs):
        b: float = rg.src("Second.State.b")

    class State(rg.NodeState):
        a: float = rg.var(init=0.0)

    def update(self, inputs: Inputs) -> State:
        return self.State(a=inputs.b + 1.0)


class Second(rg.Node):
    class Inputs(rg.NodeInputs):
        a: float = rg.src(First.State.a)

    class State(rg.NodeState):
        b: float = rg.var(init=0.0)

    def update(self, inputs: Inputs) -> State:
        return self.State(b=inputs.a + 1.0)


def build_system() -> rg.PhasedReactiveSystem:
    first = First()
    second = Second()
    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "coupled",
                nodes=(first, second),
                transitions=(rg.Goto(rg.terminate),),
                is_initial=True,
            )
        ],
    )


def main() -> None:
    try:
        build_system()
    except rg.CompileError as exc:
        for issue in exc.report.issues:
            print(f"{issue.location}: {issue.message}")
        return
    raise RuntimeError("Expected C1 violation, but system compiled.")


if __name__ == "__main__":
    main()

coupled: C1 violation: phase dependency graph is cyclic (First->Second, Second->First)

The fix is to split the instantaneous cycle across phases, or to make one edge read a state value from a previous tick instead of another node's current output in the same phase.

C2 Violated, But C2* Passes: A Dead Phase Cycle

C2 is syntactic: it rejects every cycle in the phase-transition graph. That is safe, but conservative. C2* is more precise: it checks whether the cycle can actually be traversed repeatedly under the symbolic guards.

The split-write example has a cycle in the phase graph:

flowchart LR
    phi0["phi0<br/>writes X.x"]
    phi1["phi1<br/>writes Y.y"]
    done([⊥])

    phi0 -->|"~X.x & ~Y.y"| phi1
    phi0 -->|"else"| done
    phi1 -->|"~X.x & Y.y"| phi0
    phi1 -->|"else"| done

    classDef phase fill:#4f46e522,stroke:#4f46e5;
    classDef done fill:#64748b22,stroke:#64748b;
    class phi0,phi1 phase;
    class done done;

The first traversal can go phi0 -> phi1 -> phi0, but after phi1 writes Y.y, the guard from phi0 back into the cycle requires ~Y.y. The cycle blocks itself. Therefore C2 fails syntactically, but C2* accepts the system:

Full code listing: examples/compile_checks/c2star_split_writes.py 
from __future__ import annotations

# ruff: noqa: I001
import regelum as rg
import random


class X(rg.Node):
    class State(rg.NodeState):
        x: bool = rg.var(init=False)

    def update(self, x: bool = rg.src(lambda: X.State.x)) -> State:
        return self.State(x=x & bool(random.getrandbits(1)))


class Y(rg.Node):
    class State(rg.NodeState):
        y: bool = rg.var(init=False)

    def update(self, y: bool = rg.src(lambda: Y.State.y)) -> State:
        return self.State(y=y & bool(random.getrandbits(1)))


def build_system(
    p_x: float = 0.5,
    p_y: float = 0.5,
    seed: int | None = 0,
) -> rg.PhasedReactiveSystem:
    del p_x, p_y, seed
    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "phi0",
                nodes=(X(),),
                transitions=(
                    rg.If(~rg.V(X.State.x) & ~rg.V(Y.State.y), "phi1"),
                    rg.Else(rg.terminate),
                ),
                is_initial=True,
            ),
            rg.Phase(
                "phi1",
                nodes=(Y(),),
                transitions=(
                    rg.If(~rg.V(X.State.x) & rg.V(Y.State.y), "phi0"),
                    rg.Else(rg.terminate),
                ),
            ),
        ],
    )


def main() -> None:
    system = build_system()
    print(f"compile ok = {system.compile_report.ok}")
    print("C2*(2) status = pass")
    print("cycle phi0 -> phi1 -> phi0 is dead after one traversal")


if __name__ == "__main__":
    main()

compile ok = True
C2*(2) status = pass
cycle phi0 -> phi1 -> phi0 is dead after one traversal

This is the kind of cyclic phase graph C2* is meant to keep: the graph contains a loop, but no infinite tick execution can keep following it.

C2* Violation: A Feasible Infinite Loop

Now compare a phase cycle whose guards can remain true forever:

flowchart LR
    a["a<br/>writes Mode.flag"]
    b["b<br/>writes Mode.flag"]
    done([⊥])

    a -->|"Mode.flag"| b
    a -->|"~Mode.flag"| done
    b -->|"Mode.flag"| a
    b -->|"~Mode.flag"| done

    classDef bad fill:#dc262622,stroke:#dc2626;
    classDef done fill:#64748b22,stroke:#64748b;
    class a,b bad;
    class done done;

Both phases write the same guard-relevant state variable, Mode.flag. Under the compile-time abstraction, an update method may produce any type-correct value for its state variable. Z3 can therefore find a witness in which Mode.flag remains true on every traversal. Regelum rejects the cycle:

Full code listing: examples/compile_checks/c2star_cycle.py 
from __future__ import annotations

import regelum as rg


class Mode(rg.Node):
    class State(rg.NodeState):
        flag: bool = rg.var(init=False)


def build_dead_cycle_system() -> rg.PhasedReactiveSystem:
    mode = Mode()
    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "a",
                nodes=(mode,),
                transitions=(
                    rg.If(rg.V(Mode.State.flag), "b", name="to-b"),
                    rg.If(~rg.V(Mode.State.flag), rg.terminate, name="stop-a"),
                ),
                is_initial=True,
            ),
            rg.Phase(
                "b",
                nodes=(mode,),
                transitions=(
                    rg.If(~rg.V(Mode.State.flag), "a", name="to-a"),
                    rg.If(rg.V(Mode.State.flag), rg.terminate, name="stop-b"),
                ),
            ),
        ],
    )


def build_live_cycle_system() -> rg.PhasedReactiveSystem:
    mode = Mode()
    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "a",
                nodes=(mode,),
                transitions=(
                    rg.If(rg.V(Mode.State.flag), "b", name="to-b"),
                    rg.If(~rg.V(Mode.State.flag), rg.terminate, name="stop-a"),
                ),
                is_initial=True,
            ),
            rg.Phase(
                "b",
                nodes=(mode,),
                transitions=(
                    rg.If(rg.V(Mode.State.flag), "a", name="to-a"),
                    rg.If(~rg.V(Mode.State.flag), rg.terminate, name="stop-b"),
                ),
            ),
        ],
    )


def main() -> None:
    try:
        dead = build_dead_cycle_system()
    except rg.CompileError as exc:
        print("dead cycle: compile=False")
        for issue in exc.report.issues:
            print(f"{issue.location}: {issue.message}")
    else:
        print(f"dead cycle: compile={dead.compile_report.ok}")

    try:
        build_live_cycle_system()
    except rg.CompileError as exc:
        print("live cycle: compile=False")
        for issue in exc.report.issues:
            print(f"{issue.location}: {issue.message}")
        return
    print("live cycle: compile=True")


if __name__ == "__main__":
    main()

a -> b -> a: C2*(2) violation: cycle is feasible for 2 traversal(s), R_C=['Mode.flag'], witness={...}

The important part is R_C=['Mode.flag']: the cycle controls the same state variable that its continuation guards read. Since the two-phase cycle can keep choosing values that satisfy the cycle guards, it is a possible infinite loop.

Plain C2 Violation: An Unconditional Cycle

An unconditional phase cycle is the simplest infinite-loop shape:

flowchart LR
    a["a"]
    b["b"]
    a -->|"Goto b"| b
    b -->|"Goto a"| a

    classDef bad fill:#dc262622,stroke:#dc2626;
    class a,b bad;

There is no guard that can block re-entry. C2 fails because the phase graph is cyclic, and C2* also fails immediately:

Full code listing: examples/compile_checks/unconditional_cycle.py 
from __future__ import annotations

import regelum as rg


def build_system() -> rg.PhasedReactiveSystem:
    return rg.PhasedReactiveSystem(
        phases=[
            rg.Phase(
                "a",
                nodes=(),
                transitions=(rg.Goto("b"),),
                is_initial=True,
            ),
            rg.Phase(
                "b",
                nodes=(),
                transitions=(rg.Goto("a"),),
            ),
        ],
    )


def main() -> None:
    try:
        build_system()
    except rg.CompileError as exc:
        for issue in exc.report.issues:
            print(f"{issue.location}: {issue.message}")


if __name__ == "__main__":
    main()

a -> b -> a: C2*(1) violation: cycle is feasible for 1 traversal(s), R_C=[], witness={}

R_C=[] means no cycle-owned state variable is needed to keep the loop alive; the unconditional edges alone are enough.

Runtime

After construction, the same rg.PhasedReactiveSystem object is the runtime handle. Runtime executes the compiled phase schedules and updates system state. It does not reinterpret declarations on every step.

The tick

A tick walks the phase graph from the initial phase until a transition reaches terminate. For the video player, a healthy tick visits measure → decide → play; a stalling tick visits measure → decide → drop_quality → play.

The feedback loop closes between ticks: MediaSession.buffer_seconds written in play of tick N is read by QualityPolicy in decide of tick N+1, and that read is what selects the branch.

Every system also has a built-in rg.Clock. Clock.tick is incremented after the whole tick terminates. In a discrete-only system, Clock.time advances with the tick by base_dt. When a tick contains a continuous phase, Clock.time advances immediately after that continuous phase, so later phases in the same tick can observe the new physical time. See Continuous dynamics for the ODE resolution rules.

Step order

One step() starts at the initial phase and follows transitions until the tick terminates.

For each phase:

  1. run active nodes in the compiled schedule;
  2. build each node input namespace;
  3. call update;
  4. normalize returned state variables;
  5. write state variables into state;
  6. choose the next phase from transitions.
records = system.step()

Each record contains the phase, node, inputs, and state variables. A 30-tick run of the player produces records like:

for record in records:
    print(record.phase, record.node, record.state)
# measure Network {'bandwidth_kbps': 600.0}
# decide  QualityPolicy {'stalling': False}
# play    Decoder {'fetched_seconds': 0.278}
# play    MediaSession {'buffer_seconds': 9.11}
# play    Logger {'history': [...]}

Running multiple ticks

Use update(steps=...) to execute several ticks.

system.run(steps=30)

Each tick starts from the initial phase again. State persists across ticks unless reset() is called.

State access

Use snapshot() to inspect current state. It returns user node state variables and committed ODE state values; system clock fields are read explicitly.

snapshot = system.snapshot()
print(snapshot["MediaSession.buffer_seconds"])
print(snapshot["BitrateController.value"])

Use read(...) when code has a state reference.

buffer = system.read(session.State.buffer_seconds)
tick = system.read(rg.Clock.tick)
time = system.read(rg.Clock.time)

Reset

reset() clears runtime state and history. It then applies declared initial values and optional overrides.

system.reset()
system.reset(initial_state={MediaSession.State.buffer_seconds: 5.0})

Logging nodes

A logger is just another node. It sees the values available at the point where its scheduled phase runs. The video player puts Logger last in play so it observes the buffer update from MediaSession and the freshly committed bitrate.

Rules

  • Read the compile report before debugging runtime behavior.
  • Create systems with rg.PhasedReactiveSystem(phases=[...]).
  • Use strict=False for diagnostics.
  • Resolve ambiguous class references with instance connections.
  • Add initial values only for state variables that must exist before execution.
  • Runtime follows compiled phase schedules.
  • State persists between ticks.
  • reset() clears state and history.
  • step() returns execution records.
  • snapshot() returns user-visible state; use read(rg.Clock.time) for clock fields.