Skip to content

Play a chunk in a layer and command the harness one round at a time #679

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
vertix wants to merge 10 commits into
base: main
Choose a base branch
from
Draft

Play a chunk in a layer and command the harness one round at a time #679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
6eba390
Play a chunk in a layer and command the harness one round at a time
vertix Aug 26, 2026
61292fd
Put every model session on the chunk contract, and fix what a review …
vertix Aug 27, 2026
1a4b8dc
Read a tap's answer by the side it sits on, not by its shape
vertix Aug 27, 2026
b409a02
Command nothing once the trial passes the instant it stops at
vertix Aug 27, 2026
73b80fd
Close the session `blocking()` opened before it refused it
vertix Aug 27, 2026
a807baa
Read the deadline off the field that carries it
vertix Aug 27, 2026
009593b
Let a new chunk supersede the one it replaces
vertix Aug 28, 2026
e6ded06
Pace the control loop by the instant the session names
vertix Aug 29, 2026
34af1a8
Let the caller own the calls a chunk source makes
vertix Aug 29, 2026
7390bb2
Open a policy's work per episode and play it in one `Session`
vertix Aug 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
210 changes: 0 additions & 210 deletions .basedpyright/baseline.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4871,14 +4871,6 @@
}
],
"./positronic/policy/codec.py": [
{
"code": "reportIncompatibleMethodOverride",
"range": {
"startColumn": 8,
"endColumn": 16,
"lineCount": 1
}
},
{
"code": "reportIncompatibleMethodOverride",
"range": {
Expand Down Expand Up @@ -5032,16 +5024,6 @@
}
}
],
"./positronic/policy/tests/test_layers.py": [
{
"code": "reportOptionalMemberAccess",
"range": {
"startColumn": 30,
"endColumn": 40,
"lineCount": 1
}
}
],
"./positronic/policy/tests/test_policy_io.py": [
{
"code": "reportArgumentType",
Expand Down Expand Up @@ -5075,126 +5057,6 @@
"lineCount": 1
}
},
{
"code": "reportOptionalIterable",
"range": {
"startColumn": 28,
"endColumn": 34,
"lineCount": 1
}
},
{
"code": "reportOptionalSubscript",
"range": {
"startColumn": 11,
"endColumn": 17,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 15,
"endColumn": 21,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 15,
"endColumn": 21,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 15,
"endColumn": 21,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 31,
"endColumn": 37,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 31,
"endColumn": 37,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 15,
"endColumn": 21,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 31,
"endColumn": 37,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 31,
"endColumn": 37,
"lineCount": 1
}
},
{
"code": "reportCallIssue",
"range": {
"startColumn": 11,
"endColumn": 30,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 11,
"endColumn": 30,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 15,
"endColumn": 21,
"lineCount": 1
}
},
{
"code": "reportOptionalIterable",
"range": {
"startColumn": 28,
"endColumn": 34,
"lineCount": 1
}
},
{
"code": "reportOptionalSubscript",
"range": {
"startColumn": 11,
"endColumn": 17,
"lineCount": 1
}
},
{
"code": "reportCallIssue",
"range": {
Expand All @@ -5211,22 +5073,6 @@
"lineCount": 1
}
},
{
"code": "reportCallIssue",
"range": {
"startColumn": 11,
"endColumn": 22,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 11,
"endColumn": 22,
"lineCount": 1
}
},
{
"code": "reportAttributeAccessIssue",
"range": {
Expand Down Expand Up @@ -5340,46 +5186,6 @@
"endColumn": 58,
"lineCount": 1
}
},
{
"code": "reportOptionalSubscript",
"range": {
"startColumn": 11,
"endColumn": 17,
"lineCount": 1
}
},
{
"code": "reportAttributeAccessIssue",
"range": {
"startColumn": 19,
"endColumn": 24,
"lineCount": 1
}
},
{
"code": "reportOptionalMemberAccess",
"range": {
"startColumn": 30,
"endColumn": 40,
"lineCount": 1
}
},
{
"code": "reportOptionalMemberAccess",
"range": {
"startColumn": 30,
"endColumn": 50,
"lineCount": 1
}
},
{
"code": "reportOptionalMemberAccess",
"range": {
"startColumn": 30,
"endColumn": 50,
"lineCount": 1
}
}
],
"./positronic/probe.py": [
Expand Down Expand Up @@ -5422,22 +5228,6 @@
"endColumn": 28,
"lineCount": 1
}
},
{
"code": "reportOptionalContextManager",
"range": {
"startColumn": 13,
"endColumn": 23,
"lineCount": 1
}
},
{
"code": "reportOptionalMemberAccess",
"range": {
"startColumn": 23,
"endColumn": 42,
"lineCount": 1
}
}
],
"./positronic/replay_record.py": [
Expand Down
25 changes: 13 additions & 12 deletions ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ claim an outsider can check.
## Positronic owns the control loop

The world runner and harness execute every episode: they drive the clock, deliver observations to
the policy, schedule and play back action chunks, and own resets and episode boundaries. This holds
the policy, emit the commands it answers, and own resets and episode boundaries. This holds
for every evaluation and data-collection run — in simulation and on real hardware, for any
embodiment, scene source, or scoring method. A foreign component never runs the loop and calls into
Positronic; Positronic runs the loop and calls into it.
Expand Down Expand Up @@ -116,21 +116,22 @@ control system its clock, and no component reads time at point of use. Trajector
the same time frame the observations carry, so a virtual clock, a slowed sim, or a replayed episode
changes nothing downstream.

**The layer owns the plan, the harness plays it, the driver executes.** A policy speaks in
trajectories — waypoints with absolute timestamps — because a model predicts a horizon, not an
instant. But a trajectory on the wire makes every driver buffer the future, and makes the recording
guess which prefix of that buffer actually ran. So the plan stops at the harness: a command channel
carries the single command due at the moment it is emitted, the driver executes the latest one and
**The layer owns the plan and plays it, the driver executes.** A policy predicts a horizon, not an
instant, so a chunk of waypoints is what a model answers. But a trajectory on the wire makes every
driver buffer the future, and makes the recording guess which prefix of that buffer actually ran. So
the plan stops inside the policy stack: a session answers the commands due at the moment it is
called, a command channel carries one command per emission, the driver executes the latest one and
holds otherwise, and emission time *is* execution time. Continuous-update schemes (RTC, temporal
ensembling) therefore need no special mechanism: they are layers that hand back a new trajectory
more often, and the harness keeps playing the old one until they do.
ensembling) therefore need no special mechanism: they are layers that re-query before the chunk they
hold runs out.

