Deploy Matrix-Game 2.0 Real-time World Model with WMA

git clone https://github.com/fal-ai-community/fal-demos.git
cd fal-demos
uv venv --python 3.12
source .venv/bin/activate
uv pip install 'fal[wma]==1.82.1'
fal auth login

CONTRACT = fal.wma.RealtimeContract(
    media=fal.wma.MediaContract(
        receive=(fal.wma.Track(kind="video", width=WIDTH, height=HEIGHT),)
    ),
    client_messages=ClientMessage,
    server_messages=ServerMessage,
    interaction=CommandWorld(
        commands=CommandState(Commands, "/active", "/activated"),
        setup=SerialAction(Setup, Configured),
        reset=SerialAction(Reset, ResetApplied),
        errors=CommandErrors(fal.wma.ErrorMessage, "/error"),
        ended=fal.wma.StreamExhaustedMessage,
    ),
    client_message_examples=(
        fal.wma.MessageExample(
            name="Walk forward", payload={"type": "commands", "active": ["forward"]}
        ),
        fal.wma.MessageExample(
            name="Turn right while walking",
            payload={"type": "commands", "active": ["forward", "look_right"]},
        ),
        fal.wma.MessageExample(
            name="Stop", payload={"type": "commands", "active": []}
        ),
        fal.wma.MessageExample(name="Restart", payload={"type": "reset"}),
    ),
)

    def on_commands(self, message: Dict[str, Any]) -> None:
        try:
            parsed = Commands.model_validate(message)
        except ValueError as exc:
            self.session.send({"type": "error", "error": str(exc)[:200]})
            return
        self.keys.update(list(parsed.active), list(parsed.activated))
# ...
    def request_restart(self, scene: Optional[int], ack: type) -> None:
        with self.lock:
            self.restart = (scene, ack)
        # Drop frames of the old world that wait for playback.
        while not self.frames.empty():
            self.frames.get_nowait()
            self.flushed += 1

                            if not self.push(frame):
                                return
                            try:
                                self.session.add_billable_units(1)
                            except RuntimeError:
                                return  # Billing is final, so the session is closing.

class MatrixGameApp(
    fal.wma.App,
    python_version="3.12",
    keep_alive=300,
    min_concurrency=0,
    max_concurrency=2,
    max_multiplexing=1,
):
    machine_type = "GPU-H100"
    startup_timeout = 1800  # Loading about 12 GB of weights can take many minutes.
# ...
        "pydantic>=2,<3",
    ]
    realtime_contract = CONTRACT

    async def create_backend(self, session):
        from aiortc import RTCConfiguration

        # One session at a time: the KV caches belong to the runner, not the session.
        if not self.slot.try_acquire():
            raise fal.wma.AppError(
    # ...
        session.defer(self.slot.release)
        world = WorldSession(self.engine, session)
        # Deferred cleanups run in reverse: join the thread, cap billing, release the slot.
        session.defer(world.cap_billing)
        session.defer(world.stop_and_join)

        servers, _ = await fal.wma.RunnerIceConfig.from_bridge().build_ice_servers_async(
            session.offer.ice_servers, forwarded_status=session.offer.ice_status
        )
        track = fal.wma.make_video_queue_track(world.frames, FPS)
        world.thread.start()
        return fal.wma.AiortcPeer(
            session,
            lambda pc: pc.addTrack(track),
            rtc_configuration=RTCConfiguration(iceServers=servers),
            disconnected_grace_seconds=10,
        )

fal run fal_demos/video/wma/matrix_game.py::MatrixGameApp

fal deploy fal_demos/video/wma/matrix_game.py::MatrixGameApp --auth private

npm install @fal-ai/client@1.11.0-alpha.5 @fal-ai/server-proxy@1.2.1

import { fal } from "@fal-ai/client";
import { wma } from "@fal-ai/client/realtime/wma";

fal.config({ proxyUrl: "/api/fal/proxy" });

const keyMap: Record<string, string> = {
  w: "forward", s: "back", a: "left", d: "right",
  i: "look_up", k: "look_down", j: "look_left", l: "look_right",
};
const held = new Set<string>();
let controlsReady = false;
// The element needs the muted and playsinline attributes for autoplay: <video id="output" muted playsinline>
const video = document.querySelector<HTMLVideoElement>("#output")!;
// The world is ready when the first frame plays.
video.addEventListener("playing", () => {
  controlsReady = true;
});

let ended = false;

function endSession(message: string) {
  // A failure reaches both onError and onState, so run the cleanup once.
  if (ended) return;
  ended = true;
  controlsReady = false;
  held.clear();
  video.srcObject = null;
  console.log(message);
}

const session = fal.realtime.open(wma("your-account/<app-name>"), {
  receive: ["video"],
  onMedia(stream) {
    video.srcObject = stream;
    void video.play().catch(console.error);
  },
  onData: (raw) => {
    const msg = JSON.parse(raw);
    switch (msg.type) {
      case "reset_applied":
        console.log("World restarted from scene", msg.scene);
        break;
      case "error":
        console.error(msg.error);
        break;
      case "stream_exhausted":
        endSession("The world reached its length limit.");
        break;
    }
  },
  onState(state) {
    if (state === "closed" || state === "failed") endSession("Session ended.");
  },
  onError(error) {
    console.error(error);
    endSession("Session failed.");
  },
});
try {
  await session.ready;
} catch (error) {
  endSession("Session failed to start.");
  await session.close();
  throw error;
}

function report(tapped: string[] = []) {
  if (ended) return;
  session.send({ type: "commands", active: [...held], activated: tapped });
}

window.addEventListener("keydown", (event) => {
  const command = keyMap[event.key.toLowerCase()];
  if (!command || event.repeat || !controlsReady) return;
  held.add(command);
  report([command]);
});
window.addEventListener("keyup", (event) => {
  const command = keyMap[event.key.toLowerCase()];
  if (!command) return;
  held.delete(command);
  report();
});
window.addEventListener("blur", () => {
  held.clear();
  report();
});
window.addEventListener("pagehide", () => void session.close());

    def push(self, frame: Any) -> bool:
        """Queue a frame. Block while the queue is full. False when the session stops."""
        future = asyncio.run_coroutine_threadsafe(self.frames.put(frame), self.loop)
        while not self.stop.is_set():
            try:
                future.result(timeout=0.2)
                return True
            except concurrent.futures.TimeoutError:
                continue
        future.cancel()
        return False