The policy stack cannot observe while inference runs, and does not own its schedule

[vertix]

The harness decides when the policy stack runs, and holds the schedule the stack produced. Both belong to the
stack, and the two together cap what a serving algorithm can express.

Where this starts

Harness._step returns before entering the stack whenever a call is outstanding:

if self._future is not None and not self._take(self._future, clock):
    return
obs = self._build_obs(clock)

So for the duration of a model call, no wrapper is entered at all. Two shipped wrappers need to be:

• TemporalStack fills its window by recording one sample per call. Across an inference gap it records
  nothing, and _StackBuffer.sample carries the last value forward — so a video-conditioned model trained on
  continuous frames is served a frozen one for the length of the call.

• StopOnFault reads keys.ROBOT_FAULT off an observation. During a call no observation is built, so a fault
  that starts and clears inside the inference window is never seen, and the pre-fault chunk keeps playing
  (Play the policy's trajectory in the harness, one command per channel per round #591, thread on harness.py).

  This one is worth stating as more than a race, because the arm is built to make it happen. franka.py
  clears its own errors — the control loop calls recover_from_errors() on every iteration it is in error, at
  a 2 kHz rate limit — and says so:

  if in_error:
      # The driver always clears a recoverable error itself; making it optional (hold in
      # ERROR for out-of-band recovery instead) is a config knob to add when an embodiment
      # needs it.
      robot.recover_from_errors()

  So ERROR is transient by construction, and a fault can appear and clear well inside one inference window.
  The harness looks at the arm only when it builds an observation, which it does not do while a call is
  outstanding, so the whole episode goes unseen: no stop, no re-plan, and _play carries on driving waypoints
  planned for a pose the arm has since left. Polling for a state the driver is actively erasing is the wrong
  shape — the fault is an edge, and it needs latching rather than sampling. Before playing moved into the
  harness the driver owned the trajectory and saw every state at its own rate, so it cancelled on a fault of
  any length.

Neither is a regression: before #591 the call blocked the loop thread, so both were starved identically. What
changed is that the rig now keeps moving through the gap instead of freezing in it, which makes the gap matter.

What we want to be able to write

A developer should be able to express any reasonable serving algorithm as a wrapper, with codecs and wrappers
as the tools. Concretely: real-time chunking and temporal ensembling
(Zhao et al. 2023, Black et al. 2025),
which re-query before the current chunk is exhausted and merge successive predictions rather than replacing
them.

Most of that already works. #591 gives re-query-before-exhaustion — the harness keeps playing while a call is
in flight, which _ReplanEarly in test_harness.py pins. A wrapper that wants to merge rather than replace can
hold its own chunks and return the merged whole. The gaps are the observation hole above, and that only one call
can be in flight.

The principle: the policy owns its schedule. It decides when to infer, what to release, and when. For that
to be writable rather than guessed at, the rig has to commit to what it provides, and to what it will do
regardless.

Goal 1 — the stack observes every tick

One sentence, two victims: the harness must enter the stack on every control tick, including while a call is
outstanding, so a wrapper that needs to watch can watch.

Two candidate shapes:

• AsyncInfer as a wrapper. Move the "a call is in flight" gate out of the harness and into the stack, as a
  position in the chain: layers outside it run every tick, layers inside run on the worker. The harness then
  enters the stack every tick and is told None. Needs no protocol or wire-format change — the harness already
  does executor.submit(session, obs); this does it one layer in. It does mean async_infer joins
  WIRE_WRAPPERS, so the server declares whether the rig overlaps inference with execution. That is defensible:
  overlap is a property of the serving algorithm (RTC requires it, plain chunked scheduling does not), and the
  server is the party that knows the model's latency and horizon.
• An observe split. The harness calls observe(obs) down the chain every tick and __call__ only at
  re-query. Threading stays where it is; nothing joins the wire vocabulary. The cost is a second method every
  wrapper must keep consistent with the first.

For the fault half specifically, what is delivered has to be real before delivery is fixed: today the harness
computes the fault (_is_faulted, stapled on as keys.ROBOT_FAULT) because the serializer drops a faulted arm's
frame. #637 makes robot_state.status robot data the serializer emits on every frame, which also closes #619.
No edge or latch is needed on top: on the Franka a fault stays in ERROR for seconds, so once the stack is
entered every tick every poll sees it.

Goal 2 — the contract between rig and algorithm

If the server declares the whole algorithm, local half included, then the rig owes it a description of the
ground it stands on.

What the rig guarantees the algorithm:

• Emission is execution. A returned action goes out at the instant it was read. Nothing is buffered between
  the harness and the device — _cancel_session already relies on this ("devices hold their last commanded
  position; nothing is buffered downstream to clear").
• Last command wins, immediately. No queueing, interpolation, or reduction in the driver.
• Timing bounds. Never called earlier than it asked for; a stated worst-case lateness; and a guaranteed
  maximum interval regardless, so a wrapper cannot wedge the rig by asking badly.
• One timeline. now() and obs[keys.OBS_TIME_NS] are the same clock, and an observation is no older than a
  stated bound when the call is made.
• Lifecycle. Reset before an episode ends, and no calls after.

Goal 3 — simplify wrappers (optional, independent)

Each of these stands on its own and none needs the above.

Drop timestamps from what the session returns. Today the session returns a trajectory of timestamped
actions and the harness holds it in a TrajectoryPlayer until each waypoint is due. Instead let the session
return the command to execute now, and make releasing the wrapper's own business — which is what "the policy
owns its schedule" means concretely. ChunkedSchedule holds its chunk and releases from it; the harness keeps
start, stop and the emitters. _install, _play, the players and the timestamp on the wire between stack and
harness all go, and RTC/TE become wrappers manipulating a plan they own rather than merging against a buffer
someone else holds.

The constraint that decides the design: _pace sleeps to the next waypoint precisely because the harness holds
the schedule —

due = [ts for player in self._players.values() if (ts := player.next_due()) is not None]
return pimm.Sleep(min(POLL_PERIOD_SEC, max(min(due) - clock.now_ns(), 1) / 1e9))

Move the schedule into the stack and the harness has nothing to pace on; it polls, and every waypoint lands up
to a poll period late. Sim is unaffected (fixed control period, still deterministic); a rig loses
waypoint-accurate timing, which is most of what the per-tick wire buys. So this needs the session to say when it
next has something to do — the same value TrajectoryPlayer.next_due() already is, sourced from the session and
translated by _pace into the pimm.Sleep it already returns.

Delete cancel() as a protocol. Every wrapper implements it and must remember super().cancel();
forgetting is silent. It exists only to return a session to its start, and the system already knows how to build
a fresh one. Reset becomes discard-and-rebuild of the inner session.

Tighten the return contract. A scheduling wrapper currently normalizes sloppy server output:

A single-action session may return a bare dict, and a no-codec path may omit timestamp (servers can
stamp/truncate themselves); normalize both so an immediate action executes instead of raising.

That belongs once at the wire boundary. Left where it is, every future wrapper pays the same tax.

Collapse the two classes per wrapper. StopOnFault is roughly twenty lines of scaffolding — wrapper,
_Session, wrap_session, to_spec — around four lines of behaviour. The split is needed (one policy serves
many robots; sessions are per-episode), but the ceremony could reduce to a base taking just the behaviour.

Two more, lower value: now threads through all fifteen wrap_session signatures for one consumer, though it
likely survives — ChunkedSchedule anchors to inference completion, which no observation stamp supplies, and
a callable cannot live in context. And to_spec spells each wire name a second time; deriving it from
WIRE_WRAPPERS would couple the table back to the classes it is deliberately decoupled from.

Explored and parked

Recorded so they are not re-derived, and not mistaken for requirements.

• Sessions as pimm control systems. Buys nothing. harness.py is the only file in positronic/policy that
  imports pimm, and it exists to be exactly that adapter.
• An observe/deliver fold (observations pull down one leg, commands push up the other, terminating in the
  emitters). Solves "release a command without an observation arriving", which is not a problem we have — every
  tick carries an observation. It costs the codec correlation: Codec.decode(data, context=obs) is free today
  because encode and decode happen in one call, and splitting the legs makes twelve stateless codecs stateful
  and forces request/response tagging.
• Wrappers as generators. Genuinely clearer for the two or three real state machines — _trajectory_end
  stops being a field and becomes where the generator is suspended, and reset is discard-and-rebuild. But it is
  an internal implementation style behind __call__, not a change to any contract, and pool.submit(inner, obs)
  already delegates a call to a worker without it. Adoptable per wrapper, later, on its own merits.
• A next_due that lets the harness skip building observations. Defeated by its own premise: StopOnFault
  sits outermost in every shipped pipeline (cfg/wrappers.video_context_wrappers is
  StopOnFault() | stack | ChunkedSchedule()) and always answers "next tick", so the value never survives to
  the top. next_due remains useful for pacing (above), not for skipping work.
• A safety layer the harness enforces regardless of the declared stack — stopping the arm on a fault
  without StopOnFault, refusing an observation past a staleness bound. Declined. The declared stack is the
  mechanism, and StopOnFault is how the arm stops: a server that omits it has declared a stack that does not
  stop, and the harness will not second-guess that. The harness's job on a fault is delivery — the stack must
  see every fault, including one that starts and clears inside an inference window (Goal 1) — not action.
  The trial deadline is unaffected: it is the trial's budget, not the algorithm's, and the harness already
  ends the trial on it.

[vertix]

Review of #591 sharpened how long a faulted arm can go unnoticed, so recording it here.

Harness._step returns before _build_obs whenever _take answers PENDING:

if self._future is not None and self._take(self._future, clock) is Harness._Answer.PENDING:
    return

No observation is built on such a round, so keys.ROBOT_FAULT never reaches the declared stack and StopOnFault cannot answer the empty trajectory. _play keeps emitting the schedule already installed.

On a real rig the world clock is the wall clock, so ahead in _take computes to ~0 and every round with a call still in flight returns PENDING. A fault arriving just after a submit is therefore invisible for the whole call — tens to hundreds of milliseconds for a real model, unbounded for one that hangs. Under a fixed latency the release window does the same.

This is not a regression: upstream/main ran inference inline on the loop thread, so the loop was inside the model for the call's duration and equally blind, with each driver's own TrajectoryPlayer still playing. The floor described in this issue is what closes it in either design — the harness clearing the players on a fault it sees itself, rather than the arm's stopping depending on which wrappers the declared stack happens to carry.

[vertix]

A second finding from #591's review, sharper than the latency one above: a short fault is not merely seen late, it is not seen at all.

FrankaState is a pimm.shared_memory.NumpySMAdapter, so the arm's state crosses to the harness through a shared buffer rather than a queue. The status is a level in that buffer, and encode rewrites it every driver round:

self.array[FrankaState.STATUS_OFFSET] = (
    RobotStatus.AVAILABLE.value if state.error == 0 else RobotStatus.ERROR.value
)

The Franka driver runs that loop at 2000 Hz (pimm.RateLimiter(clock, hz=2000)); the harness samples at up to 100 Hz (POLL_PERIOD_SEC = 0.01). A reflex the driver clears within one harness poll therefore leaves no trace the harness can read — not probabilistically, but by construction, since a shared buffer keeps only the newest write. StopOnFault never fires and the schedule the harness planned before the fault keeps playing into an arm the reflex has moved.

Before #591 each driver held the schedule itself and dropped it on its own error, with no harness involvement:

if in_error:
    robot.recover_from_errors()
    # Drop the in-flight trajectory so the arm holds position rather than resuming a stale
    # waypoint once the error clears.
    player.set([])

That guard was local and unconditional. With the schedule owned by the harness there is no local equivalent: the driver cannot invalidate a plan it does not hold, and the harness cannot act on a fault it cannot sample.

Closing it means making the fault survive sampling rather than making the harness poll faster — latching it until the harness acknowledges, or carrying it as a monotonic fault epoch in the state so a reader detects the change whatever its rate. Either touches State and every arm driver, which is why it belongs to this issue's floor rather than to a review round. It also overlaps #619, where the fault already travels stapled on beside robot_state instead of out of the serializer.