# RGB.live — real-time video sessions for robots, cameras and models

> Written for coding agents and developers. Read it top to bottom once; everything needed to build
> against RGB is here. Spec status: DRAFT 0.2 (2026-10-30). "Status" at the end separates what is
> implemented today from what is proposed.

A **session is a graph**: participants (cameras, robots, viewers, programs) and models, connected
by tracks. Creating a session admits the first version of the graph; `attach` admits a change;
`detach` removes one. There are no fixed pipelines: a model can be attached, re-pointed or removed
while the session runs, and the same camera can feed several models. Every attach is an admission
decision made by the platform at that moment. When there is no room, the attach fails and the
session keeps running with what it had. One model failing fails that attachment, not the session.

Everything travels as typed **tracks** over one transport per session, through a relay. Nothing
needs an inbound port. The relay never looks inside a track.

Install (packages are not yet published; see Status):

    pip install rgb-live            # Python: import rgb
    npm i @rgb-live/sdk             # JavaScript/TypeScript: import * as rgb from '@rgb-live/sdk'

## 1. Concepts (six words)

- **Session**: a graph plus a clock, held as state by the control plane. Created by your backend
  with a duration; everything in it is addressed under its id.
- **Participant**: anything that joins. A camera, a robot, a browser, a Python script. A participant
  joins with a role (`camera`, `viewer`, `robot`, `controller`), what it publishes and what it
  subscribes to, and receives an **access** credential it uses to connect.
- **Model**: a catalog entry (`sam3-detector-v1`, `qwen-monitor-v1`) attached to a participant's
  video at runtime, under a name you choose (`as: 'objects'`). The platform admits it onto a node
  with capacity, or refuses. Models publish results as tracks.
- **Track**: a named stream inside a node's broadcast. Three kinds: **video** (encoded access units
  plus a *track description*; Appendix A), **JSON stream** (one record per message, lossless, ordered),
  **JSON snapshot** (latest value). Every frame and every record carries a **timestamp on the
  session clock**, so a result always names the exact source frame it refers to.
- **Watch**: a subscription to any track from any participant in any language.
- **Transport**: how a session's bytes move. One transport per session (section 8). The API above
  the transport is the same.

The rule that makes this an API rather than a demo: a new kind of participant, model or watcher
needs no new infrastructure and no deployment. It needs a name, the tracks it reads, and the
tracks it writes.

## 2. Addressing

Every node is a path under the session. Clients never see the relay's internals; the SDK maps these.

    {session}/{participant}                 video track + its description, published by the participant
    {session}/{participant}/{as}            results of the model attached under that name
    {session}/link                          link state (demo 4), published by the publisher that owns the link

Names: `[a-z0-9][a-z0-9-]{0,31}`. A participant and an attachment name together identify an
attachment: `cam0/objects`. In `watch('cam0/objects')` that path means the attachment's primary
output; `watch('cam0/objects:checks')` names another output of the same attachment.

The session graph itself (participants, attachments, their admission state, revision) lives in the
control plane and is read through the API (section 7); it is not a relay track.

## 3. Python SDK

```python
import rgb                                                    # RGB_API_KEY and RGB_ORIGIN from the environment

session = rgb.sessions.create(name="warehouse-demo", duration_seconds=3600)   # backend: idempotent by name

# participants
cam = session.add_source(rgb.FileSource("clip.mp4", id="cam0", loop=True))   # join as a camera and publish from this process
cam = session.source("web0")                                                # refer to a camera that joined elsewhere
cam = session.add_source(rgb.RtspSource("rtsp://10.0.0.5/stream1", id="dock"))  # proposed; IP camera, no re-encode

# models: attach a catalog model to a camera, under a name. Admitted now, or refused now.
objects = session.attach("sam3-detector-v1", to=cam, as_="objects", params={"prompts": ["person", "pallet"]})
alerts  = session.attach(rgb.Monitor(rules=["a person entered the robot work zone"]), to=cam, as_="alerts")  # typed sugar
objects.configure(prompts=["person", "forklift"])             # re-admits the attachment with new params; tracking restarts

try:
    depth = session.attach("depth-anything-v2", to=cam, as_="depth")
except rgb.AttachRefused as e:                                # no node has room: the session keeps running
    print(e.reason)                                           # "no_capacity: 0 of 4 detector slots free on pod-1"

# watching
@objects.watch
def on_detections(rec): ...                                   # one record per processed frame
@alerts.watch
def on_event(e): ...                                          # sparse, confirmed events only
alerts.outputs.checks.watch(print)                            # any named output, typed
session.watch("cam0/alerts:checks", print)                    # any output by path, untyped

session.run()                                                 # blocking; or `async with session:`
```

`rgb.sessions.create(name=None, duration_seconds=86400, transport="moq")` creates the session in the
control plane and returns a handle. With `name`, create is idempotent: a live session of that name
is returned instead of a new one (the same rule Gridcaster applies to creation keys). `rgb.sessions.get(id)`,
`.list()`, `.close(id)`. `rgb.Client(api_key=..., origin=...)` builds a client explicitly;
`rgb.sessions` is the default client built from `RGB_API_KEY` and `RGB_ORIGIN`.

