# 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=3600, 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..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). **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 | 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 20), `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 or change count (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 ` (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 | | `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) | | `POST /sessions/{id}/record` | proposed: `501` | | `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/demo3_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), 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..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`. Proposed (designed, not built): `record`, 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//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": "", # 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 { readonly name: string; watch(cb: (rec: T) => void): Unsubscribe; } interface Attachment { model: string; kind: string; to: string; as: string; params: Record; state: 'admitted' | 'unplaced'; node: string | null; rev: number } interface Model, 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 }; watch(cb: (rec: O[P]) => void): Unsubscribe; // sugar for outputs[primary].watch configure(changes: Partial): Promise; // 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 { 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; attachments: Record; link?: Record } 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; addSource(s: S): S; // join as camera + publish from here source(id: string): Source; attach>(m: M | string, opts: { to: Source | string; as?: string; params?: Record }): Promise; detach(m: Model): Promise; watch(path: string, cb: (rec: T) => void): Unsubscribe; // 'cam0/objects' or 'cam0/objects:checks' watchVideo(sourceId: string, onFrame: (f: VideoFrame) => void): Promise; graph(): Promise; watchGraph(cb: (g: Graph) => void): Unsubscribe; } interface Client { sessions: { create(o: { name?: string; durationSeconds?: number; transport?: string }): Promise; get(id: string): Promise; list(): Promise; close(id: string): Promise }; catalog(): Promise } ``` 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.