omni-adapter (Python)

Join a Connect Omni stage from Python — scripts, services, hardware bridges (Raspberry Pi, serial gateways), AI loops — and exchange messages with prototypes in real time. Mirrors the TypeScript adapter's wire contract and concepts on an asyncio surface.

First time? The step-by-step guide for a first connection (the adapter key, the install and the adapter name, in all six languages) is docs/getting-started.md, shipped in the adapter's npm and NuGet packages. Everything a Python program needs is also in this README and in llms.txt.

For AI assistants — the full usage guide llms.txt ships in the package — omni_adapter/llms.txt, printed by python -c "import omni_adapter, pathlib; print(pathlib.Path(omni_adapter.__file__).with_name('llms.txt'))". Must-know rules for Python:

• Credential: only a cst_… adapter token (Share → Adapter keys → Create key, then Copy) or a bare ck_… key with url= + stage_id=; read it from an environment variable, never hard-code it. Whitespace around it → ConfigError.
• Close on exit: run inside with omni_adapter.stop_on_signals(stop.set): and await client.close() in a finally within it — the first SIGINT/SIGTERM sets stop, close() waits up to 1 s for the server to confirm the leave, and a second signal exits at once (a bare try/finally loses the leave on Ctrl+C under Python 3.10, and loop.add_signal_handler does not exist on Windows). An unclean exit leaves the old seat in the member list for a few seconds.
• Lifecycle events: client.on("stateChange" | "error" | "memberJoin" | "memberLeave" | "history" | "ended" | "queueOverflow", cb) — camelCase strings on the same on(); a snake_case name such as "state_change" subscribes to a channel of that name and never fires. Callbacks run on the asyncio loop in arrival order; never block in one.
• latest() is not a delivery check: it folds your own sends in. To confirm a publish use ack=True (a server ack, not a delivery receipt) or watch the channel from a second connection.
• exclusive_name standby: only instances on the same adapter key; while standby new sends are dropped (or, acked, raise Standby), and becoming standby drops whatever was queued — nothing stale goes out on promotion. Read client.standby.
• 60 messages a second per connection: the adapter paces sends to the server's limit; a faster source fills the queue and loses its oldest values — throttle() it. close() drops queued fire-and-forget sends.
• A plain def callback cannot await: client.send(...) inside it only creates a coroutine that is never awaited — nothing is sent. Make the callback async def and await, or schedule it with asyncio.create_task(client.send(...)).

import asyncio
import os

import omni_adapter


async def main():
    # One credential is all you need (stage → Share → Adapter keys → Create key, then Copy):
    token = os.environ.get("CONNECT_TOKEN", "")  # cst_… — from the environment, never hard-coded
    client = await omni_adapter.connect(secret_key=token, display_name="Sensor Bridge #1")

    @client.on("hvac.setTemp")
    def on_set_temp(msg):
        print("prototype asked for", msg.value)

    stop = asyncio.Event()
    # Ctrl+C, docker stop, systemd: the first signal sets `stop`; a second one exits at once
    with omni_adapter.stop_on_signals(stop.set):
        try:
            await client.send("temp.reading", 23.5)
            await stop.wait()  # keep the bridge running
        finally:
            await client.close()  # leave the stage — the seat is freed at once


try:
    asyncio.run(main())
except KeyboardInterrupt:  # Ctrl+C before the handlers are installed, or a second Ctrl+C
    pass

stop_on_signals matters on Python 3.10: there, a Ctrl+C that reaches asyncio.run as KeyboardInterrupt cancels every task at once — the socket's own included — so the leave sent from the finally does not reach the server (Python 3.11+ delivers it either way). Inside the block the first SIGINT or SIGTERM only calls stop.set on the loop, so the leave goes out; a second signal acts as if no handler were installed (Ctrl+C raises KeyboardInterrupt, a SIGTERM with the default handler ends the process by that signal), so a close() held up by a slow server never traps you; leaving the block restores the previous handlers, so a Ctrl+C after asyncio.run returns behaves as usual. A send waiting for its ack keeps going after the first Ctrl+C until it settles (at most the ack timeout, 15 s by default); a second Ctrl+C exits at once. Measured on macOS with Python 3.9, 3.10, 3.11 and 3.14 against a server that counts leaves: with this snippet the leave arrived on every try (three Ctrl+C and two SIGTERM per version), while a bare try/finally lost it on all three Ctrl+C tries under 3.9 and 3.10.