`session.join(as_="camera", id="cam0", publish=["video"], subscribe=[])` registers a participant and
returns its **access**: `{session_id, participant_id, token, relay, transport, expires_at}`. A device
process never holds the API key; it gets an access from the backend and calls `rgb.connect(access)`,
which returns the same `Session` handle limited to publishing and watching. `add_source` is sugar
for join-as-camera plus publish-from-here; `source(id)` only refers.

`attach(model, to=, as_=, params=)` POSTs the attachment to the control plane and returns a `Model`
handle whose `.attachment` carries the admission result (`state`, `node`, `rev`). It raises
`AttachRefused` with a reason when no node can serve it. `detach(model)` removes it. The model's
`params` are the catalog model's config (section 6); the typed classes `Detector`, `Monitor`, `Agent`
fill in the catalog id and validate params.

Sources you publish from Python expose runtime controls: `cam.set_bitrate(bps)`, `cam.keyframe()`,
and with `hd_ring_s` set, `await cam.backfill(seconds=6, to_ts_us=..., link=link)`.
`session.link(sources, budget_bps=...)` shares one budget between sources (demo 4).

## 4. JavaScript SDK

Same verbs. In a backend, the SDK creates sessions and attaches models with the API key. In a
browser, it connects with an access and is a viewer, a controller and a camera at once. The demo
pages do both from the browser because they are demos.

```js
import * as rgb from '@rgb-live/sdk';

const client = rgb.client({ apiKey: KEY, origin: 'https://rgb.example' });
const session = await client.sessions.create({ name: 'warehouse-demo', durationSeconds: 3600 });
const cam = await session.join({ as: 'camera', id: 'web0', publish: ['video'] });
const objects = await session.attach('sam3-detector-v1', { to: 'web0', as: 'objects', params: { prompts: ['person'] } });

// browser (or the same page)
await session.connect(cam.access, { publish: { camera: await navigator.mediaDevices.getUserMedia({ video: true }) } });
objects.watch((rec) => drawBoxes(rec));                   // data, not burned-in pixels
session.watch('web0/objects', drawBoxes);                 // the same, by path
await session.watchVideo('web0', (frame) => ctx.drawImage(frame, 0, 0));   // decoded VideoFrame
```

`session.attach(new rgb.Detector({ prompts }), { to, as })` is the typed form. `attach` rejects with
`rgb.AttachRefused` when there is no capacity. `model.outputs.<name>.watch(cb)` subscribes to any
named output; `session.watch(path, cb)` to any by path. `session.graph()` returns the current graph
(section 7) and `session.watchGraph(cb)` follows it.

## 5. The four demos, as code

Each demo is the same session model with different participants and attachments. Read sections 3
and 4 first for the verbs. Lines marked `# proposed` use API that is designed but not yet implemented.

**Demo 1: open-vocabulary detection.** Type what to look for; boxes and track IDs come back as data.

```python
session = rgb.sessions.create(name="warehouse-demo")
cam = session.add_source(rgb.FileSource("clips/unitree_warehouse_60s.mp4", id="cam0", loop=True))
objects = session.attach("sam3-detector-v1", to=cam, as_="objects", params={"prompts": ["spray can", "robot dog", "hand"]})
@objects.watch                                                   # one record per processed frame
def on_detections(rec):
    print(rec["frame_ts"], [(t["label"], t["track_id"]) for t in rec["tracks"]])

session.run()
```
```js
const session = await client.sessions.create({ name: 'warehouse-demo' });
const objects = await session.attach('sam3-detector-v1', { to: 'cam0', as: 'objects', params: { prompts: ['person'] } });
await session.connect();                                         // viewer
objects.watch(drawBoxes);                                        // data on a canvas, not pixels in the video
await session.watchVideo('cam0', (frame) => ctx.drawImage(frame, 0, 0));
```

**Demo 2: natural-language alerts.** Watch for an incident described in plain language. A cheap
detector gates every frame; VLMs confirm and explain before anything is raised.

```python
cam = session.source("cam0")                                     # published by someone else
alerts = session.attach("qwen-monitor-v1", to=cam, as_="alerts",
                        params={"rules": ["a fire broke out", "a person is in the robot's path"]})
@alerts.watch                                                    # confirmed events only
def on_event(e):
    page_operator(f"{e['camera_id']}: {e['rule']} — {e['reason']}")

alerts.outputs.checks.watch(log)                                 # every VLM call, for the timeline
session.run()
```
```js
const alerts = await session.attach('qwen-monitor-v1', { to: 'cam0', as: 'alerts', params: { rules: ['a fire broke out', "a person is in the robot's path"] } });
alerts.watch((e) => showAlert(`${e.rule} — ${e.reason}`));        // confirmed events only
alerts.outputs.checks.watch((c) => timeline.add(c));             // every VLM call, for the timeline
```
When a fire breaks out, the detector (watching for "fire", "smoke", derived from the rule) wakes the
monitor within a second, the screen and verify steps take about 2 s, and one event per cooldown is
emitted with the frames it was judged on in `clip_ref`.

The dynamic graph in one example: attach a second model only when the first one sees something.

```python
@objects.watch
def on_detections(rec):
    if any(t["label"] == "person" for t in rec["tracks"]) and "cam0/depth" not in session.graph()["attachments"]:
        session.attach("depth-anything-v2", to=cam, as_="depth")          # admitted now, on a node with room
        session.record(pre_roll_seconds=10, seconds=30)                   # proposed
```

