Reactor Realtime Engine

A small, model-agnostic package for serving stateful models that generate per-session output chunks. It defines the engine lifecycle, a reusable scheduling loop, and implementations for batched and per-session execution.

The package depends only on NumPy. Models may use PyTorch, CUDA, or another backend without adding those dependencies to the engine.

Contents

• Install
• Quick start
• Choose an engine
• ComposedEngine
• Contract
• Latency
• Develop and test

Install

pip install reactor-realtime-engine

Python 3.10 or newer is required.

Quick start

from realtime_engine import ComposedEngine, SessionInit
from realtime_engine.examples import CpuStageModel

engine = ComposedEngine(
    CpuStageModel(max_sessions=2),
    pipeline=False,
)

engine.attach(SessionInit("alice", {"seed": 1, "prompt": "a castle"}))
engine.attach(SessionInit("bob", {"seed": 2, "prompt": "a desert"}))

results = engine.step(
    {
        "alice": {"move": 1},
        "bob": {"move": -1},
    }
)

for result in results:
    print(result.session_id, result.output.shape, result.stats)

engine.finalize([result.session_id for result in results])
engine.detach("alice")
engine.detach("bob")

CpuStageModel is a deterministic example. Production models implement StageModel or the lower-level RealtimeInterface contract.

Choose an engine

• ComposedEngine shares a fixed-row batch for efficiency. Model stages must keep per-row state isolated.
• ReplicaEngine gives each session an independent B=1 child. This simplifies state isolation but provides no cross-session batching benefit and may use more memory.
• A custom RealtimeInterface provides full control for models that need their own scheduling or execution strategy.

ReplicaEngine pools successfully detached child engines for reuse. It is also useful as a B=1 correctness reference while adding batched model support.

ComposedEngine

ComposedEngine drives a StageModel through HOT generation, optional DEFERRED decoding, and isolated ENCODE work for changed heavy conditioning. It owns admission, fixed-row batching, cohort membership, and completion events. Pipelining uses lazily created CUDA streams when available and otherwise runs the same schedule sequentially.

Contract

Every engine implements manifest, attach, step, finalize, and detach.

• attach() raises AdmissionError when capacity is unavailable.
• step(due) advances a batch of {session_id: conditioning} entries. It may return fewer results during pipeline warm-up or temporary row isolation; callers must not assume one result per due session.
• SetActive(session_id, False) removes an attached session from scheduled steps while preserving its model state and latest conditioning.
• finalize() performs optional post-chunk maintenance.
• detach() releases session state and is idempotent.

An engine may implement serve(inbox, outbox, *, stop) to own its scheduling loop. Otherwise, run_engine() uses default_serve(), which serializes lifecycle calls on one worker thread and paces toward the declared rate.

Inbox and Outbox are in-process queue adapters for lifecycle messages and generated chunks. Reactor Runtime's RealtimePipeline connects these queues to client state, media tracks, output pacing, and connection lifecycle.

Latency

• input_latency_chunks: when boundary-aligned conditioning affects generation.
• pipeline_depth_chunks: how many scheduler ticks prime the output pipeline.

action-to-output chunks = input_latency_chunks + pipeline_depth_chunks - 1

End-to-end action-to-pixels latency also includes queueing, transport, output pacing, and rendering.

Develop and test

cd backend/realtime_engine
pip install -e ".[dev]"
ruff check src tests && black --check src tests && isort --check-only src tests
mypy src/realtime_engine
pytest -q

The test suite covers admission, deterministic row reuse, solo-versus-batched invariance, cross-row isolation, pipeline warm-up, heavy-conditioning changes, lifecycle ordering, failure recovery, replica pooling, and B=1 parity.