**The harness stays thin.** It is the one layer standing between any policy and any embodiment, so
anything it encodes about either side breaks the any-to-any goal. It assembles the observation
dict, calls the session, plays the returned trajectory one command per channel per round, and runs
episode lifecycle — nothing else. Scheduling, blending, history stacking and error recovery live in
the layer stack around the policy; a session returning `None` means "keep executing the current
trajectory".
dict, calls the session, emits the commands it answers, and runs episode lifecycle — nothing else.
Scheduling, blending, history stacking and error recovery live in the layer stack around the policy.
A call answers `(commands, resume_at_ns)`: what to emit now, and the instant the session wants its
next call. That instant is what paces the loop. The harness may call earlier: the episode deadline
cuts a round short, and a floor and a ceiling of its own bound every one of them.

**Inference cost is a fact of the trial, owned by the harness.** The policy declares its heavy work
as functions, and the framework runs each one off the loop thread. That work costs the trial either
Expand Down
54 changes: 27 additions & 27 deletions docs/connect-your-model.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ How the client fills the delay and merges successive predictions is a swappable

Four small concepts make up the API. You meet them whether you use a built-in server or write your own.

**Policy and Session.** A `Policy` is your loaded model: it holds the weights and knows how to start an episode. `policy.new_session()` begins one episode and returns a `Session`. You call the session once per timestep with the latest observation and your clock reading, and it returns the next actions to run. Per-episode state (history, the trajectory in flight) lives in the session — so one `Policy` can serve several robots at once, each with its own `Session`.
**Policy, episode and Session.** A `Policy` is your loaded model: it holds the weights and knows how to start an episode. `policy.episode()` opens one episode and answers the work that episode runs, by name — the model call under `infer`, and whatever that call needs for as long as the episode lasts (a connection, a client, a reset model). The framework runs each function off the loop thread and hands the handles to a `Session`, which it opens with `policy.new_session(rt)` and calls once per timestep with the latest observation and its clock reading. A session answers the commands to run now and the instant it wants its next call. A policy served behind the `remote` marker answers a chunk from `infer`; the rig's `ChunkPlayer` is the session that holds that chunk and emits each waypoint at its own time. One `Policy` serves several robots at once, each with its own episode.

**Codec.** Different models want different inputs: end-effector pose vs joint angles, absolute targets vs deltas, 224×224 vs 512×512 images. A `Codec` translates between the robot's raw data (what is on the wire) and your model's format — `encode` on the way in, `decode` on the way out. The same codec prepares the training data, so a model is served exactly the way it was trained. The full catalog is in the [Codecs Guide](codecs.md).

Expand Down Expand Up @@ -172,16 +172,24 @@ Implement a `Policy`, close a pipeline over it with `PolicySource`, and hand the
```python
from positronic.drivers.roboarm import command
from positronic.offboard import PolicyServer
from positronic.policy import Policy, Session
from contextlib import contextmanager

from positronic.drivers.roboarm import command
from positronic.policy import INFER, Policy
from positronic.policy.spec import PolicySource, remote
from positronic.policy.layers import ChunkedSchedule, StopOnFault
from positronic.policy.layers import ChunkPlayer, StopOnFault


class MySession(Session):
class MyPolicy(Policy):
def __init__(self, model):
self._model = model

def __call__(self, obs, time_ns):
@contextmanager
def episode(self, context=None):
# Per-episode setup goes here, and the teardown after the yield.
yield {INFER: self._infer}

def _infer(self, obs):
# obs holds the raw keys from the wire table above. Pick what you need:
images = obs['image.exterior']
ee = obs['robot_state.ee_pose']
Expand All @@ -192,33 +200,25 @@ class MySession(Session):
for pose in predicted_poses
]


class MyPolicy(Policy):
def __init__(self, model):
self._model = model

def new_session(self, context=None, rt=None):
return MySession(self._model) # per-episode setup goes here

@property
def meta(self):
return {'type': 'my_model'}


pipeline = StopOnFault() | ChunkedSchedule() | remote | PolicySource(MyPolicy(load_my_model()))
pipeline = StopOnFault() | ChunkPlayer() | remote | PolicySource(MyPolicy(load_my_model()))
PolicyServer(pipeline, host='0.0.0.0', port=8000).serve()
```