**Demo 3: an agent drives a world, a person can take over.** A simulator (MuJoCo today; Cosmos or any world
model later) joins as a participant that publishes video and reads a `commands` input. An Agent attached to it
looks at the frame and writes commands from a plain-language instruction; a viewer watches the same video and
publishes the same input to take the wheel. The world decides who wins (the operator while their commands are fresh).

```python
world = session.add_source(rgb.Node("world", publishes=["video"], subscribes=["commands"]))
pilot = session.attach("qwen-agent-v1", to=world, as_="pilot", params={"prompt": "drive to the red crate and push it", "rate_hz": 2})

@world.inputs.commands.watch                                     # (publisher, record): the agent's or a person's
def on_command(who, cmd): ...
state = await world.output("state", snapshot=True)               # anything the world wants to say, as a JSON snapshot
world.write(frame_rgb, timestamp_us)                             # one frame in, H.264 out
```
```js
await session.watchVideo('world', draw);                         // watch what the pilot sees
session.publish('world', 'commands', { throttle: 1, steer: 0, role: 'operator' });   // a person takes the wheel
session.watchOutput('world', 'state', showWhoIsDriving, { snapshot: true });
pilot.watch((rec) => timeline.add(rec.command.reason));          // the agent explains each decision
```
Inputs live at `{session}/{node}/{input}/{publisher}` (one broadcast per publisher, so an agent and a person never
collide); the node sees every publisher and applies its own rule. The agent publishes to the world's input and to
its own output track `{session}/world/pilot` (`commands`, one record per decision with the frame it looked at,
the command and the reason).

**Recording.** `session.record(to, as_="rec", pre_roll_s=10, seconds=30, tracks=["objects:detections"])` is an
attachment of `record-v1`, admitted onto the recorder node like any model. `to` is a participant (`"cam0"`) or a
model's video output (`"cam0/depth"`). The recorder subscribes the Hang track with the relay's retention window (that
is the pre-roll), writes the encoded frames into an MP4 as they arrive (Mediabunny; no decode, no re-encode;
timestamps kept on the session clock), writes each listed JSON track as JSONL, uploads, and reports to the registry.
Detach to stop early; with `seconds` it detaches itself. Playback is a plain `<video src=url>`; replaying a
recording into a session as a participant is not built.

```python
rec = session.record("cam0", pre_roll_s=10, seconds=30, tracks=["objects:detections"])
session.recordings()      # [{id, url, status, frames, first_ts, last_ts, duration_s, codec, width, height, files}]
```

**Clips: the stream is held, a clip is a verb on it.** `cam.clip(before=5, after=10)` pipes the last 5 s and the next
10 s of that stream to a file now (the pre-roll from the relay's window); `cam.clip(on=alerts, before=5, after=10,
notify={...})` arms it: the recorder keeps the last `before` seconds of the stream (and of the listed JSON tracks) in a
ring buffer, and every record on the `alerts` model's events track (`"alerts:checks"` names another track) becomes a
clip, extended while events keep coming (120 s cap), uploaded, registered with its triggering events (`events`,
`event_ts`), then `notify` is called: `webhook` gets a JSON POST (`{type: "rgb.event", session, camera, rule, reason,
event_ts, events, recording: {id, url, duration_s, ...}}`), `slack` an incoming-webhook message with the reason and a
link to the clip, `ntfy` a push to a topic (or full URL) with the clip attached. An armed clip is a `record-v1`
attachment whose `status` says `armed` (with the seconds buffered and the clips made); it shows in the graph as a badge
on the stream, not as a node, and while a clip is being written the status is `recording` and the graph shows the clip.
`buffer="relay"` subscribes nothing until the event and takes the pre-roll from the relay's window instead; that window
depends on the track's other subscribers (measured 20 to 30 s with only browsers watching, 2 s once a Python-bindings
subscriber such as a model node is on the track), so it is opt-in until the relay's retention is understood.
Notification targets live in the attachment's params, so anyone who can read the session graph can read them: use
per-session webhooks you can rotate.

```python
alerts = session.attach("qwen-monitor-v1", to="cam0", as_="alerts", params={"rules": ["a person fell"]})
cam = session.source("cam0")
cam.clip(on=alerts, before=5, after=10, notify={"slack": SLACK_WEBHOOK, "ntfy": "rgb-site-7"})
```

**Demo 4: model-aware transport.** Four cameras share a 1.5 Mbps link; a rule firing on one camera
moves the bandwidth to it, forces a keyframe, and backfills the HD pre-roll behind the live stream.
This is the demo that needs the transport to be MoQ (section 8).

```python
cams = [session.add_source(rgb.FileSource(f"clips/cam{i}.mp4", id=f"cam{i}", hd_ring_s=15)) for i in range(4)]
link = session.link(cams, budget_bps=1_500_000, mode="rgb", share=0.8, hold_s=20)   # one budget, enforced at the encoders
alerts = [session.attach("qwen-monitor-v1", to=c, as_="alerts", params={"rules": ["a person entered the work zone"]}) for c in cams]

