Metadata-Version: 2.4
Name: reactor-realtime-engine
Version: 0.2.0
Summary: Standalone realtime inference engine: the RealtimeInterface contract, a default scheduling loop, and a CPU reference engine — driven entirely through inbox/outbox queues
Author-email: Reactor Team <team@reactor.inc>
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: numpy>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Requires-Dist: black; extra == "dev"
Requires-Dist: isort; extra == "dev"
Requires-Dist: mypy; extra == "dev"

<!-- Copyright (c) 2026 Reactor Technologies, Inc. All rights reserved. -->
# realtime_engine — a standalone realtime inference engine

A standalone realtime inference engine, driven
entirely through inbox/outbox queues — so *any* caller matching the contract can
run it.

This follows the design's build order: **prove the engine on its own first** (the
queue boundary gives isolation now, and a path to a process/IPC split later
without a process boundary today). Full design write-up lives in Notion.

## Contents

- Layout
- Batched engines: the lockstep cohort
- Develop & test
- Contract recap
- Next (not built yet)

## Layout

Standard `src/` layout: `src/realtime_engine/` (package) + `tests/` + `pyproject.toml`.

| module | what |
|---|---|
| `contract.py` | `RealtimeInterface` (Protocol) + `EngineManifest` / `SessionInit` / `StepResult` / `AdmissionError` — Level 1: `manifest`/`attach`/`step`/`finalize`/`detach` |
| `queues.py` | the boundary: inbox msgs (`Attach`/`Detach`/`Reset`/`Action`) + `Chunk`; `Inbox`/`Outbox` (asyncio.Queue wrappers) |
| `serve.py` | `default_serve` — the Level-2 loop a Level-1 engine reuses (drain → batched `step` → outbox → finalize → tick; generate-from-default); `run_engine` picks an engine's own `serve` if it has one |
| `composed.py` | `ComposedEngine` — the generic **lockstep-cohort** batched engine (slot arena + two-stream output pipeline + isolated heavy-encode side lane), driven by a `StageModel`'s lane-tagged `Stage` list (`Lane` / `StageBatch`) |
| `cpu_reference.py` | `CpuReferenceEngine` — numpy-only deterministic fake (no GPU/ML); proves admission, batching, **invariance**, **distinctness** |
| `cpu_stage_model.py` | `CpuStageModel` — numpy-only reference `StageModel` (+ worked example); its chunk is a pure function of (seed, age, cheap conditioning, applied heavy value), so the cohort semantics are equality-assertable |
| `tests/test_engine.py` | standalone tests that feed the inbox and drain the outbox |
| `tests/test_composed.py` | the cohort semantics: arena/admission, invariance, distinctness, membership boundaries, deferred lag, heavy-encode isolation, failure recovery |

## Batched engines: the lockstep cohort

`ComposedEngine` is the batching machinery written **once**, so a model is just a
`StageModel`: a manifest, an ordered list of lane-tagged stages, and `seed`/`detach`.

- **Lanes.** `Lane.HOT` is the critical path (denoise/KV-finalize) and produces
  the latent. `Lane.DEFERRED` decodes chunk N-1 while HOT produces N: together
  they are the steady-state two-stream output pipeline. `Lane.ENCODE` batches
  changed rows on an on-demand third stream, so an over-budget prompt/image
  encode cannot queue ahead of the next decode and stall unrelated rows. ENCODE
  stages must mutate only those rows and avoid unsynchronized shared workspaces.
- **Arena.** `max_sessions` fixed rows; `attach` takes the lowest free row (over
  capacity → `AdmissionError` = backpressure, and re-attaching an already-attached
  session is idempotent); `detach` frees the row for reuse.
- **Arming.** Rows accumulate as *pending* until the optional
  `StageModel.seed_ready(pending)` says go (default: as soon as any row is
  pending); then `seed` is called once for the whole pending set, on the HOT
  stream at a step boundary. This is a scheduler gate, not an application-input
  gate: the caller should send `Attach` only after required prompt/image/scene
  context exists. A join preserves residents' in-flight output. A detach drains
  all execution lanes before releasing its row and removes only that row from a
  deferred snapshot. Drain the arena and the next attach re-seeds — that is what
  makes Reset (detach + re-attach) work. `midrun_attach=False` rejects a later
  arrival instead of seeding it incrementally.
- **Warm-up chunk.** When pipelined, `DEFERRED(chunk N-1)` runs on the overlap
  lane while `HOT(chunk N)` runs on the hot lane, so the first step emits nothing
  and `manifest().pipeline_depth_chunks` is 2 (else 1). This is distinct from
  `input_latency_chunks`: a cheap action can still affect the next generated
  chunk. Non-pipelined the two stages run sequentially in one step and emit
  immediately — the b=1 shape.
- **Bystander isolation.** A change to a *heavy* conditioning key (declared by the
  optional `StageModel.heavy_keys()`) batches all changed rows through `ENCODE`;
  each row sits out until its event completes (normally one chunk) and rejoins
  with the new value applied. A slow encode stays isolated while ready rows keep
  advancing. The default equality handles nested Python values and NumPy arrays;
  tensor/revision-heavy models can provide `heavy_equal()`.
- **Failure boundary.** `seed`, `detach`, and stage failures propagate after
  execution lanes are drained; the in-flight output batch is discarded rather
  than emitted stale or twice. Stage implementations must be boundary-atomic,
  because a generic scheduler cannot roll back a partially mutated KV/cache.
- **Torch-optional.** A lane is a real CUDA stream when torch + CUDA are present
  and a no-op passthrough otherwise, so there is exactly one step body and the CPU
  path never imports torch. `torch` is not a dependency: CUDA availability is
  probed cheaply at construction (so the manifest is stable), while stream
  *creation* is deferred to the first `step` — `torch.cuda.stream` is thread-local
  and the engine is constructed on the load thread but stepped on `default_serve`'s
  pinned worker thread.

## Develop & test

```bash
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
```

Tests cover: distinctness (co-resident sessions differ), invariance (a session is
identical solo vs batched), admission (over-capacity `attach` raises
`AdmissionError`), finalize, and output shape — plus, for the batched engines, the
slot arena and row reuse, membership only at chunk boundaries, the deferred lane's
exactly-one-chunk lag, once-per-change heavy encodes with bystander isolation,
lossless pipelined join/detach, due-subset ages, NumPy conditioning equality,
over-budget encode isolation, lifecycle stream ordering, and stage failure
recovery, all with the engine running standalone. Lint/type/test also run in CI
(`.buildkite/pipeline.yml`).

## Contract recap

- **Level 1** (every engine): `manifest` / `attach` / `step(due) -> [StepResult]` / `finalize` / `detach`. `step` is **batched** (one `StepResult` per due session).
- **Level 2** (the loop): optional `serve(inbox, outbox)`. Omit it → `default_serve`. Implement it → own loop (e.g. an sglang-style scheduler).
- **Conditioning** arriving via `Action` is **action-class** (already client→action-translated by the caller); the engine stays ignorant of client semantics.
- **Latency** uses separate manifest fields: `input_latency_chunks` for action-to-generation responsiveness and `pipeline_depth_chunks` for output warm-up.

## Next (not built yet)

- A real GPU model behind `ComposedEngine`: the Waypoint `StageModel` lives in
  reactor-models, in a separate PR — this package stays ML-free (sole dep: numpy).
- The runtime caller: a `ReactorModel` that pumps the inbox (state → `Action`) and routes the outbox into streaming.
- **Open validation item** (flagged in review): measure whether the outbox queue adds latency vs a direct callback.
