From 2a5ebda11fc55a99e74a33170b1676fad65b04fa Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Tue, 8 Sep 2026 18:14:31 +0000 Subject: [PATCH 1/2] Skip building an observation nothing below will read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rig runs its control loop far faster than it re-queries a policy, and two layers were doing their most expensive work on every tick regardless. `TemporalStack` stacked a whole history window per call, and a codec resized every frame of it, and the scheduling layer below then answered `None` without looking at either. Measured on a real rollout: the window is stacked 57 times per chunk for 1886 ms of CPU, on the thread that drives the arm, in a chunk period of 3394 ms. 41 of those samples run while the arm is playing, with no round trip open at all. `Session.reads_observation(time_ns)` says whether a call would read what it is given. A scheduling layer answers False while its own chunk plays; a remote session answers False while a round trip is in flight. The call happens either way — a session waiting on a function in flight is answered by being called, not by the observation it carries — so a layer told no hands on what it already has. The default is True and a delegating session keeps it, because the answer is about the `__call__` that gives it: a recording tap wraps another session and still reads every observation, and a forwarded no would have let the layers above hand it something other than what went to the server. The two layers that forward are the two that genuinely pass the observation through. It reads the caller's clock rather than the observation, because a served pipeline runs these layers over observations the harness never stamped. That clock runs at or past the observation's, so the gate can only answer True where the call then answers None: a window built and unused, never one skipped and wanted. Ticket: Positronic-Robotics/internal#1168 #refs --- positronic/policy/base.py | 13 +++++ positronic/policy/codec.py | 7 ++- positronic/policy/layers.py | 12 +++++ positronic/policy/remote.py | 3 ++ positronic/policy/tests/test_layers.py | 75 +++++++++++++++++++++++++- 5 files changed, 108 insertions(+), 2 deletions(-) diff --git a/positronic/policy/base.py b/positronic/policy/base.py index b3d93fac9..6ed81e435 100644 --- a/positronic/policy/base.py +++ b/positronic/policy/base.py @@ -73,6 +73,19 @@ def __call__(self, obs: Mapping[str, Any], time_ns: int) -> list[dict[str, Any]] ``time_ns`` is the caller's clock reading in nanoseconds. A session reads no clock of its own. """ + def reads_observation(self, time_ns: int) -> bool: + """Whether a call made at ``time_ns`` would read the observation it is given. + + Answering False permits the caller to hand over whatever it already has instead of building + what this session asked for, so a session whose ``__call__`` reads the observation must answer + True — including one that only records it. The call happens either way. + + ``time_ns`` is the caller's clock, the same reading ``__call__`` takes, so this asks nothing of + the observation itself: a served pipeline runs these layers over observations the harness + never stamped. + """ + return True + @property def meta(self) -> dict[str, Any]: """What this session reports about its model and its episode.""" diff --git a/positronic/policy/codec.py b/positronic/policy/codec.py index 78c0f6bb2..56363b389 100644 --- a/positronic/policy/codec.py +++ b/positronic/policy/codec.py @@ -114,8 +114,13 @@ def __init__(self, inner: Session, codec: 'Codec'): super().__init__(inner) self._codec = codec + def reads_observation(self, time_ns): + # Forwarded because this session encodes rather than reads: told no, it hands its own input on. + return self._inner.reads_observation(time_ns) + def __call__(self, obs, time_ns): - encoded = self._codec.encode(obs) + reads = self._inner.reads_observation(time_ns) + encoded = self._codec.encode(obs) if reads else obs action = self._inner(encoded, time_ns) if action is None: return None diff --git a/positronic/policy/layers.py b/positronic/policy/layers.py index 530039e06..746c996cf 100644 --- a/positronic/policy/layers.py +++ b/positronic/policy/layers.py @@ -80,6 +80,14 @@ def __init__(self, inner: Session): super().__init__(inner) self._trajectory_end: float | None = None + def reads_observation(self, time_ns): + # Mirrors ``__call__``'s guard against the same field, so the two cannot drift. The + # call's clock runs at or past the observation's, so this can only answer True where + # ``__call__`` then answers None: a window built and unused, never one skipped and wanted. + if self._trajectory_end is not None and time_ns / 1e9 < self._trajectory_end: + return False + return self._inner.reads_observation(time_ns) + def __call__(self, obs, time_ns): if self._trajectory_end is not None and _obs_time(obs) < self._trajectory_end: return None @@ -180,7 +188,11 @@ def __init__(self, inner: Session, keys: tuple[str, ...], offsets_sec: tuple[flo def __call__(self, obs, time_ns): now = _obs_time(obs) + # Unconditional, including on the ticks that skip sampling below: a gap here is a hole in + # the window every later sample is taken from. self._buffer.append(now, {k: obs[k] for k in self._keys}) + if not self._inner.reads_observation(time_ns): + return self._inner(obs, time_ns) return self._inner({**obs, **self._buffer.sample(now)}, time_ns) def cancel(self): diff --git a/positronic/policy/remote.py b/positronic/policy/remote.py index 875e5620b..71f740309 100644 --- a/positronic/policy/remote.py +++ b/positronic/policy/remote.py @@ -93,6 +93,9 @@ def __call__(self, obs: cabc.Mapping[str, Any], time_ns: int) -> list[dict[str, return None return [result] if isinstance(result, dict) else result + def reads_observation(self, time_ns: int) -> bool: + return self._answer is None + def cancel(self): # The cancel says the world the chunk applies to has gone. The session still reads the round trip # for its failure, and drops the chunk that comes with it. diff --git a/positronic/policy/tests/test_layers.py b/positronic/policy/tests/test_layers.py index 28a5d1007..be08880dc 100644 --- a/positronic/policy/tests/test_layers.py +++ b/positronic/policy/tests/test_layers.py @@ -12,7 +12,7 @@ from positronic.geom import Rotation, Transform3D from positronic.policy import spec from positronic.policy.action import AbsoluteJointsAction, AbsolutePositionAction, IKJointsAction, JointDeltaAction -from positronic.policy.base import Layer, Policy, Session +from positronic.policy.base import DelegatingSession, Layer, Policy, Session from positronic.policy.codec import ( ActionHorizon, ActionTimestamp, @@ -326,6 +326,79 @@ def test_every_arm_channel_is_stamped(self): np.testing.assert_array_equal(decoded[keys.TARGET_JOINTS], np.zeros(7)) +class _DecliningSession(Session): + """A session that reads no observation — what a remote one answers with a round trip in flight.""" + + def __init__(self): + self.seen: dict[str, Any] = {} + + def __call__(self, obs, time_ns): + self.seen = dict(obs) + return None + + def reads_observation(self, time_ns): + return False + + +class _CapturingInner(Session): + def __init__(self): + self.seen: dict[str, Any] = {} + + def __call__(self, obs, time_ns): + self.seen = dict(obs) + return None + + +class TestReadsObservation: + """The gate that lets a layer skip building an observation nothing below will read.""" + + def test_a_session_reads_its_observation_by_default(self): + assert _ConstSession(None).reads_observation(0) is True + + def test_a_schedule_reads_nothing_while_its_chunk_plays(self): + session = ChunkedSchedule().make_session(_ConstSession([{keys.ACTION_TIMESTAMP: 2.0}])) + assert session.reads_observation(0) is True # nothing emitted yet + session(_obs(0), 0) # emits, ending 2 s from now + assert session.reads_observation(int(1.0e9)) is False # mid-chunk + assert session.reads_observation(int(3.0e9)) is True # played out + + def test_the_gate_agrees_with_the_call(self): + """Both read `_trajectory_end`, in two places, so they are pinned against each other: a call + that answers None is one the gate would have declined.""" + session = ChunkedSchedule().make_session(_ConstSession([{keys.ACTION_TIMESTAMP: 2.0}])) + session(_obs(0), 0) + for t_ns in (int(0.5e9), int(1.9e9), int(2.0e9), int(2.1e9), int(5.0e9)): + declined = not session.reads_observation(t_ns) + answered_none = session(_obs(t_ns / 1e9), t_ns) is None + assert declined is answered_none, f'disagreed at {t_ns} ns' + + def test_a_delegating_session_answers_for_itself(self): + """A session that wraps another still reads its own observation — a recording tap is the live + case. Forwarding the inner answer would let a layer above hand it something else to log.""" + + class _Tap(DelegatingSession): + def __call__(self, obs, time_ns): + self.logged = dict(obs) + return self._inner(obs, time_ns) + + assert _Tap(_DecliningSession()).reads_observation(0) is True + + def test_a_stack_is_not_sampled_when_nothing_below_reads_it(self): + """The window is the expensive part, and most ticks do not use it. Recording still happens on + every tick — the history it builds is what the next real call is sampled from.""" + inner = _DecliningSession() + session = TemporalStack(keys=('cam',), offsets_sec=(-0.2, 0.0)).make_session(inner) + session({**_obs(0), 'cam': np.zeros((4, 4, 3), np.uint8)}, 0) + assert inner.seen['cam'].shape == (4, 4, 3), 'passed through unstacked' + + def test_a_stack_is_sampled_when_something_below_reads_it(self): + inner = _CapturingInner() + session = TemporalStack(keys=('cam',), offsets_sec=(-0.2, 0.0)).make_session(inner) + session({**_obs(0), 'cam': np.zeros((4, 4, 3), np.uint8)}, 0) + # A stacked entry carries the window's length on a new leading axis. + assert inner.seen['cam'].shape == (2, 4, 4, 3) + + class TestTemporalStack: OFFSETS = (-0.2, -0.1, 0.0) From baf741665d2efd17a26e73a282b667a458b83b1a Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Tue, 8 Sep 2026 20:21:38 +0000 Subject: [PATCH 2/2] Ask the read gate about the observation the call will read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ChunkedSchedule` judged expiry at the observation's own instant in the call and at the caller's clock in the gate. Where the two straddle the chunk's end — an observation ahead of the clock, which the schedule's own test permits — the gate declined and the call went through, so the layer above handed the inner session a raw, unstacked observation. The query now takes what the call takes, and the schedule makes the same comparison in both. A test pins the case where the two readings differ. Ticket: none - a review finding on an unmerged branch --- positronic/policy/base.py | 10 +++++----- positronic/policy/codec.py | 6 +++--- positronic/policy/layers.py | 12 +++++------- positronic/policy/remote.py | 2 +- positronic/policy/tests/test_layers.py | 22 +++++++++++++++------- 5 files changed, 29 insertions(+), 23 deletions(-) diff --git a/positronic/policy/base.py b/positronic/policy/base.py index 6ed81e435..a76cd0bec 100644 --- a/positronic/policy/base.py +++ b/positronic/policy/base.py @@ -73,16 +73,16 @@ def __call__(self, obs: Mapping[str, Any], time_ns: int) -> list[dict[str, Any]] ``time_ns`` is the caller's clock reading in nanoseconds. A session reads no clock of its own. """ - def reads_observation(self, time_ns: int) -> bool: - """Whether a call made at ``time_ns`` would read the observation it is given. + def reads_observation(self, obs: Mapping[str, Any], time_ns: int) -> bool: + """Whether the call this query precedes would read the observation it is given. Answering False permits the caller to hand over whatever it already has instead of building what this session asked for, so a session whose ``__call__`` reads the observation must answer True — including one that only records it. The call happens either way. - ``time_ns`` is the caller's clock, the same reading ``__call__`` takes, so this asks nothing of - the observation itself: a served pipeline runs these layers over observations the harness - never stamped. + It takes what ``__call__`` takes, so a session deciding on a field decides on the same value in + both: a query answered against the caller's clock and a call answered against the observation's + own stamp disagree wherever the two readings straddle a boundary. """ return True diff --git a/positronic/policy/codec.py b/positronic/policy/codec.py index 56363b389..b837c7619 100644 --- a/positronic/policy/codec.py +++ b/positronic/policy/codec.py @@ -114,12 +114,12 @@ def __init__(self, inner: Session, codec: 'Codec'): super().__init__(inner) self._codec = codec - def reads_observation(self, time_ns): + def reads_observation(self, obs, time_ns): # Forwarded because this session encodes rather than reads: told no, it hands its own input on. - return self._inner.reads_observation(time_ns) + return self._inner.reads_observation(obs, time_ns) def __call__(self, obs, time_ns): - reads = self._inner.reads_observation(time_ns) + reads = self._inner.reads_observation(obs, time_ns) encoded = self._codec.encode(obs) if reads else obs action = self._inner(encoded, time_ns) if action is None: diff --git a/positronic/policy/layers.py b/positronic/policy/layers.py index 746c996cf..96b43bee9 100644 --- a/positronic/policy/layers.py +++ b/positronic/policy/layers.py @@ -80,13 +80,11 @@ def __init__(self, inner: Session): super().__init__(inner) self._trajectory_end: float | None = None - def reads_observation(self, time_ns): - # Mirrors ``__call__``'s guard against the same field, so the two cannot drift. The - # call's clock runs at or past the observation's, so this can only answer True where - # ``__call__`` then answers None: a window built and unused, never one skipped and wanted. - if self._trajectory_end is not None and time_ns / 1e9 < self._trajectory_end: + def reads_observation(self, obs, time_ns): + # The same guard ``__call__`` makes, against the same instant, so the two cannot disagree. + if self._trajectory_end is not None and _obs_time(obs) < self._trajectory_end: return False - return self._inner.reads_observation(time_ns) + return self._inner.reads_observation(obs, time_ns) def __call__(self, obs, time_ns): if self._trajectory_end is not None and _obs_time(obs) < self._trajectory_end: @@ -191,7 +189,7 @@ def __call__(self, obs, time_ns): # Unconditional, including on the ticks that skip sampling below: a gap here is a hole in # the window every later sample is taken from. self._buffer.append(now, {k: obs[k] for k in self._keys}) - if not self._inner.reads_observation(time_ns): + if not self._inner.reads_observation(obs, time_ns): return self._inner(obs, time_ns) return self._inner({**obs, **self._buffer.sample(now)}, time_ns) diff --git a/positronic/policy/remote.py b/positronic/policy/remote.py index 71f740309..1d5f9aa1a 100644 --- a/positronic/policy/remote.py +++ b/positronic/policy/remote.py @@ -93,7 +93,7 @@ def __call__(self, obs: cabc.Mapping[str, Any], time_ns: int) -> list[dict[str, return None return [result] if isinstance(result, dict) else result - def reads_observation(self, time_ns: int) -> bool: + def reads_observation(self, obs: cabc.Mapping[str, Any], time_ns: int) -> bool: return self._answer is None def cancel(self): diff --git a/positronic/policy/tests/test_layers.py b/positronic/policy/tests/test_layers.py index be08880dc..a95d581f0 100644 --- a/positronic/policy/tests/test_layers.py +++ b/positronic/policy/tests/test_layers.py @@ -336,7 +336,7 @@ def __call__(self, obs, time_ns): self.seen = dict(obs) return None - def reads_observation(self, time_ns): + def reads_observation(self, obs, time_ns): return False @@ -353,14 +353,14 @@ class TestReadsObservation: """The gate that lets a layer skip building an observation nothing below will read.""" def test_a_session_reads_its_observation_by_default(self): - assert _ConstSession(None).reads_observation(0) is True + assert _ConstSession(None).reads_observation(_obs(0), 0) is True def test_a_schedule_reads_nothing_while_its_chunk_plays(self): session = ChunkedSchedule().make_session(_ConstSession([{keys.ACTION_TIMESTAMP: 2.0}])) - assert session.reads_observation(0) is True # nothing emitted yet + assert session.reads_observation(_obs(0), 0) is True # nothing emitted yet session(_obs(0), 0) # emits, ending 2 s from now - assert session.reads_observation(int(1.0e9)) is False # mid-chunk - assert session.reads_observation(int(3.0e9)) is True # played out + assert session.reads_observation(_obs(1.0), int(1.0e9)) is False # mid-chunk + assert session.reads_observation(_obs(3.0), int(3.0e9)) is True # played out def test_the_gate_agrees_with_the_call(self): """Both read `_trajectory_end`, in two places, so they are pinned against each other: a call @@ -368,10 +368,18 @@ def test_the_gate_agrees_with_the_call(self): session = ChunkedSchedule().make_session(_ConstSession([{keys.ACTION_TIMESTAMP: 2.0}])) session(_obs(0), 0) for t_ns in (int(0.5e9), int(1.9e9), int(2.0e9), int(2.1e9), int(5.0e9)): - declined = not session.reads_observation(t_ns) + declined = not session.reads_observation(_obs(t_ns / 1e9), t_ns) answered_none = session(_obs(t_ns / 1e9), t_ns) is None assert declined is answered_none, f'disagreed at {t_ns} ns' + def test_the_gate_reads_the_observation_instant_the_call_reads(self): + """An observation ahead of the caller's clock: a gate reading the clock would decline, while + the call goes through to the inner session — which then gets whatever the layer above kept.""" + session = ChunkedSchedule().make_session(_ConstSession([{keys.ACTION_TIMESTAMP: 2.0}])) + session(_obs(0), 0) + assert session.reads_observation(_obs(2.5), int(1.0e9)) is True + assert session(_obs(2.5), int(1.0e9)) is not None + def test_a_delegating_session_answers_for_itself(self): """A session that wraps another still reads its own observation — a recording tap is the live case. Forwarding the inner answer would let a layer above hand it something else to log.""" @@ -381,7 +389,7 @@ def __call__(self, obs, time_ns): self.logged = dict(obs) return self._inner(obs, time_ns) - assert _Tap(_DecliningSession()).reads_observation(0) is True + assert _Tap(_DecliningSession()).reads_observation(_obs(0), 0) is True def test_a_stack_is_not_sampled_when_nothing_below_reads_it(self): """The window is the expensive part, and most ticks do not use it. Recording still happens on