@session.watch_all(alerts)
def on_event(e):
    hot = session.sources[e["camera_id"]]
    if link.prioritize(hot, reason=e["rule"]):                               # 80 % to hot, keyframe cut, hysteresis inside
        session.spawn(hot.backfill(seconds=6, to_ts_us=e["event_ts"], link=link))   # HD pre-roll as its own track, paced out of the budget
session.run()
```
```js
session.watchLink((st) => drawBars(st.bitrates, st.hot, st.backfills));   // who has the bandwidth, right now
await session.prioritize('cam2', { reason: 'operator' });                  // the same trigger from a page
await session.setLink({ budget_bps: 800_000, mode: 'even' });             // the slider and the comparison toggle
```
The link publishes its state at `{session}/link` (track `state`): budget, mode, hot source and remaining
hold, per-source bitrates, backfills in flight, recent switches. `mode: "even"` is the comparison: the
same cameras, the budget split equally, no reaction to events. Link commands go through the control
plane (`PUT /sessions/{id}/link`) and reach the publisher through the session graph.

## 6. Catalog models and their records

`GET /catalog` lists what can be attached. Each entry has an id, a kind, the params it accepts, the
outputs it publishes and the slots it costs on a node. Today:

| id | kind | params | outputs (primary first) | slots |
|---|---|---|---|---|
| `sam3-detector-v1` | detector | `prompts`, `threshold`, `stride`, `image_size`, `masks` | `detections` | 1 detector |
| `qwen-monitor-v1` | monitor | `rules`, `cooldown_s`, `heartbeat_s`, `min_confidence` | `events`, `checks` | 1 monitor (+1 detector if the camera has none) |
| `qwen-agent-v1` | agent | `prompt`, `rate_hz`, `output` (input name), `schema`, `system` | `commands` | 1 agent |
| `record-v1` | record | `pre_roll_s`, `seconds`, `tracks`, `on`, `notify`, `buffer` | an MP4 (+ JSONL per track) in the recordings registry; with `on`, one per event, then a webhook / Slack / ntfy notification (`cam.clip(...)` is the sugar) | 1 record (the recorder node) |
| `rtmpose-v1` | pose | `stride`, `max_people` | `pose` | 1 pose (Apple silicon node) |
| `depth-anything-v2-small` | depth | `stride`, `smoothing` | **video** at `{to}/{as}.hang` + `stats` | 1 depth (Apple silicon node) |
| `upscale-vdsr-v1` | upscale | `stride`, `bitrate` | **video** at 2x + `stats` | 1 upscale (Apple silicon node) |

A model output can be **video**. It lives at `{session}/{to}/{as}.hang`, the same shape as a participant's own
video, so `watchVideo('cam0/depth')`, the recorder and the relay treat it like any camera. Every output frame
carries the timestamp of the input frame it was computed from. Depth is 8-bit grayscale (1..99 percentile of
relative inverse depth to 0..255; the `stats` track carries `min`, `max` per frame); the upscale is RGB at 2x.

**Pose** record (`pose`): `{frame_ts, camera_id, model, latency_ms, size, people: [{id, score, box, keypoints: [[x, y, score] x 17 COCO]}]}`.
Measured on an M6 Mac mini at 640x360: pose 18 ms, depth 16 ms, upscale 4.5 ms per frame (each alone), all three
at once at about 20 fps each.

All records are JSON. `*_ts` fields are integer microseconds on the session clock, equal to the
timestamp of the video frame they describe. Coordinates are pixels in the source frame; `size` gives
the frame size.

**Detector** (open-vocabulary detection + tracking; SAM 3). Params: `prompts: [str]`, `threshold`
(0..1, default 0.85), `stride` (run every Nth frame, default 5), `masks` (bool). Track `detections`:

    {"frame_ts": 11802160000, "camera_id": "cam0", "model": "objects", "latency_ms": 148,
     "size": [640, 360], "prompts": ["spray can", "hand"],
     "tracks": [{"track_id": 2, "label": "hand", "score": 0.93, "tracker_score": 1.0,
                 "box": [496, 134, 599, 245], "mask_rle": {...}}]}

Track IDs persist while an object stays visible; an object that leaves and returns gets a new ID.
Changing `prompts` re-admits the attachment and starts a new tracking session (IDs restart). About 4
to 8 records per second per camera at `stride` 5; `latency_ms` is the model's time on that frame.

**Monitor** (plain-language rules, verified by a VLM cascade). Params: `rules: [str]`,
`cooldown_s` (default 15), `heartbeat_s` (default 1: the re-check cadence when nothing changes), `motion` (a watched
object whose box moves more than this fraction of its size triggers a check; default 0.25, 0 turns it off), `min_confidence` (default 0.6).
Track `events` (confirmed only):

    {"event_ts": 7004400000, "camera_id": "cam0", "rule": "an object is placed on the robot dog",
     "verdict": true, "confidence": 0.95, "reason": "A hand places a spray can onto the top of the robot dog.",
     "trigger": "detector: appeared=['spray can']", "clip_ref": {"source": "cam0", "from_ts": 7001400000, "to_ts": 7004400000},
     "model": "Qwen/Qwen3.5-9B", "latency_ms": 1258}

Track `checks` (every VLM call, including negatives, with `"stage": "screen" | "verify"`).
How it works: the runtime derives detector prompts from each rule; when those objects appear,
vanish, change count or move (or on the heartbeat), a small VLM screens the rule on the last 3 s of
frames and a larger VLM confirms and explains. One event per rule per cooldown.

**Agent** (a model that acts; Qwen3.5-4B). Params: `prompt` (the instruction), `rate_hz` (default 2), `output`
(the participant's input to write, default `commands`), `schema` (JSON schema of one command; the default is a
categorical rover command: reason, target_visible, target_position, steer, throttle, mapped to numbers), `system`.
Each step it looks at the newest frame of the participant's video and writes one command to
`{session}/{to}/{output}/{as}` and one record to its output track `commands`:

    {"frame_ts": 190600000, "camera_id": "world", "model": "pilot", "prompt": "drive to the red crate", "step": 25,
     "command": {"throttle": 0.6, "steer": -0.5, "reason": "The red crate is left of centre, ahead.", "target_position": "left", ...},
     "latency_ms": 575, "vlm": "Qwen/Qwen3.5-4B"}

About 0.6 to 0.9 s per decision on the 4B model. A person publishing the same input takes precedence by the
world's rule, not the platform's.

## 7. The control plane

Three layers, each owned by something that exists:

1. **Session graph** (control plane): sessions, participants, attachments, link commands, revisions.
   Create admits the first version; attach admits a change; detach removes one.
2. **Node router** (in the control plane): decides where each attachment runs. A node that already
   serves the participant's camera first (the frame is decoded once and shared); another node when
   the first is full or lacks the model; re-admitted elsewhere when a node dies. Capacity is counted
   per attachment, not per session.
3. **Nodes** (GPU runtimes): a node is a pool. It serves many cameras and many models in parallel,
   registers the catalog ids it serves and its slots, heartbeats, and follows its assignments.

The relay is the data plane between all of them. Nodes and participants only ever dial out to it.

REST, JSON, `Authorization: Bearer <api key>` (participants may use their access token for reads).
Base path `/api/v1`.

| Call | Does |
|---|---|
| `POST /sessions` `{name?, duration_seconds?, transport?}` | create (idempotent by `name`); returns the graph |
| `GET /sessions`, `GET /sessions/{id}`, `DELETE /sessions/{id}` | list, read, close |
| `GET /sessions/{id}?since={rev}&wait=25` | long-poll: returns when the graph's revision passes `since` |
| `POST /sessions/{id}/participants` `{id?, as, publish?, subscribe?}` | join; returns the access |
| `DELETE /sessions/{id}/participants/{pid}` | leave |

JS: `await session.leave()` removes this page's participant (pass `{ keepalive: true }` from `pagehide` so it completes during unload).
A browser viewer should also join with `ttl_s` (a lease) and re-join with the same id before it lapses; the control plane sweeps a
participant whose lease lapsed, so a closed tab disappears within the ttl even when `leave` never ran.
| `POST /sessions/{id}/attachments` `{model, to, as, params?}` | attach; `201` admitted, `409 no_capacity` refused, `404 unknown_model` |
| `DELETE /sessions/{id}/attachments/{to}/{as}` | detach |
| `PUT /sessions/{id}/link` `{budget_bps?, mode?, prioritize?}` | link commands (demo 4) |
| `GET /recordings[?session=]`, `GET /recordings/{id}`, `DELETE /recordings/{id}` | the recordings registry (workspace objects; the recorder node writes them). Delete marks the entry `deleting`; the recorder node removes the files (local and S3) and confirms, then the entry goes. SDKs: `client.deleteRecording(id)`, `rgb.sessions.delete_recording(id)` |
| `GET /catalog` | models, params, outputs, slots |
| `GET /nodes`, `POST /nodes` `{id, models, slots}`, `POST /nodes/{id}/heartbeat` | node registry (runtimes call these) |
| `GET /nodes/{id}/assignments?since={rev}&wait=25` | long-poll: what this node should be running, by session |

The graph document:

```json
{"id": "warehouse-demo", "rev": 7, "transport": "moq", "relay": "https://.../anon",
 "created_at": 1793400000.0, "expires_at": 1793403600.0,
 "participants": {"cam0": {"as": "camera", "publish": ["video"], "subscribe": [], "joined_at": 1793400001.0}},
 "attachments": {"cam0/objects": {"model": "sam3-detector-v1", "kind": "detector", "to": "cam0", "as": "objects",
                                   "params": {"prompts": ["person"]}, "state": "admitted", "node": "pod-1", "rev": 7}},
 "link": {"budget_bps": 1500000, "mode": "rgb"}}