It uses signal.signal plus loop.call_soon_threadsafe, the mechanism Python 3.11's asyncio.Runner uses for Ctrl+C. On Windows, the handler runs promptly on Python 3.10 because the default Proactor event loop wakes itself on a signal (it registers its self-pipe with signal.set_wakeup_fd, since Python 3.8); under WindowsSelectorEventLoopPolicy there is no wakeup fd, so the stop waits until the loop's next select returns. Windows never delivers SIGTERM (os.kill(pid, signal.SIGTERM) is TerminateProcess), so the SIGTERM half is POSIX-only. The Windows part follows from reading the CPython source, not from a measurement on Windows. loop.add_signal_handler does not exist on Windows.

Install

pip install omni-adapter

The module is omni_adapter. Requires Python ≥ 3.10. Runtime deps: python-socketio, aiohttp (plus python-engineio, which python-socketio brings anyway — it is listed only to pin its version range).

Working from a checkout of the adapter repository instead (contributors with access to it)? pip install <repo>/python, where <repo> is the checkout; the [stream] extra is pip install '<repo>/python[stream]'.

Server requirements

Joining, sending, receiving, acks and reconnects work against any Connect Omni server, but ck_ adapter keys, strict_routing, exclusive_name, provides_view, screen broadcasting and remote input need an adapter-enabled server (or web build). The minimum server for each feature is in the compatibility matrix, docs/COMPATIBILITY.md in the adapter's npm and NuGet packages. On an older server a bare ck_ key is refused as a bad share token (AuthRevoked from connect()) and ensure_stream_layer() raises NotSupported, while the declarations degrade without an error: the server strips strict_routing (plain broadcast), exclusive_name (instances stay independent) and provides_view (the adapter never appears as a screen candidate), and no viewer shows operate controls for remote input.

Connecting

Two forms, same as the TS adapter. Both carry an adapter key (ck_…) issued from Stage → Share → Adapter keys; anything else (a share token, for instance) is a ConfigError before any network call.

client = await omni_adapter.connect(secret_key="cst_…")                     # ① self-describing stage token (recommended)
client = await omni_adapter.connect(                                        # ② bare key + explicit coordinates
    url="http://192.168.0.10:9981", stage_id="01HXYZ", secret_key="ck_…",
)

Declarations and selective delivery work identically to TS:

client = await omni_adapter.connect(
    secret_key="cst_…",
    display_name="RPi Bridge",
    os_hostname=__import__("socket").gethostname(),
    receives=["actuator.cmd"], sends=["sensor.temperature"],
    strict_routing=True,                     # server delivers ONLY declared channels
)

Without strict_routing, receives may use wildcards ("robot.*"); with it, only exact names.

The adapter name and restarts

The key name typed when creating the key only labels it in the key list — the name the stage shows is display_name (default Omni adapter). The same adapter key is never renamed — a restart keeps its name. Another program connected under the same name with a different key makes yours join as name (2); for a provides_view="stream" adapter, so does an Adapter Stream layer with that name that is bound to another adapter key that is still valid — usually a rotated key whose old copy was never revoked: revoke it. Read the name you got from client.me.display_name. A person who takes a program's name is the one renamed, never the program. (Older servers, from before same-key name sharing, renamed a restart that came within a few seconds of an unclean exit to name (2).)

A program that exits without close() stays in the member list for a few seconds. A service manager's SIGTERM ends a Python process without running any finally unless you catch the signal — the quickstart above handles SIGINT and SIGTERM the same way: the signal sets an event, and finally awaits client.close(), which waits up to 1 s for the server to confirm the leave.

Networks that block WebSocket — transports=["polling"]

Behind a proxy that refuses the WebSocket upgrade, pass transports=["polling"]. python-engineio connects with the first transport in the list only, so the default ["websocket", "polling"] does not fall back to polling on its own.

The Connect server packs every queued message into one long-poll response, and python-engineio rejects a response with more than 16 packets (dropping the connection and every message in it). So whenever "polling" is in transports (the default list included), the adapter raises engineio.payload.Payload.max_decode_packets to 100 000. ⚠️ That is a process-wide class attribute: it also raises the per-request packet limit of any python-engineio server running in the same process. The adapter only ever raises it, and only once per process — set your own value after connecting if you need another one.

