The Step Loop

Your model does not produce its output in one call. The runtime calls generate() again and again, and each call is one step that produces the next piece: a frame, a chunk of frames, a stretch of audio. A step takes the current state, calls generate(), and streams what comes back. Two optional hooks sit around generate(), one to prepare what goes in and one to shape what comes out. This page builds the loop up one piece at a time. The examples continue the SANA-WM world model from Managing State: a prompt, camera controls, and a chunk of frames per step.

The default step

Without the hooks, a step is one call. The runtime hands generate() the state and streams the Output it returns:

Client input enters from the left, carrying the state, and goes into the top of a generate() box. The output leaves from the bottom of the box and goes back out to the clients. — Without the hooks: the state goes into generate(), and what it returns goes out to the clients.

sana_wm.py

class SanaWMState(InputState):
    prompt: str = InputField(
        default="a misty valley", moderate=True
    )
    move: str = InputField(
        default="idle", choices=["idle", "forward", "back"]
    )
    yaw: int = InputField(default=0, ge=-1, le=1)
    pitch: int = InputField(default=0, ge=-1, le=1)


class SanaWMOutput(Output):
    main_video: Video


class SanaWM(ReactorApp):
    state: SanaWMState
    fps = 16

    def load(self, config_path: Path | None) -> None:
        weights = get_weights_path()
        self.pipe = SanaWMPipeline.from_pretrained(weights)

    def generate(self, input: SanaWMState) -> SanaWMOutput:
        # ✅ The state as clients last set it
        chunk = self.pipe.next_chunk(
            prompt=input.prompt,
            move=input.move,
            yaw=input.yaw,
            pitch=input.pitch,
        )

        # ✅ An Output, streamed as is
        return SanaWMOutput(main_video=chunk)

For many models this is the whole loop. It is the runtime’s default run(). A model that has to drive itself, to emit several times per step or block on an input before it can produce anything, can replace it; Controlling the loop shows how.

Handlers run between steps

Every @event handler and lifecycle hook waits for the step in flight to finish before it runs. A command that arrives while generate() is busy is not lost and is not applied halfway through; it runs in the gap before the next step.

A timeline of steps. A set_prompt command arrives during the first step, waits, and its handler runs in the gap before the second step, which reads the new prompt. — A command that arrives mid-step runs once the step is over. The next step sees what it changed.

So nothing changes under a step: the state it started with is the state it finishes with, and the model is not touched between the calls that make it up. A value a handler writes is read by the next step. A method a handler calls on the model is seen by the next generate(). Steps run only while a session is live and at least one client is connected. When the last client leaves, the loop stops at the next step boundary and waits.

Preparing the input

process_input() runs before generate(). Override it when the state is not quite what the model wants, or when there are moments the model should not run at all.

The state enters process_input() from the left. process_input() passes an input to generate(), and the output leaves generate() toward the clients. — process_input() takes the state and decides what generate() receives.

It takes no parameters. It reads self.state, and the media tracks if the model declares any, and returns whatever generate() should receive. That return type is yours. A small dataclass is the usual choice, because it gives generate() a signature that says exactly what one step needs:

sana_wm.py

@dataclass(frozen=True)
class SanaWMInput:
    prompt: str
    controls: tuple[str, int, int]
    first_frame: np.ndarray | None
    rollout_id: int


class SanaWM(ReactorApp):
    state: SanaWMState

    async def process_input(self) -> SanaWMInput:
        state = self.state

        # ✅ Decide whether this step should run at all
        if state.paused:
            raise ApplicationError("paused")
        if state._first_frame is None:
            raise ApplicationError("no first frame yet")

        # ✅ Work out what the model needs this time
        applied = state._applied_rollout_id
        new_rollout = state._rollout_id != applied
        first_frame = state._first_frame if new_rollout else None

        # ✅ Hand generate() exactly that, in a type you own
        return SanaWMInput(
            prompt=state.prompt,
            controls=(state.move, state.yaw, state.pitch),
            first_frame=first_frame,
            rollout_id=state._rollout_id,
        )

    def generate(self, input: SanaWMInput) -> SanaWMResult:
        return self.pipe.step(input)

Raising ApplicationError skips the step. The model is not called, the reason is logged once rather than on every turn, and the runtime asks again a few milliseconds later. This is how a paused model costs nothing, and how a model that needs an upload waits for it without special cases inside generate().

Shaping the output

process_output() runs after generate(). Override it when the model’s result is not already an Output, when a step should also send a message, or when the model can raise an error you want to recover from.

The state enters process_input(), which passes an input to generate(), which passes a result to process_output(), which sends the output to the clients. — process_output() takes the result and decides what the clients receive.

It receives a StepOutcome, which is the runtime’s envelope around what generate() did. Inside it, outcome.result is exactly the object generate() returned, in your type. It returns the Output to stream, or None to stream nothing this step:

