The Run Loop

run() is where your model’s logic lives. The runtime calls it once after load() and it runs for the lifetime of the model. What happens inside is entirely yours.

Basic pattern

async def run(self) -> None:
    while True:
        # ✅ Park here while nobody is watching
        await self.connected.wait()
        while self.connected.is_set():
            frame = self.pipe.forward(prompt=self.prompt)
            # ✅ Hand the frame to the transport
            await self.emit(MyOutput(main_video=frame))

Each iteration of the inner loop runs your forward pass, emits the result, and re-checks whether anyone is still connected. The two-loop shape matters: the outer loop keeps the model alive across clients, while the inner one runs only while someone is watching. Without it, an idle model would spin a GPU generating frames nobody receives. run() is required. A model whose whole job lives in @event handlers still needs one, and parks instead of generating:

async def run(self) -> None:
    await asyncio.Event().wait()

The connection signal

self.connected is an asyncio.Event the runtime owns. It is set while at least one client is connected and cleared once the last one leaves.

await self.connected.wait()      # block until someone is here
self.connected.is_set()          # is anyone still here?

You never set or clear it yourself. It is set before the first @connected handler runs and cleared before the @disconnected handler for the last client to leave. For a model serving one client at a time, that is simply the moment they arrive and the moment they go.

Emitting frames

emit() takes an instance of your Output class, with one payload per declared track:

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

Every track on the class must be supplied. A video payload is an RGB uint8 array shaped (H, W, 3); an audio payload is int16 shaped (1, samples). emit() waits while the frames it already handed downstream are still playing, so a model that generates faster than its playout rate is throttled to that rate instead of piling up latency. The wait happens off the model’s event loop, so commands and lifecycle hooks keep dispatching while it holds. Rate limiting is the runtime’s job. A producer that would rather skip a frame than wait — anything driven by a live source, where the newest frame is the only one worth sending — passes drop=True, and the overflow is discarded downstream:

await self.emit(MyOutput(main_video=frame), drop=True)

Batches

Models that produce several frames per forward pass can emit them in one call. Pass a (N, H, W, 3) array and the runtime splits it into individual frames downstream:

frames = self.pipe.forward(prompt=self.prompt)
print(frames.shape)  # (10, 720, 1280, 3)
# ✅ Emit the whole batch in one call — the runtime paces it out
await self.emit(MyOutput(main_video=frames))

Emit the full batch in a single call rather than looping over it. The runtime uses the batch as the unit it paces against, so splitting it yourself produces choppier playback.

Frame rate

Each emitted chunk carries the rate its frames should play out at. There are two ways to set it. Declare a fixed rate with the fps class attribute when your model produces at a predictable speed:

class MyModel(ReactorModel):
    fps = 24

It defaults to 30. Measure your own compute time when throughput varies, and the runtime derives the rate from it:

t0 = time.perf_counter()
frames = self.pipe.forward()
# ✅ The chunk plays back at the rate you actually produced it
await self.emit(MyOutput(main_video=frames), compute_time=time.perf_counter() - t0)

The playback rate becomes n_frames / compute_time, so the stream tracks your real throughput. A model that speeds up after a warm-up, or slows down under a heavier setting, stays in sync without you touching fps. When you pass compute_time, fps is ignored.

Pass the honest measured time or none at all. Playout follows the tag, so a doctored compute_time makes the model permanently outrun its own playback.

Controlling playout

self.output is the model’s handle onto its outbound stream. It is bound for you, and every operation on it fans out to each connected client — including clients that connect later.

self.output.fps = 24     # re-pace what is queued, and tag emits from here on
self.output.flush()      # drop what is queued and cut playout to black
await self.output.emit(MyOutput(main_video=frame))   # self.emit() is an alias

Assigning fps re-paces frames that are already queued rather than waiting for the next emit, which is what makes it usable from a command handler:

@event(name="set_speed", description="Playback rate in frames per second")
async def set_speed(self, fps: float = InputField(default=24, ge=1, le=60)) -> None:
    self.output.fps = fps

The assignment holds until a chunk emitted with compute_time supersedes it, so it is a durable setting on a model that declares fps and a one-chunk nudge on a model that measures every pass. Call flush() when generation resets or restarts — a new scene, a cleared prompt, a seek. It drops everything queued, releases a producer waiting in emit(), and cuts the client to black, so none of the old content plays after the reset:

@event(name="reset", description="Clear the scene and start over")
async def reset(self) -> None:
    self._step = 0
    self.output.flush()

The recording keeps running across the cut, so a clip spanning a reset contains both sides of it.

Buffered latency

buffer_size declares how many frames may sit between your model and each client. It is the latency bound: a smaller value means a command takes effect on screen sooner, and a larger one absorbs more variance in your generation speed.

class MyModel(ReactorModel):
    fps = 24
    buffer_size = 8

Leave it undeclared to accept the runtime’s default. The bound is never applied below one emitted chunk, so a model that emits batches always fits a whole batch no matter how small the number. A declared value must be positive; zero or less fails at startup.

Sending messages

Send structured data to clients from inside run():

await self.send(Progress(step=self._step, total=100))

See Events & Messages for defining message types.

Concurrency

@event handlers run on the same event loop as run(), so a handler can fire at any await in your loop. If you read an attribute, await something, and read it again, it may have changed in between. Snapshot what a forward pass depends on before you start it:

async def run(self) -> None:
    while True:
        await self.connected.wait()
        while self.connected.is_set():
            # ✅ Snapshot before the forward pass — handlers can fire at any await
            prompt = self.prompt
            step = self._step
            frame = self.pipe.forward(prompt=prompt, step=step)
            await self.emit(MyOutput(main_video=frame))
            self._step += 1

This keeps one frame internally consistent. A command that lands mid-pass takes effect on the next frame rather than half-applying to the current one.

Sessions & Clients

Session hooks, multiple clients, and what resets when.

Managing State

One event loop, shared by your loop and your handlers.

Get Started

Build a Model

Media & Transport

Releases

Basic pattern

The connection signal

Emitting frames

Batches

Frame rate

Controlling playout

Buffered latency

Sending messages

Concurrency

Next

Sessions & Clients

Managing State

​Basic pattern

​The connection signal

​Emitting frames

​Batches

​Frame rate

​Controlling playout

​Buffered latency

​Sending messages

​Concurrency

​Next

Sessions & Clients

Managing State

Basic pattern

The connection signal

Emitting frames

Batches

Frame rate

Controlling playout

Buffered latency

Sending messages

Concurrency

Next