API sketch

await client.send("door.open", True)                 # fire-and-forget
await client.send("door.open", True, ack=True)       # wait for the server ack
await client.send_to(participant_id, "cmd.stop", 1)  # unicast
await client.send_json("robot.status", {"pose": [1, 2]})

@client.on("button.tap")            # exact channel (sync or async callback)
def cb(msg): ...
@client.on("robot.*")               # wildcard — one or more dot segments
def cb2(msg): ...
@client.on_json("robot.status")     # parsed JSON values: cb(value, msg)
def cb3(value, msg): ...

client.latest("temp.reading")        # last value (history-seeded, includes own sends; 1000 most recent channels)
client.latest_json("robot.status")
client.history()                     # replayed ring snapshot (≤1000, not persistent)
client.history("chat.*")             # …filtered to one channel (wildcards ok)
client.history_json("chat.message")
await client.members()  # each Member also carries receives · sends · client_type · strict_routing · exclusive_name · os_hostname (absent on older servers); member.declares_channel("robot.status")
await client.listeners("robot.status")  # who will receive it: .declared / .broadcast (self excluded); records no-listener when nobody declared it

push = client.throttle("nav.position", hz=10)  # coalesced sender (trailing flush)
push(json.dumps({"lat": 37.56, "lng": 126.97}))

# Lifecycle events share on() and use these exact camelCase names:
client.on("stateChange", lambda state: ...)      # "connecting" | "connected" | "reconnecting" | "ended"
client.on("error", lambda err: ...)              # AdapterError — subscribe in every headless program
client.on("memberJoin", lambda member: ...)      # Member
client.on("memberLeave", lambda participant_id: ...)
client.on("history", lambda messages: ...)       # the (re)join snapshot (list of StageMessage)
client.on("ended", lambda reason: ...)           # str | None — the client is finished
client.on("queueOverflow", lambda dropped: ...)  # PublishInput the offline queue dropped
await client.close()                              # leaves the stage; waits ≤1 s for the server's confirmation

Callbacks. on() returns an unsubscribe function (the decorator form returns your function; unsubscribe it with client.off(name, fn)). Callbacks are invoked one at a time in arrival order on the asyncio loop, so a plain def callback must not block — a slow one holds up every event behind it. A plain def callback also cannot await: calling client.send(...) there creates a coroutine that is never awaited, and nothing is sent (Python only prints a RuntimeWarning). Use an async def callback — the adapter runs it as a task, keeps a reference and logs its exceptions — or schedule the send yourself:

@client.on("actuator.cmd")
async def on_cmd(msg):                       # async def: await works here
    await client.send("actuator.ack", msg.value)

background = set()                           # asyncio keeps only weak references to tasks

@client.on("actuator.ping")
def on_ping(msg):                            # plain def: schedule the coroutine instead
    task = asyncio.create_task(client.send("actuator.pong", msg.value))
    background.add(task)
    task.add_done_callback(background.discard)

async def callbacks start in arrival order but interleave with the next callback after their first await. An exception in a callback is logged and never reaches the adapter. Prototypes and the Console send every value as a string ("42", not 42) — or null when the message carries none — so convert before comparing; values from other programs keep their type.

Roster. await client.members() returns a list[Member] (you included) and await client.listeners(channel) a Listeners with channel, declared and broadcast (tuples of Member, you excluded). Re-read members() after every (re)connect — joins and leaves that happened while you were disconnected are not replayed as events. members() and listeners() are server round trips: they fail with NotConnected while the client is not connected.

Sending. send() validates the channel and value when you await it and raises ValidationError for a bad one; send_json() raises ValidationError for a value that is not plain JSON (a dataclass, datetime, set, bytes, NaN/Infinity, a circular reference) — convert it first. The push function throttle() returns validates each value where you call it and raises ValidationError for a bad one; a valid value is delivered fire-and-forget. ack=True waits for the server's ack (not a delivery receipt from a peer): no ack within ack_timeout_s (15 s) raises AckTimeout, which does not prove the server never got it, and an acked send in flight when you close() or the socket drops raises AckTimeout at once. ack_timeout_s must be above zero and at most 2147483.647 s (2^31−1 ms) — anything else, None and math.inf included, fails with ConfigError before the client contacts the server. A plain send is at-most-once, direct or queued: it asks for no ack, so the adapter never sees a server rejection of it (the rate limit included), and one lost to a drop right after it went out is not retried — only acked sends see the server's answer. A queued acked send that already went out once before a drop and is later failed with QueueOverflow, Standby or SessionEnded may have reached the server too — on a queued acked send, none of these codes proves non-delivery. Because your own sends fold into latest(), it cannot tell you whether a send reached the server — use ack=True for that.