```

Attachment states: `admitted` (placed on a node), `unplaced` (its node died; re-admitted when a node
with room appears), `refused` is never stored: a refused attach returns `409` and changes nothing.
Attaching to an existing `{to}/{as}` re-admits it with the new params (what `configure` does).
Sessions expire at `expires_at`; the control plane removes them and nodes stop their workers.

## 8. Transport

One transport per session, chosen at creation (`transport`), and the API above it does not change.

- `moq`: Media over QUIC. What these demos run and what the robot-to-GPU path uses when the link is
  the problem: many cameras to one GPU with no browser on the sending end, priority between streams
  on one uplink, HD backfill as a lower-priority track, and non-video data as first-class tracks on
  the same clock (demo 4 shows all four).
- `webrtc`: reserved. The intended default for browsers and existing WebRTC deployments (WHIP
  encoders, OpenTok-style apps). An implementation would put an SFU where the relay is and carry
  JSON tracks over data channels; demos 1 to 3 would be unchanged, demo 4's backfill and priority
  would need a different story.

What the SDKs promise regardless of transport: records carry our own microsecond timestamps, never
the transport's; addressing is by session and name, never by relay URL or room id; attach, detach
and link commands go through the control plane; keyframe and bitrate requests are SDK calls on the
source. The transport is a connect option (`rgb.connect(access, transport=...)`) that defaults to the
session's.

## 9. Custom nodes

A node is any process that publishes or subscribes tracks under the session. `rgb.Node` lets a simulator, a
world model or your own model join as a participant:

```python
world = session.add_source(rgb.Node("cosmos", publishes=["video"], subscribes=["commands"], fps=25))
world.inputs.commands.watch(lambda who, cmd: queue.put(cmd))   # JSON records from whoever publishes them
world.inputs.commands.latest                                   # {publisher: (received_at, record)}, for "who wins" rules
world.write(frame_rgb, timestamp_us=...)                       # raw frame in, H.264 out, description published for you
telemetry = await world.output("state", snapshot=True)         # a JSON track of its own at {session}/cosmos/state
await session.publish("cosmos", "commands", {...})             # any participant can write a node's input
```
`demos/demo5_world.py` is the reference: a MuJoCo scene, 25 fps, a chase camera, an operator-beats-agent rule.

## 10. Ownership and conflicts (partly proposed)

The graph has one writer, the control plane, and a revision that increases on every change; clients
read it with `since` and get the whole document, so there is no reconciliation on the client. What
is not built: per-client ownership of attachments (any key holder can detach any attachment),
`takeover`, scoped tokens (today a participant's access token is accepted for reads only and the one
API key does everything), and optimistic revisions on writes.

## 11. Batch jobs (proposed)

A job is not a session: inputs by reference, every frame processed exactly once in order, outputs as
files, and an end. Same models, same record schemas, same workers as live sessions. Calls build a
graph; nothing runs until `run()`.

```python
job = rgb.Job("run-42")
clip = job.asset("s3://bucket/run-42/cam0.mp4")                 # symbolic: nothing is fetched
det  = job.attach("sam3-detector-v1", to=clip, params={"prompts": ["person", "pallet"]})
mon  = job.attach("qwen-monitor-v1", to=clip, params={"rules": ["a fire broke out"]})
job.output(det.detections, to="s3://bucket/run-42/out/detections.jsonl")
job.output(mon.events,     to="s3://bucket/run-42/out/events.jsonl")