sana_wm.py

@dataclass(frozen=True)
class SanaWMResult:
    frames: np.ndarray  # (24, H, W, 3) uint8
    chunk_index: int
    rollout_id: int


class SanaWM(ReactorApp):
    state: SanaWMState

    async def process_input(self) -> SanaWMInput:
        ...  # as above

    def generate(self, input: SanaWMInput) -> SanaWMResult:
        ...  # as above

    async def process_output(
        self, outcome: StepOutcome
    ) -> SanaWMOutput | None:
        # ✅ An error out of generate() lands here first
        if outcome.error is not None:
            raise outcome.error

        # ✅ The result, in your type, as generate() returned it
        result: SanaWMResult = outcome.result
        self.state._applied_rollout_id = result.rollout_id

        # ✅ A message sent here arrives before the frames
        if result.chunk_index % 20 == 0:
            await self.send(Progress(chunk=result.chunk_index))

        # ✅ Only an Output is media the runtime can stream
        tag = {"chunk": result.chunk_index}
        metadata = [tag] * len(result.frames)
        video = TrackPayload(result.frames, metadata=metadata)
        return SanaWMOutput(main_video=video)

StepOutcome carries result and error, exactly one of which is set, plus elapsed, the seconds the step took.

StepOutcome

The three fields, and to_output(), which is what the default process_output() calls.

When generate() raises

An exception out of generate() does not end anything by itself. It arrives in process_output() as outcome.error, and you decide. If it is an error the model is known to raise, recover: put the model back into a good state, cut playout if a stale frame must not follow, tell the client, and return None. The loop carries on with the next step.

async def process_output(
    self, outcome: StepOutcome
) -> SanaWMOutput | None:
    # ✅ An error you expect: recover and carry on
    if isinstance(outcome.error, RolloutExhausted):
        self.pipe.reset()
        self.output.flush()
        await self.send(RolloutRestarted(reason="limit reached"))
        return None

    # ✅ Anything else: let it end the session
    if outcome.error is not None:
        raise outcome.error

    ...  # the normal path, as above

Re-raise anything you did not expect. An exception out of process_output() ends the session with an error the client can see, which is better than streaming a broken world in silence.

Controlling the loop

Everything on this page so far is the runtime’s default run(). Once load() has finished, the runtime calls run() exactly once, and that call is the model’s whole life: as long as it is running, the model is up. The default implementation is the loop you have seen, waiting for a client, taking a step, streaming the result, and asking again. You can replace it. Override run() and you own the loop: when to wait, when to call the model, when to emit. The three hooks are not called anymore, and nothing reads the state for you. Commands and lifecycle hooks still run, and so do self.connected, self.send(), and the tracks.

class SanaWM(ReactorApp):
    def load(self, config_path: Path | None) -> None:
        weights = get_weights_path()
        self.pipe = SanaWMPipeline.from_pretrained(weights)
        self.prompt = "a misty valley"

    @event(name="set_prompt", description="The scene to show.")
    def set_prompt(self, prompt: str) -> None:
        self.prompt = prompt

    async def run(self) -> None:
        while True:
            # ✅ No client connected: wait for one
            await self.connected.wait()
            self.pipe.reset()

            # ✅ Your loop, for as long as a client is connected
            while self.connected.is_set():
                chunk = self.pipe.next_chunk(prompt=self.prompt)
                await self.emit(SanaWMOutput(main_video=chunk))

self.connected is set while at least one client is connected and cleared when the last one leaves, so the inner loop ends on its own. emit() puts an Output on the tracks and waits while the frames already handed over are still playing, which is what paces a loop like this. If run() returns or raises, the runtime ends the session with an error, so a loop that should live for the whole session is wrapped in while True. Write your own run() when the default does not fit: a model that emits several times per step, or one that has to block on an input before it can produce anything. A model whose whole job lives in @event handlers can simply park:

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

With a hand-written run(), keep your values in your own attributes and write @event handlers for them, as the example above does. Declaring state: still generates the set_<field> commands, but nothing reads the state on your behalf, and nothing bounds when a write lands relative to your loop’s reads.

Application and Model

How far to take the split between the code that answers clients and the code that runs weights.

Video & Audio Tracks

Read the client’s camera and microphone in process_input().

Get Started

CLI & Platform

Build a Model

Media & Transport

Releases

The default step

Handlers run between steps

Preparing the input

Shaping the output

StepOutcome

When generate() raises

Controlling the loop

Next

Application and Model

Video & Audio Tracks

​The default step

​Handlers run between steps

​Preparing the input

​Shaping the output

StepOutcome

​When generate() raises

​Controlling the loop

​Next

Application and Model

Video & Audio Tracks

The default step

Handlers run between steps

Preparing the input

Shaping the output

When generate() raises

Controlling the loop

Next