Offline sends queue (default 1000, oldest dropped with a queueOverflow event) and flush in order on reconnect, as fast as the rate limit below allows — plain items without waiting for acks, an acked item only after its server answer — (including after a session-level rejoin — auto_rejoin=True by default waits for a closed stage to reopen and rejoins, while a revoked credential ends immediately); TooManyRequests from the server (only acked sends see it) pauses sending with exponential backoff (1s → ×2 → 60s) and retries from the queue — silently, with no send-dropped entry in client.diagnostics(). close() drops queued fire-and-forget sends (queued acked sends fail with SessionEnded): to be sure the last values went out, await an acked send or wait until queued_count is 0 first. Reconnects revalidate the adapter key automatically; a revoked key surfaces as AuthRevoked. A network blip (reconnect=True, the default) reconnects with the same token — no new validation, so the participant does not change — every reconnect_delay_s × 2ⁿ up to reconnect_delay_max_s, each wait randomized by ±20% so adapters do not all come back at the same instant after a server restart.

Rate limit — 60 messages a second. The server accepts at most 60 messages a second per connection and silently drops fire-and-forget messages above that, so the adapter paces every send — direct or queued — with a token bucket (60 per second, bursts of up to 30); a send with no token free waits in the queue. Keep a source above 60/s up and the queue fills: the oldest items are dropped (a queueOverflow event for each, one queue-overflow line in the diagnostics ring per episode) and the newest value still goes out. After a disconnect, a sender at 60/s or more stays behind by up to queue_limit ÷ 60 seconds (the queue drains at 60 a second, so a full default queue of 1000 takes about 17 s) — throttle it or lower queue_limit. Acked sends go one per round trip, so never use ack=True for a high-rate stream.

Errors are AdapterError with .code, .message and often .hint — a one-line next action. Several codes carry no hint (AckTimeout, SessionEnded, most ValidationErrors, some ConfigErrors), and an error without a message of its own has the code as its message (AckTimeout), so print the code, whichever text exists, and which call failed:

client.on("error", lambda err: print(f"[{err.code}] {err.hint or err.message}"))

Console output. The adapter logs through the standard logging logger omni_adapter. Until you configure logging, its warnings reach stderr through Python's last-resort handler (for example ended: client-closed); configure logging to route them, or quiet them with logging.getLogger("omni_adapter").setLevel(logging.ERROR).

One active per name — exclusive_name

exclusive_name=True declares that the display name is this adapter's contract: the server keeps only the newest instance active and marks the others standby, and the client then stops sending while standby (receiving continues, and it is promoted automatically when the active one leaves). client.standby exposes the state. It works only between instances that use the same adapter key — a copy with a different key joins as name (2) instead.

While standby, new sends are dropped (fire-and-forget, debug log only) or raise Standby (ack=True) — never deferred. When an instance becomes standby — after a reconnect or when presence changes — everything it had queued is dropped the same way (the standby entry in client.diagnostics() ends with — dropped N queued sends), so a later promotion sends nothing stale. After each (re)connect the adapter needs one roster round trip to judge standby; sends made meanwhile are held (sharing queue_limit), then queued if this instance is active (or the judgment failed) or treated by the standby rule. There is no standby-change event — read client.standby, or look for the standby / active entries in client.diagnostics().

Error codes

Every failure is an AdapterError with .code (branch on this — it is the stable identifier), .message ("what went wrong") and often .hint — a one-line next action; print err.hint or err.message. Errors raise from the call that caused them, and lifecycle failures also arrive on client.on("error", …).

An AuthRevoked whose message says stage-locked means the stage is closed to guests, and adapters join as guests; the hint asks you to have the stage owner (or whoever runs your Connect) reopen it to guests.

A 400 from the join carries the server's own issue list instead of a bare (400) — that failure is almost always a version skew between server and adapter, not a bad token:

