Managing State

State is plain instance attributes. Handlers write them, run() reads them. The whole subject is when that happens, and the answer comes from one fact: your model runs on a single event loop.

Write in the handler, read in the loop

Every parameter a client can change has the same three parts:

class MyModel(ReactorModel):
    def load(self, config_path: Path | None) -> None:
        self.pipe = load_checkpoint()
        self.prompt = "a sunny meadow"  # 1. initialize

    @event(name="set_prompt", description="Scene the model renders")
    async def set_prompt(self, prompt: str = InputField(default="a sunny meadow")) -> None:
        self.prompt = prompt  # 2. write

    async def run(self) -> None:
        while True:
            await self.connected.wait()
            while self.connected.is_set():
                frame = self.pipe.forward(prompt=self.prompt)  # 3. read
                await self.emit(MyOutput(main_video=frame))

load() sets a starting value, so the first frame renders before any client sends a command. The handler stores the new value on self, and run() reads the attribute again on every pass, so the change applies to the next frame.

One loop, one thing at a time

run(), your command handlers, and your lifecycle hooks are three tasks on the same event loop. The runtime starts them together after load(), and exactly one of them holds the loop at any moment. A command that arrives while run() is working waits on a queue. It is dispatched at the next point run() awaits:

while self.connected.is_set():
    frame = self.pipe.forward(prompt=self.prompt)  # loop held, commands wait
    await self.emit(MyOutput(main_video=frame))    # loop yielded, handlers run here

Two things follow, and together they are why models on this runtime need no locks. A blocking call is safe. A synchronous forward pass holds the loop for its whole duration, so no handler can land in the middle of it. self.prompt is the same value on the first line of the pass as on the last, and a half-applied command is not a state your model can reach. emit() is your yield point. It always awaits, so a loop that emits every pass always picks up whatever queued up during the pass. While emit() waits for downstream room, the loop is free and handlers dispatch there too, which is why backpressure slows your generation without freezing your commands.

A loop that never awaits never picks up commands. If a code path can run for a while without emitting — a warm-up, a retry, a long wait on something external — put await asyncio.sleep(0) in it so handlers get their turn.

Snapshot across awaits

The guarantee holds for one uninterrupted stretch, so a pass with several awaits in it can read two different values of the same attribute. Copy what the pass depends on into locals first:

while self.connected.is_set():
    prompt = self.prompt  # fixed for this frame
    brightness = self.brightness

    latents = await self.pipe.encode(prompt)
    frame = await self.pipe.decode(latents)
    frame = (frame * brightness).clip(0, 255).astype("uint8")

    await self.emit(MyOutput(main_video=frame))

Without the snapshot, encode() can use one prompt while the brightness step uses a value from a command that arrived after it, and one frame mixes two states. The same applies when you move a slow pass onto a worker thread with asyncio.to_thread — that frees the loop, so handlers run during the pass.

Keep a handler shorter than a frame

A handler can do real work, and when that work depends only on the new value it belongs there: the handler runs once per change, while run() runs continuously.

@event(name="set_prompt", description="Scene the model renders")
async def set_prompt(self, prompt: str = InputField(default="")) -> None:
    self.prompt = prompt
    self._embedding = self.encoder.encode(prompt)

run() then reads self._embedding and never encodes again. The underscore marks a value the model derives for itself rather than one a client sets. The budget is the same one run() spends: a handler holds the loop while it runs, so a slow one delays the next frame. When the work takes longer than a frame, store the input and let run() pick it up.

State that resets

load() runs once for the life of the process, and one process serves session after session. State that belongs to a session — a step counter, a conversation, a cache of what this audience has seen — is set up in @session_started and released in @session_ended:

@session_started
async def on_session_start(self) -> None:
    self.prompt = "a sunny meadow"
    self._history = []

That is also the reliable place for it. A session can end with clients still attached, and the runtime tears those connections down without firing @disconnected, so cleanup hung off the per-client hook is skipped. See Sessions & Clients.

Media Input

Read the client’s camera and microphone.

The Run Loop

Emitting frames, batches, and frame rates.

Get Started

Build a Model

Media & Transport

Releases

Write in the handler, read in the loop

One loop, one thing at a time

Snapshot across awaits

Keep a handler shorter than a frame

State that resets

Next

Media Input

The Run Loop

​Write in the handler, read in the loop

​One loop, one thing at a time

​Snapshot across awaits

​Keep a handler shorter than a frame

​State that resets

​Next

Media Input

The Run Loop

Write in the handler, read in the loop

One loop, one thing at a time

Snapshot across awaits

Keep a handler shorter than a frame

State that resets

Next