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. Every model refuses a command this way:

waypoint.py

from reactor_runtime import CommandError


@event(
    name="set_image",
    description="Upload the seed frame. Fails with `unsupported_media` when the file "
    "is not an image.",
)
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.")
    ...

The code is a short snake_case token, and clients branch on it. Keep each code the same from one release to the next. Name each code in the command’s description, so the schema tells a client what can fail. The message crosses the wire unchanged. Write it for the client, not for a log. A raise has three effects:

The caller learns on its own call. The runtime answers the command’s request with a failure that carries the code and the message. In Python, send_command() raises a ReactorError. In JavaScript, sendCommand() resolves undefined and reports the failure on the error event.
Only the caller gets the failure. The other clients in the session do not see a refusal of a command they did not send.
The runtime records that the command failed. Some sessions start with a step count. In such a session, a raise from a starting command ends the session and fails the steps that remain.

Never send an error and return

Do not refuse a command with await self.send(...) followed by a bare return. The runtime then sees a success and answers the caller with an empty acknowledgement. A JavaScript caller resolves exactly as it does on success. Every other client receives a message about a command it did not send. A session started with a step count keeps running with nothing to do. Do not return a different message type to report a failure either. The return annotation is the reply the schema publishes, and a client reads that type’s fields from whatever you return. If your published schema already declares an error message that clients listen for, keep sending it just before the raise until your next major release. Removing a message breaks the clients that listen for it.

from reactor_runtime import CommandError
from my_model_types import CommandError as CommandErrorMessage  # your published message

await self.send(CommandErrorMessage(command="enqueue", reason="The queue is full."))
raise CommandError("queue_full", "The queue is full.")

Other exceptions

The runtime treats any other exception as a bug. It logs the exception with its traceback and answers the client with internal_error and a fixed message. Paths, queries, or credentials in the exception text never leave the container. When a check raises its own exception, such as a ValueError from a decoder, catch it and raise a CommandError with your own code:

try:
    seed = decode(image.data)
except ValueError as error:
    raise CommandError("invalid_image", "The file is not a readable image.") from error

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.

Which clients receive the failure

Every SDK Reactor publishes receives a raised failure: JavaScript 3.x, Python, C++, Swift, and Java. The retired JavaScript SDK 2.x (@reactor-team/js-sdk before 3.0) does not. Its commands carry no request id, so the runtime cannot connect a failure to them.

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

Never send an error and return

Other exceptions

Which clients receive the 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

​Never send an error and return

​Other exceptions

​Which clients receive the 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

Never send an error and return

Other exceptions

Which clients receive the failure

Sending messages

Messages to one client

File uploads

Next