menlo-rcs-types 0.1.1

Typed Pydantic models for the RCS → S1 (LiveKit dispatch worker) wire contract. Shared between RCS and any worker dispatched into a robot session.

menlo-rcs-types

Typed Pydantic models for the RCS → S1 wire contract — the JSON payloads RCS attaches to LiveKit calls that any dispatched worker (Uranus, rcw, future) reads on connect.

LiveKit is the transport; this package is just the shape of the fields RCS writes onto LiveKit primitives.

What's in it

Two models. That's the whole boundary.

DispatchMetadata (menlo_rcs_types.dispatch)

The JSON RCS passes to livekit.create_agent_dispatch(metadata=...) when injecting a worker into a room. Each worker process receives this on its JobContext.job.metadata.

class DispatchMetadata(BaseModel):
    robot_id: str               # the robot this session is for
    user_id: str                # who initiated the session
    scene_id: str | None        # optional opaque scene tag (env workers use this)

RoomMetadata (menlo_rcs_types.room)

The JSON RCS writes onto the LiveKit room itself. Visible to every participant; carries the orchestrator's view of room state so workers and clients can read it.

class RoomMetadata(BaseModel):
    robot_id: str
    status: str                                                 # "initializing" | "active" | …
    dispatched_agent_names: list[str]                           # what RCS dispatched
    required_participants: dict[str, RequiredParticipantSlot]   # per-type counts
    dispatches: dict[str, DispatchStatusEntry]                  # RCS's dispatch tracking
    scene_id: str | None
    created_at: datetime

Participant slot matching

RCS, webhooks, and SDK clients share the same identity ↔ slot rule:

• an identity equals the slot name (sim-worker), or
• starts with slot- (sim-worker-abc)

The rule lives on the models it describes — this package exposes only types:

• RequiredParticipantSlot.identity_matches(identity, slot_name) — the bare string rule, a static helper for callers matching raw candidate names (e.g. RCS webhook classification).
• RequiredParticipantSlot.matches(identity) — does this identity belong to this slot.
• RoomMetadata.required_slots() — slot names with min_count > 0.
• RoomMetadata.satisfied_by(identities) — are all required slots filled. Assigns each identity to its longest matching slot, so overlapping prefixes (sim vs sim-worker) never double-count one participant.

Forward compatibility

All three models declare model_config = ConfigDict(extra="allow"). Adding optional fields in a later version of this package does not break workers built against an older version — unknown fields silently pass through model_validate_json.

This is the contract:

• New optional fields → safe to add anytime
• Renaming or removing a field → breaking change; bump version, coordinate rollouts
• Workers should never strictly-validate (extra="forbid") against this schema

Usage

# RCS side — building the metadata it passes to LiveKit
from menlo_rcs_types import DispatchMetadata

md = DispatchMetadata(robot_id="rb_abc", user_id="user_long", scene_id="warehouse")
await livekit.create_agent_dispatch(
    room=room_name,
    agent_name="sim-worker",
    metadata=md.model_dump_json(),
)


# Worker side — parsing the metadata on dispatch
from menlo_rcs_types import DispatchMetadata

async def runtime_session(ctx: JobContext) -> None:
    md = DispatchMetadata.model_validate_json(ctx.job.metadata or "{}")
    logger.info(
        "dispatched",
        robot_id=md.robot_id,
        user_id=md.user_id,
        scene_id=md.scene_id,
    )

What's NOT in this package

• AttachSessionRequest or any other client → RCS HTTP body shapes. Those are RCS's HTTP API surface, not the RCS → S1 wire. They live in RCS.
• Per-capability variants (SimHostMetadata, RuntimeExecutorMetadata, …). Both current workers use the same flat DispatchMetadata; capability-specific extras can ride on extra="allow" until they justify their own model.
• Worker self-registration / heartbeat protocols. Workers register with LiveKit, not RCS, in V1.

Development

make dev       # uv sync --group dev
make check     # lint + fmt-check + type + test (CI gate)

Python 3.11+, Pydantic v2, ruff, mypy --strict. Same toolchain as runtime-tools / asimov-protocol.

Related

• packages/py/runtime-tools/ — the agent-facing Runtime Tools API (different layer: tools that run inside a session, after the worker is in the room)
• packages/py/asimov-protocol/ — robot-firmware-level protobuf (lower than RCS)
• services/rcs/ — produces these payloads
• services/uranus/, services/rcw/ — parse these payloads