job.validate()      # optional and free: raises with a path-qualified error, or returns {models, gpu_seconds, cost}
job.run()           # validates (fails before anything is billed), runs; returns the output locations
# -> {"detections": "s3://bucket/run-42/out/detections.jsonl", "events": "s3://bucket/run-42/out/events.jsonl"}
```
```js
const job = new rgb.Job('run-42');
const clip = job.asset('s3://bucket/run-42/cam0.mp4');
const det = job.attach('sam3-detector-v1', { to: clip, params: { prompts: ['person', 'pallet'] } });
job.output(det.detections, { to: 's3://bucket/run-42/out/detections.jsonl' });
const outputs = await job.run();          // { detections: 's3://bucket/run-42/out/detections.jsonl' }
```

Outputs are one JSON record per line, the same records as the live tracks; the format follows the
file extension (`.jsonl`, later `.mcap`), and omitting `to` writes next to the inputs. Handles such as
`det.detections` are references to future outputs, not values: data-dependent logic is expressed
as nodes (a monitor rule, a filter), not as `if` statements in the building code. The graph a job
builds is a JSON document (Appendix C), which is what the CLI runs and what you can check in.

## 12. Timing facts you should design around

- Video latency source to viewer through the relay: about 200 to 400 ms.
- Attach to first record: about 1 to 2 s (admission is immediate; the node subscribes the video,
  waits for a keyframe and warms the tracker).
- Detector: 130 to 200 ms per processed frame on a 48 GB Ada-class GPU at 560 px; about 4 cameras
  per GPU at 5 fps each before adding VLMs. Records lag the live frame by that much; draw them on the
  frame whose timestamp they carry, or accept the lag.
- Monitor: screen about 0.6 s, verify about 1.2 s; events arrive 2 to 4 s after the moment.
- A source that stops publishing for more than 2 s should be treated as gone by watchers.

## 13. Failure modes

- `attach` raises `AttachRefused`: no node has a free slot for that model, or no node serves it.
  The session is unchanged. Detach something, or wait for a node, then retry.
- `attach` is admitted but no records arrive: the camera has not published video yet, or the node
  is still warming up. Read the graph: the attachment's `node` says who should be serving it.
- Records arrive with very old timestamps on connect: JSON streams replay recent history to a new
  subscriber; use the latest record and ignore the backlog.
- Boxes for an object that left the frame: should not happen (the runtime drops tracks whose
  presence score falls); if it does, lower `threshold` is not the fix, report it.
- The graph shows an attachment `unplaced`: its node stopped heartbeating. It is re-admitted on the
  next node with room; nothing else in the session changed.

## 14. Local development against this repo

The packages above are not published. In this repository the same APIs are the local modules:
Python `rgb/` (`pip install -e .` installs it as `rgb-live`), JavaScript `sdk/js/` (built to
`dashboard/lib/rgb-sdk.js`; the pages import it by relative path and map `@moq/net` with an import
map). The control plane is `control/server.py`; run it locally with a local relay, or point
`RGB_ORIGIN` at the hosted one (`INFRA.md`). The current implementation appends a container suffix
to video broadcast names on the wire; the SDKs hide it and it will go away.

## Status (2026-10-30)

Implemented and running: sessions (create by name, duration, close), participants (join, access), nodes on own machines
(the RTX 5090 with the CUDA backend; Mac minis with the Core ML backend: `rtmpose-v1`, `depth-anything-v2-small`,
`upscale-vdsr-v1`), video model outputs, custom nodes
(`rgb.Node` with inputs and outputs, `session.publish`), `qwen-agent-v1`,
attach with admission and refusal, detach, re-attach with new params, the node registry with slots
and heartbeats, re-admission when a node dies, the graph with revisions and long-poll, FileSource
(with HD ring and `backfill`), CameraSource, `source(id)`, `sam3-detector-v1`, `qwen-monitor-v1`,
`watch` by handle and by path, `outputs.<name>.watch`, `watch_all`, `watchVideo`, `session.link`
with `prioritize`, `set_budget`, `set_mode` and the `{session}/link` state track, the record schemas
in section 6, the timing facts in section 12, transport `moq`.
Implemented: recording (`record-v1`, the recorder node, the registry). Proposed (designed, not built): replay into a session, world models as nodes (Cosmos3; the Node API is ready for them),
ownership and scoped tokens, RtspSource, RosSource, transport `webrtc`, the WHIP gateway as a
managed node, H.265/AV1 baselines, batch jobs and the `rgb jobs` CLI, published packages and a
multi-node router beyond first-fit. Source: github.com/<org>/rgb-demos (private).

## Appendix A. Video: one representation, many sources

A session has exactly one internal form for video: **encoded access units** (H.264 today; H.265
and AV1 planned), a **track description**, and a **capture timestamp per frame on the session
clock**. Sources convert to it at the edge; decoders consume it without knowing the source type.
The relay never inspects it.

The track description is what any decoder needs and nothing more:

    {"codec": "avc1.42001f",          # WebCodecs-style codec string
     "width": 640, "height": 360, "framerate": 25,
     "description": "<hex>",          # optional out-of-band parameter sets; absent when they are in-band
     "clock": {"timescale": 1000000, "origin": "capture"}}   # microseconds on the session clock

Each transport carries the description in its own slot (a catalog next to the track, SDP, a
container header); the SDKs read and write it for you.

Sources, and who does the conversion:

| Source | Normalized by | What happens |
|---|---|---|
| `FileSource` (a clip played as a live camera) | Python SDK | decoded and re-encoded: paced to real time, no B-frames, keyframe every second |
| `CameraSource` (webcam in a browser) | JS SDK | WebCodecs encodes; description from the encoder |
| webcam or app publishing over WebRTC | gateway node (WHIP endpoint) | re-published without re-encoding; parameter sets from SDP and in-band |
| `RtspSource` (proposed; third-party IP camera) | ingest node | pulls RTSP, stream-copies H.264 packets, no re-encode; parameter sets from the RTSP SDP |
| `RosSource` (proposed; ROS 2 image topic) | ingest node on the robot | raw frames encoded in-process |
| `Node` (simulator, world model, your own code) | your process via the SDK | raw frames encoded in-process, or pre-encoded access units passed through |

Sources the SDK controls expose `set_bitrate(bps)`, `keyframe()` and `set_priority(p)`; third-party
cameras do not, and their own settings apply.

Where interop breaks, and what the SDK does about it:

- **Codec mismatch.** Many IP cameras default to H.265, and browser H.265 decoding is uneven. The
  session baseline is H.264; an ingest node transcodes anything else (CPU or GPU encoder cost).
- **Clocks.** Our own sources stamp capture time directly. RTSP gives 90 kHz RTP time and a wall-clock
  mapping only when RTCP sender reports arrive; without them the ingest node stamps arrival time and
  the "same clock as the results" guarantee loosens to tens of milliseconds. Timestamps are as good
  as the source's clock; the description says which case applies (`clock.origin`: `capture` or `arrival`).
- **Keyframe cadence.** Third-party cameras often send a keyframe every 2 to 5 s, so a late joiner or
  a quality switch waits that long. Sources we control default to 1 s and accept `keyframe()`.
- **Frame order.** Files may contain B-frames (decode order differs from display order); `FileSource`
  re-encodes without them. A live camera that sends B-frames (rare) needs reordering at the decoder.

## Appendix B. Types

A model declares its output tracks as part of its type; subscribing to any of them is one
operation. Inputs of a custom node are the same `Track` type seen from the other side.

```ts
type Unsubscribe = () => void;

