Metadata-Version: 2.5
Name: omni-adapter
Version: 0.1.0b1
Summary: Omni Adapter for Python — join a Connect Omni stage from scripts, services and hardware bridges, and exchange messages with prototypes in real time.
Project-URL: Homepage, https://github.com/ProtoPie/omni-adapter/tree/main/python#readme
Project-URL: Repository, https://github.com/ProtoPie/omni-adapter
Project-URL: Changelog, https://github.com/ProtoPie/omni-adapter/blob/main/python/CHANGELOG.md
Project-URL: Issues, https://github.com/ProtoPie/omni-adapter/issues
Author: Studio XID, Inc.
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Keywords: adapter,asyncio,connect-omni,omni-adapter,protopie,prototyping,realtime,socket.io
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: aiohttp<4,>=3.14.4
Requires-Dist: python-engineio<5,>=4.13.2
Requires-Dist: python-socketio<6,>=5.16.2
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=1.3.0; extra == 'dev'
Requires-Dist: pytest>=9.0.3; extra == 'dev'
Provides-Extra: stream
Requires-Dist: aiortc>=1.9; extra == 'stream'
Requires-Dist: numpy>=1.24; extra == 'stream'
Description-Content-Type: text/markdown

# 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?** [Connect your first program to a stage](https://github.com/ProtoPie/omni-adapter/blob/main/docs/getting-started.md) walks
through the adapter key, the install and the adapter name step by step.

**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'))"` — and
sits at the repo root ([`llms.txt`](https://github.com/ProtoPie/omni-adapter/blob/main/llms.txt));
the Claude Code skill installs with `/plugin marketplace add ProtoPie/omni-adapter` then
`/plugin install omni-adapter@studioxid` (or copy
[`skills/using-omni-adapter/`](https://github.com/ProtoPie/omni-adapter/tree/main/skills/using-omni-adapter) into `.claude/skills/`). 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(...))`.

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

```bash
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 repository instead (contributors, or a change that has not
reached PyPI yet)? `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](https://github.com/ProtoPie/omni-adapter/blob/main/docs/COMPATIBILITY.md).
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.

```python
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:

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

```python
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:

```python
@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 `ValidationError`s,
some `ConfigError`s), 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:

```python
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", …)`.

| Code | Meaning | adapter behaviour |
| --- | --- | --- |
| `AuthRevoked` | Adapter key rejected, revoked, or expired | Terminal — auto-rejoin gives up (`ended: "rejoin-abandoned: AuthRevoked"`); issue a new key |
| `ProtocolVersionMismatch` / `ClientAheadOfServer` | Wire version gate | Terminal — upgrade one side; rejoin gives up rather than retrying forever |
| `ValidationError` | Bad channel/value locally (including a value `send_json` cannot encode, or a bad value pushed through `throttle()`), or the server rejected the join body (400) | Local checks fail before any network call; a join 400 names the fields the server rejected (see below) |
| `RoomEnded` | No live session: nobody with Editor/Owner access has the stage open in Connect (people who joined from a share link cannot start one), and the key was not created with "Let adapters start a session" (Share → Adapter keys) | `connect()` raises it at once; after a successful join, auto-rejoin (on by default) **keeps waiting** and joins when the session opens |
| `SessionEnded` | The client has ended (closed, or the session is over for good) | Sends raise; queued acked sends raise |
| `StageFull` | Participant cap reached | Surfaced from the join; rejoin keeps waiting — a seat may free up |
| `StageArchived` | The stage is archived | Surfaced from the join — ask its owner to restore it |
| `TooManyRequests` | Server rate limit (60 messages a second per connection), seen only on acked sends | Automatic backoff (1s → ×2 → 60s) and retry from the offline queue — not raised |
| `TooManyRequests` from `connect()` | The server refused the join (429): too many failed join attempts from this network address — 20 within 5 minutes, usually another program here still using a revoked or mistyped token — or more than 60 joins a minute with one adapter key (a restart loop); servers without the per-key join limit also refuse a sixth join from one address within a minute | Raised from `connect()`, not retried — wait up to 5 minutes, and fix the program that causes it |
| `KeyMaterialUnavailable` from `connect()` | The server (503) cannot read the stored material of an older adapter key on this stage — a server-side key change left it behind; after 30 of these from one address within a minute, a correct key gets the same answer until the minute ends | Raised from `connect()`, not retried (a joined client keeps retrying while it rejoins) — ask whoever runs your Connect to revoke the old keys |
| `AckTimeout` | No server ack within `ack_timeout_s` (15 s), or the socket closed with the ack pending | Raised to the caller of an acked send (at once when the socket closes) — the server may still have it |
| `QueueOverflow` | Queue cap hit (default 1000 — sending faster than 60/s, or a long disconnect) | Oldest item dropped and rejected, plus `queueOverflow` event |
| `NotConnected` / `JoinTimeout` / `ConfigError` | Client side: unreachable server (or `members()`/`listeners()` while not connected) · connected but never admitted · bad options | Surfaced to the caller — `ConfigError` before any connection is attempted |
| `Unauthorized` / `NotSupported` | The server refused a roster request (e.g. this connection was already marked as left) · the call is not available on this transport or server (`ensure_stream_layer()` on a server without the route — its other errors are `RoomEnded`, `TooManyRequests`, `ValidationError`, `NotConnected` and `SessionEnded` after `close()`; an error of its 401 re-validation passes through as `connect()` would report it, for example `AuthRevoked`) | Raised from that call |
| `Standby` | An acked send while another instance with your `display_name` is active, or a queued acked send discarded when this instance became standby | Raises instead of resolving — an acked send must not claim the server took something that was never sent. Un-acked sends are dropped with a debug log |

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:

```text
[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`.

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

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

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

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

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

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

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

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

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

```python
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).