[ValidationError] token/validate rejected the request body (400):
clientType — Invalid literal value, expected "adapter"

Up to three fields are shown, then (+N more).

CLI — omni-adapter (no code required)

Installing the package adds an omni-adapter command with the same subcommands as the npm package — doctor, send, watch, record, replay, simulate, latency, report — with identical flags, output and exit codes. The credential always comes first. If the npm one is installed too, call the Python one unambiguously as python -m omni_adapter.

python -m omni_adapter doctor "$CONNECT_TOKEN"                     # diagnose the connection
omni-adapter send   "$CONNECT_TOKEN" door.open true                # publish (acked → exit 0/1)
omni-adapter send   "$CONNECT_TOKEN" robot.status '{"x":1}' --json # JSON convention
omni-adapter send   "$CONNECT_TOKEN" cmd.stop 1 --to <participantId>   # unicast
omni-adapter watch  "$CONNECT_TOKEN" 'chat.*' --count 5 --timeout 30   # NDJSON lines
omni-adapter watch  "$CONNECT_TOKEN" --history --timeout 10        # include replay backlog

watch prints one JSON object per line ({ts, channel, value, from}) — composable with jq — and exits on --count, --timeout, session end or Ctrl+C; without --count/--timeout it runs until the session ends. It exits 0 even when --timeout passes with zero messages, so to check that a peer is alive, count the output lines rather than trusting the exit code:

test "$(omni-adapter watch "$CONNECT_TOKEN" 'line.*' --count 1 --timeout 5 | wc -l)" -ge 1 && echo alive

Doctor — diagnose the connection

doctor walks credential → url shape → server reachable (GET /api/about) → adapter screens → remote input → stage session (token/validate) → a real socket join, stops at the first failure and prints the fix. Only ✖ lines need fixing — △ (warning) and · (skipped) lines are information, and the exit code is 1 only when a step fails. It checks the connection, not delivery: a green doctor with a peer that still does nothing points at the channel name, routing or the receiving side.

The adapter-screens line catches a server that predates providesView (the adapter connects but no layer can pick it) — it warns rather than fails, and its probe creates no member. The remote-input line says whether the server has the operate surface at all (standalone servers do). After joining it also lists the screen adapters on the stage by display name — the name a screen layer must be pointed at — and the same field is on the roster (Member.provides_view).

The same sequence is importable. Run it where connect() fails, so the report lands in your program's own output:

import asyncio
import os

import omni_adapter


async def main():
    token = os.environ.get("CONNECT_TOKEN", "")
    try:
        client = await omni_adapter.connect(secret_key=token, display_name="RPi Bridge")
    except omni_adapter.AdapterError as err:
        print(f"[{err.code}] {err.hint or err.message}")
        print(omni_adapter.format_doctor_report(await omni_adapter.run_doctor(token)))
        raise SystemExit(1)
    try:
        ...                        # your program
    finally:
        await client.close()


asyncio.run(main())

For a bug report, build_report(credential) bundles the doctor run, /api/about and the adapter version (no credential — only a fingerprint). Its diagnostics list stays empty unless you pass the live client's ring: await build_report(token, diagnostics=client.diagnostics()). The CLI's omni-adapter report "$CONNECT_TOKEN" has no client, so its list is always empty.

Examples

.venv/bin/python examples/01_quickstart.py         "cst_…"   # first acked round-trip
.venv/bin/python examples/02_sensor_bridge.py      "cst_…"   # throttled sensor stream
.venv/bin/python examples/03_operations_console.py "cst_…"   # interactive ops shell

01/02/03 are the six-language common scenarios (quickstart · resilience · bidirectional+roster). Every other adapter ships the same three — declared in docs/parity.json (commonExamples) and kept honest by the CI parity job.

Testing

cd python && python -m venv .venv && .venv/bin/pip install -e '.[dev]'
.venv/bin/pytest              # unit tests, no server needed

The wire contract is pinned by tests/test_protocol_drift.py, which compares this package's constants against the sibling TypeScript adapter sources in the same repo (which in turn drift-checks against the server's protocol package).

Screens and remote input

Screen broadcast (examples/04_screen_broadcast.py)

pip install 'omni-adapter[stream]'
python examples/04_screen_broadcast.py "cst_..."

Renders a gauge with numpy and broadcasts it to the stage's Adapter Stream layer (adapter name python-gauge). Send gauge.value (0–100) from a prototype to move it; the adapter publishes gauge.viewers back.

