Events & Messages

Besides video and audio, a model and its clients talk to each other with structured data. A client sends a command to the model. The model answers, and it can also send a message to the client at any time: a status, a result, an event the client should react to. Both directions are declared in your class. @event marks a method as a command handler, and a ModelMessage subclass declares a message. The runtime turns those declarations into the model’s schema, which is what typed client SDKs are generated from. The examples on this page come from the runtime’s Waypoint example, a world model that starts from an uploaded image.

Defining a command

The state gives you one set_<field> command per field, for values a client sets and the model keeps reading. For anything else, write a handler with @event. The method’s parameters are the data the client sends:

waypoint.py

from reactor_runtime import InputField, UploadedFile, event


@event(
    name="set_image",
    description="Upload the seed frame. The next step starts a new world from it.",
)
async def set_image(
    self, image: UploadedFile = InputField(description="The seed frame.")
) -> None:
    seed = await asyncio.to_thread(_fit, image.data)
    self.state._seed = seed
    self.state._seed_id += 1

The client calls it as set_image with an image parameter. The runtime validates the payload against the signature before your method runs, and it runs between steps, so the state is never changed while generate() is reading it. A handler can be async def or a plain def. A command does not need parameters. This one restarts the world from the image the model already has:

waypoint.py

@event(name="reset", description="Restart the world from the current seed frame.")
def reset(self) -> None:
    self.engine.reset()
    self.state._applied_seed_id = None
    self.output.flush()

Constraining a parameter

InputField on a parameter carries the same rules as on a state field, and the runtime checks them before your handler runs:

strength: float = InputField(default=0.5, ge=0.0, le=1.0)
prompt: str = InputField(default="hello", min_length=1, max_length=500)
style: str = InputField(default="none", choices=["none", "oil_paint", "sketch"])

A value outside the rules is refused with invalid_command and your handler is not called. The constraints also travel into the schema, so a client sees the accepted range before it sends anything.

Replying to a command

A client that sends a command usually wants to know what happened. Return a ModelMessage from the handler and the runtime delivers it to that client as the reply to that exact command. A client awaiting the call receives it as the result:

waypoint.py

class WaypointStatus(ModelMessage):
    has_image: bool = MessageField(description="Whether a seed frame has been accepted.")
    paused: bool = MessageField(description="Whether generation is held.")
    step_index: int = MessageField(
        description="The last completed step within the current world, or -1 before the first."
    )


@event(name="reset", description="Restart the world from the current seed frame.")
def reset(self) -> WaypointStatus:
    self.engine.reset()
    self.state._applied_seed_id = None
    self.output.flush()
    return WaypointStatus.of(self.state, -1)

status = await reactor.send_command("reset", {})
# {"type": "waypoint_status", "data": {"has_image": true, "paused": false, "step_index": -1}}

const status = await reactor.sendCommand("reset", {});

Return None when there is nothing to say. Reply when the client should confirm what took effect: a clamped value, a resolved default, the state after a change. The return annotation has to be one ModelMessage subclass or None. A union such as WaypointStatus | None is rejected when the class is declared, because the schema publishes one reply shape per command.

Reporting a failure

When a command cannot be carried out, raise CommandError with a code and a message. Both reach the client as the reply to that command, so an awaiting caller gets an error instead of waiting forever:

waypoint.py

from reactor_runtime import CommandError


@event(name="set_image", description="Upload the seed frame.")
async def set_image(
    self, image: UploadedFile = InputField(description="The seed frame.")
) -> WaypointStatus:
    if not image.mime_type.startswith("image/"):
        raise CommandError("unsupported_media", f"{image.name} is not an image.")
    ...

Write the message for the client, not for a log. Any other exception is treated as a bug: the runtime logs it with its traceback and answers the client with a generic internal_error, so paths, queries, or credentials in the exception text never leave the container. The runtime uses the same channel for its own refusals: invalid_command when a payload fails validation, and unresolved_upload when a command references a file that cannot be fetched.

Sending messages

A ModelMessage subclass declares a typed message. Send one with self.send() from anywhere in your class, and every connected client receives it:

waypoint.py

class WaypointStatus(ModelMessage):
    has_image: bool = MessageField(description="Whether a seed frame has been accepted.")
    paused: bool = MessageField(description="Whether generation is held.")
    step_index: int = MessageField(description="The last completed step, or -1 before the first.")


async def process_output(self, outcome: StepOutcome) -> WaypointOutput | None:
    ...
    if result.index % 50 == 0:
        await self.send(WaypointStatus.of(self.state, result.index))
    return WaypointOutput(main_video=TrackPayload(result.frames, metadata=metadata))

The client receives {"type": "waypoint_status", "data": {...}}. The wire type is the snake_case form of the class name. MessageField is optional; it attaches a description or a default that shows up in the schema, and the class docstring documents the message itself. A message with no fields is valid and is the right shape for a pure signal. A message sent from process_output() goes on the wire before that step’s frames, so a client that reads the status and then sees the frame gets them in that order.

Messages to one client

self.send() goes to everyone. To reach the client a handler or hook is running for, add a client: ClientInfo parameter. The runtime fills it in, and client.send() delivers to that client alone:

waypoint.py

@connected
async def on_connect(self, client: ClientInfo) -> None:
    await client.send(WaypointStatus.of(self.state, self.last_index))

This is how a client that joins a running session gets a snapshot without every other client receiving it too. Sessions & Clients covers the client handle and serving several clients at once.

File uploads

To receive a file with a command, annotate a parameter as UploadedFile. The runtime fetches the upload and hands your handler the bytes:

@event(name="set_reference_image", description="Image the style is drawn from")
async def set_reference_image(
    self,
    reference_image: UploadedFile,
    strength: float = InputField(default=0.5, ge=0.0, le=1.0),
) -> None:
    if not reference_image.mime_type.startswith("image/"):
        raise CommandError("unsupported_media", "Expected an image.")
    self.state._reference = decode(reference_image.data)

UploadedFile carries name, mime_type, data (the raw bytes), and size. Check mime_type before decoding, because a client can upload anything. A parameter can also be a list[UploadedFile], and an upload nested inside a dict or a dataclass field resolves the same way. A client can also upload a file on its own, outside any command. Declare a @file_uploaded hook to receive those:

from reactor_runtime import ClientInfo, UploadedFile, file_uploaded


@file_uploaded
async def on_file(self, uploaded_file: UploadedFile, client: ClientInfo) -> None:
    if uploaded_file.mime_type.startswith("image/"):
        self.state._reference = decode(uploaded_file.data)

The handler takes one parameter named uploaded_file, plus the optional client. Prefer putting a file on the command it belongs to, so the schema tells the client what the file is for. Use @file_uploaded for a drop zone or a client that uploads first and decides later.

Sessions & Clients

Session and connection hooks, the client handle, and serving several clients.

Managing State

The state object and the commands it generates.

Get Started

CLI & Platform

Build a Model

Media & Transport

Releases

Events & Messages

Defining a command

Constraining a parameter

Replying to a command

Reporting a failure

Sending messages

Messages to one client

File uploads

Next

Sessions & Clients

Managing State

​Defining a command

​Constraining a parameter

​Replying to a command

​Reporting a failure

​Sending messages

​Messages to one client

​File uploads

​Next

Sessions & Clients

Managing State

Defining a command

Constraining a parameter

Replying to a command

Reporting a failure

Sending messages

Messages to one client

File uploads

Next