The pipeline reads left to right: everything left of the `remote` marker is the client-side stack the server declares in its handshake (here the standard `StopOnFault` and `ChunkedSchedule`); everything right of it runs on the server. `PolicySource` is the pipeline's terminal — a model source that serves one already-built policy.
The pipeline reads left to right: everything left of the `remote` marker is the client-side stack the server declares in its handshake (here the standard `StopOnFault` and `ChunkPlayer`); everything right of it runs on the server. `PolicySource` is the pipeline's terminal — a model source that serves one already-built policy.

The left side is not optional: a pipeline with nothing there is refused when the server starts, and a rig refuses a handshake that declares nothing. It needs a scheduler in particular, and `StopOnFault` outside that scheduler — an arm that is faulted or busy is not taking the plan it was given, so the layer answers the empty trajectory and the rig stops rather than resuming a chunk stamped before. Actions come back timestamped relative to their chunk, and `ChunkedSchedule` is what turns those into times on the rig's clock; a stack that leaves them relative — or anchors them twice — makes the harness reject the chunk at the first inference, since it schedules nothing more than `MAX_ACTION_SKEW_SEC` from now.
The left side is not optional: a pipeline with nothing there is refused when the server starts, and a rig refuses a handshake that declares nothing. It needs a player in particular, and `StopOnFault` outside that player — an arm that is faulted or busy is not taking the plan it was given, so the layer commands nothing and drops the chunk rather than playing one stamped before. Actions come back timestamped relative to their chunk, and `ChunkPlayer` is what holds the chunk and emits each waypoint at its own time on the rig's clock; a stack with no player answers a chunk where the harness expects commands, and the episode refuses to open. A chunk that reaches the player already anchored is refused too, since the player places nothing more than `MAX_ACTION_SKEW_SEC` from the call.

The session's `time_ns` argument is the caller's clock reading in nanoseconds, the same unit the observation's `obs_time_ns` carries. A session reads no clock of its own, and a policy that schedules nothing accepts the value and ignores it.
`infer` takes the observation and nothing else: the clock belongs to the session that plays what it answers, and each action carries its own offset from the call. A session's `time_ns` argument is the caller's clock reading in nanoseconds, the same unit the observation's `obs_time_ns` carries. A session reads no clock of its own.

If you put a `Codec` right of the marker (`ChunkedSchedule() | remote | codec | PolicySource(...)`), your session works entirely in *model space* — it receives encoded observations and returns model-native actions, and the codec handles the wire format. A codec that encodes images should also bound them on the rig, so full-resolution frames never cross the wire — that is what the built-in vendor pipelines do:
If you put a `Codec` right of the marker (`ChunkPlayer() | remote | codec | PolicySource(...)`), your `infer` works entirely in *model space* — it receives encoded observations and returns model-native actions, and the codec handles the wire format. A codec that encodes images should also bound them on the rig, so full-resolution frames never cross the wire — that is what the built-in vendor pipelines do:

```python
StopOnFault() | ChunkedSchedule() | RestrictImageSize() | remote | codec | source
StopOnFault() | ChunkPlayer() | RestrictImageSize() | remote | codec | source
```

Give it the geometry your codec encodes to — `RestrictImageSize(224, 224)` for a 224x224 model — so a frame is shrunk once, on the rig. The default is a loose 640x640, for a codec that resizes to nothing in particular. Leaving it out costs bandwidth, not correctness.
Expand Down Expand Up @@ -253,15 +253,15 @@ Every message is msgpack. Numpy arrays use a custom extension:
robot commands:

```python
import time

from positronic.offboard.protocol import serialise, deserialise

session = policy.new_session() # one Session per episode/connection
async for message in websocket.iter_bytes():
obs = deserialise(message) # dict with numpy arrays
actions = session(obs, time.time_ns()) # list of action dicts (or None)
await websocket.send_bytes(serialise({"result": actions}))
from positronic.policy import INFER

with policy.episode() as fns: # one episode per connection
infer = fns[INFER]
async for message in websocket.iter_bytes():
obs = deserialise(message) # dict with numpy arrays
actions = infer(obs) # list of action dicts (or None)
await websocket.send_bytes(serialise({"result": actions}))
```

A server written against another stack cannot import that module. Answer with the command as the plain
Expand Down
Loading