interface Track<T> {
  readonly name: string;
  watch(cb: (rec: T) => void): Unsubscribe;
}

interface Attachment { model: string; kind: string; to: string; as: string; params: Record<string, unknown>; state: 'admitted' | 'unplaced'; node: string | null; rev: number }

interface Model<O extends Record<string, unknown>, P extends keyof O, C = unknown> {
  readonly kind: string;
  readonly modelId: string;                           // catalog id, e.g. 'sam3-detector-v1'
  readonly name: string;                              // the `as` name
  readonly attachment: Attachment | null;             // admission result, once attached
  readonly outputs: { [K in keyof O]: Track<O[K]> };
  watch(cb: (rec: O[P]) => void): Unsubscribe;        // sugar for outputs[primary].watch
  configure(changes: Partial<C>): Promise<this>;      // re-admits with new params
}

interface Box { x1: number; y1: number; x2: number; y2: number }   // records carry [x1, y1, x2, y2]
interface TrackedObject { track_id: number; label: string; score: number; tracker_score?: number; box: [number, number, number, number]; mask_rle?: unknown }
interface DetectionRecord { frame_ts: number; camera_id: string; model: string; latency_ms: number; size: [number, number]; prompts: string[]; tracks: TrackedObject[] }
interface EventRecord { event_ts: number; camera_id: string; rule: string; verdict: true; confidence: number; reason: string; trigger: string; clip_ref: { source: string; from_ts: number; to_ts: number }; model: string; latency_ms: number }
interface CheckRecord extends Omit<EventRecord, 'event_ts' | 'verdict' | 'clip_ref'> { frame_ts: number; stage: 'screen' | 'verify'; verdict: boolean }