omni_adapter.start_screen_broadcast owns the signalling and has no dependencies — it announces the adapter, answers each viewer's offer and cleans up on leave. The media engine is injected: the stream extra pulls aiortc, and users who only exchange values never install it.

Connect with provides_view="stream": on a server with adapter stream layers (unreleased) the server then creates and binds the Adapter Stream layer itself when the adapter joins — in the first group placed on the stage, matched by the name the server gave the adapter (client.me.display_name, which may read name (2)), and a deleted layer comes back on the next join. Without the declaration no layer is created and the helper records stream-undeclared once in client.diagnostics(). When the session factory returns None (nothing to send yet), the viewer gets no answer; call broadcast.source_changed() once the source exists so that viewer offers again — source_changed() is for when the source starts or changes, not per frame: calls are coalesced to at most one announcement per second, and it is safe from any thread. await client.ensure_stream_layer() is optional — an idempotent call that returns the layer: renamed is True when the server gave another name than requested (another program on a different key uses this name, or you rotated keys without revoking the old one — rename one, or revoke the old key; a layer under the original name will not show this adapter), and the client logs a warning. name_conflict is a deprecated alias of renamed. Errors: NotSupported (server too old), RoomEnded (no live session), TooManyRequests (429), ValidationError with the server's message, NotConnected (5xx, network, ack timeout, or a server failure while re-validating the key after a 401), SessionEnded after close(); any other error of that re-validation passes through as connect() would report it (for example AuthRevoked). The helper keeps one session per viewer receiver: the web keeps one receiver per (viewer page × adapter participant) and sends its receiverId, so two tabs of the same guest each get their own session instead of closing each other's — your factory is called once per receiver, with the same participant id, and viewer_count counts receivers. Answers and ICE echo the viewer's negotiationId and receiverId. A viewer keeps at most MAX_RECEIVERS_PER_VIEWER (4) sessions: a fifth receiver evicts first one whose session has not sent its answer yet (an answered session stays live while it renegotiates), then the least recently negotiated session — the helper cannot see an injected connection's state, so report closures, or a vanished tab (a PIN or share-link guest's refresh keeps the participant id) holds a slot and a live tab that never renegotiates can be the one evicted. To close a session as soon as its connection fails, pass it back — broadcast.session_closed(session) when aiortc's connectionState becomes "failed" or "closed" (safe from any thread; an unknown or already replaced session is ignored). When the adapter comes back with a new participant id (a rejoin into a new session, the 6-hour guest-token revalidation), the sessions from before the drop are closed. Viewer ICE that arrives early (before the offer, or before your session's answer() returns) is queued and handed to the session afterwards — candidates waiting for an offer expire after 10 s, and at most 4 receivers per viewer (64 in all) can wait. The ICE-server lookup gives up after 5 s and uses the last known servers; a failed lookup is never cached, so the next viewer asks again.

Roadside sign (examples/05_roadside_sign.py) — streamed and operable

pip install 'omni-adapter[stream]'
python examples/05_roadside_sign.py "cst_..."

A sign broadcast to an Adapter Stream layer (python-roadside) that an editor presses on the stage: speed limit, drive mode, day/night. Each press lands on channels the other examples already listen to — unity.speedLimit and unity.driveMode (the Unity car), hmi.ambient and hmi.notice (the AAOS cluster) — so four adapters share one stage with no glue between them.

It also acts with nobody pressing: it watches carTelemetry_speed and sends one hmi.notice when the car runs over the posted limit. That is what a service adapter is for — no UI of its own, never asleep.

Frames are raw numpy, so the sign carries a 5×7 bitmap font (~40 lines, no new dependency) — a numpy array has no text renderer, and a sign that cannot be read is not a sign. The TypeScript mirror (examples/13-driver-console.ts) uses fillText on a canvas instead.

It also serves a local preview you can click at http://127.0.0.1:4184/ (PREVIEW_PORT=0 disables it): an MJPEG stream — aiohttp is already a core dependency and the JPEG encoding rides on PyAV, so this costs no new install. Local clicks are normalized and fed through the same hit-test path as stage pointers, so whatever works locally works on the stage and vice versa.

Remote input (standalone servers)

An Editor on the stage can drive a broadcast screen with their pointer. Remote input runs on standalone servers — the only servers that run adapters — and needs no registration code: the stage arms it for Editors as soon as the screen is streaming, so the adapter only has to accept it — declare a 4th session-factory parameter, hand it the incoming data channel, and pass on_remote_input:

async def open_session(viewer, send_ice, ice_servers, wire_input):
    pc = RTCPeerConnection(...)
    pc.on("datachannel", wire_input)   # aiortc channel works as-is
    ...

broadcast = await omni_adapter.start_screen_broadcast(
    client, create_session=open_session,
    on_remote_input=lambda msg, viewer: inject(msg))

The helper whitelists directions, clamps coordinates to [0,1], and synthesizes cancel for open pointers when the channel closes. Injection is your code. Standalone servers send lease-holder notices (key adapters are registered automatically on join and told the holder): input is accepted only from the Editor holding the lease — the Connect editor claims it automatically on the first pointer-down — and PIN or share-link viewers never control. Older servers that predate the notices never announce a holder; then the gate stays permissive and accepts every viewer's input. No code change is needed for either. The registration API — await client.submit_registration_code(code) (returns a registration_id and a device_secret, which is a secret), await client.resume_registration(device_secret), client.on_remote_input_holder(listener) — is for registration-code deployments. A server without the adapter remote-input surface never opens the channel (doctor's remote-input line says which you have).

on_remote_input may be a plain function or an async def. Frames are handed to it in arrival order; an async callback starts in that order but interleaves with the next frame's callback after its first await (same as channel handlers). An exception in it is logged and never reaches the data channel — the next frame still arrives.

Testing without a server — MemoryStage

from omni_adapter import MemoryStage

stage = MemoryStage()
sensor = await stage.join(display_name="sensor", sends=["sensor.temperature"])
panel = await stage.join(display_name="panel", receives=["sensor.*"])
panel.on("sensor.*", lambda m: print(m.value))
await sensor.send("sensor.temperature", 23.5, ack=True)   # panel receives it; sensor does not (no echo)

stage.inject("ui.tap", 1)            # as if a prototype sent it
stage.disconnect(panel); stage.reconnect(panel)   # drop + recover: history snapshot, queue flush
stage.end("host closed the stage")   # everyone gets `ended`

Emulates the wire rules in-process (no self-echo, strict delivery, history ring, exclusive_name standby, listeners()); reject_publish= injects failures. A harness, not a server — no doctor, no REST (ensure_stream_layer() keeps one in-memory layer per name, memory-layer:<name>, so the screen-layer path can be unit-tested — and, like the server, refuses a member that did not join with provides_view="stream"), no signaling, and no 60-messages-a-second pacing (its transport sets bypass_rate_limit).

Testing without hardware — replay · simulate · record

from omni_adapter import parse_recording, replay, simulate

# $ omni-adapter record "cst_…" 'sensor.*' --out run.ndjson --timeout 60
records = parse_recording(open("run.ndjson", encoding="utf-8").read())
await replay(client, records, speed=2)                                   # gaps ÷ 2; math.inf = no gaps
await simulate(client, "sensor.temperature", waveform="sine", min=18, max=26, hz=2, duration_ms=30_000)
# $ omni-adapter simulate "cst_…" sensor.temperature --sine 18..26 --hz 2 --seconds 30

Both publish through client.send() (rate limit, queue, standby and backoff unchanged), wait for the queue to drain (giving up after 15 s without progress) before returning, stop on ended or a set stop event (checked before each send and during a sleep), and return the partial count. The CLI's replay/simulate exit 1 if something is still queued after that wait. sleep=/random= are injectable for deterministic tests. They only pace the sends — the 15 s no-progress limit runs on the real clock, so a no-op wait does not make it give up early.

History is state, not triggers

client.history()/client.latest() are re-seeded from the server snapshot on every (re)join whose joined payload carries one (the stage:joined fallback re-seeds nothing) and the history event fires — but the snapshot is never replayed into client.on() handlers. A command published while this adapter was disconnected is lost as a trigger and survives only as a last value: read latest() when stateChange reaches "connected" for state channels; treat trigger channels as at-most-once across a disconnect. Deliberate — replaying into handlers would re-fire a stale command minutes late. latest() remembers the 1000 most recently updated channels (like the history ring, older ones fall out; a stale out-of-order value does not count as an update).