interface DetectorConfig { prompts: string[]; threshold?: number; stride?: number; image_size?: 560 | 1008; masks?: boolean }
interface MonitorConfig  { rules: string[]; cooldown_s?: number; heartbeat_s?: number; min_confidence?: number }

declare class Detector implements Model<{ detections: DetectionRecord }, 'detections', DetectorConfig> { constructor(config: DetectorConfig) }   // sam3-detector-v1
declare class Monitor  implements Model<{ events: EventRecord; checks: CheckRecord }, 'events', MonitorConfig> { constructor(config: MonitorConfig) }  // qwen-monitor-v1
declare class Agent    implements Model<{ commands: unknown }, 'commands'> {}                    // proposed; output schema is config

interface Source { readonly id: string; readonly path: string }
interface ControlledSource extends Source { setBitrate(bps: number): void; keyframe(): void; setPriority(p: number): void }

interface Access { session_id: string; participant_id: string; as: string; token: string; relay: string; transport: 'moq' | 'webrtc'; expires_at: number }

interface Graph { id: string; rev: number; transport: string; relay: string; expires_at: number;
  participants: Record<string, { as: string; publish: string[]; subscribe: string[] }>;
  attachments: Record<string, Attachment>; link?: Record<string, unknown> }

interface Session {
  readonly id: string;
  join(opts: { id?: string; as: 'camera' | 'viewer' | 'robot' | 'controller'; publish?: string[]; subscribe?: string[] }): Promise<{ access: Access }>;
  connect(access?: Access, opts?: { publish?: { camera?: MediaStream }; transport?: 'moq' | 'webrtc' }): Promise<this>;
  addSource<S extends Source>(s: S): S;                                     // join as camera + publish from here
  source(id: string): Source;
  attach<M extends Model<any, any>>(m: M | string, opts: { to: Source | string; as?: string; params?: Record<string, unknown> }): Promise<M>;
  detach(m: Model<any, any>): Promise<void>;
  watch<T = unknown>(path: string, cb: (rec: T) => void): Unsubscribe;    // 'cam0/objects' or 'cam0/objects:checks'
  watchVideo(sourceId: string, onFrame: (f: VideoFrame) => void): Promise<void>;
  graph(): Promise<Graph>;
  watchGraph(cb: (g: Graph) => void): Unsubscribe;
}

interface Client { sessions: { create(o: { name?: string; durationSeconds?: number; transport?: string }): Promise<Session>; get(id: string): Promise<Session>; list(): Promise<Graph[]>; close(id: string): Promise<void> }; catalog(): Promise<unknown> }
```

Python mirrors this with `Track[T]` (a `Protocol` whose `watch` also works as a decorator),
`Model.outputs` as an attribute namespace (`alerts.outputs.checks`), `TypedDict`s for the records,
`dataclass` configs, `as_=` for the reserved word, and synchronous control-plane calls.

## Appendix C. The job document

What `rgb.Job` builds, and what `rgb jobs validate|run job.json` consume. Written by hand, by an
agent, or by the builder, it is validated the same way.

```json
{"version": "0.1",
 "assets": {"clip": {"kind": "video", "url": "s3://bucket/run-42/cam0.mp4"}},
 "attachments": {"det": {"model": "sam3-detector-v1", "to": "clip", "params": {"prompts": ["person", "pallet"]}},
                 "mon": {"model": "qwen-monitor-v1",  "to": "clip", "params": {"rules": ["a fire broke out"]}}},
 "outputs": {"detections": {"from": "det", "to": "s3://bucket/run-42/out/detections.jsonl"},
             "events":     {"from": "mon", "to": "s3://bucket/run-42/out/events.jsonl"}}}
```

`job.document` returns this; `rgb jobs validate` checks it, resolves models and returns the estimate
without touching media; `rgb jobs run` validates, runs and prints the output locations. Asset ids are
identity; filenames never are.