From 1e77545910f6d70f8f0ccbfcf46bfdd4efae2726 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Tue, 8 Sep 2026 21:45:06 +0000 Subject: [PATCH 1/5] Count the waypoints a late round drops, and how late the one it sends is The loop calls the policy and plays the trajectory in one round, so a round longer than the control period pops every waypoint that came due and emits only the newest. It discarded the rest silently: nothing counted them and nothing logged them. Each command channel now keeps its own account of the episode - the waypoints the schedule took, the rounds that sent one, the waypoints a round overtook, and a histogram of how far past its own due time each emitted waypoint went out. The episode's static meta carries it under 'eval.schedule..*', beside the other eval facts a rollout report reads. The emit behaviour does not change. The arm goes to the newest setpoint, and this change measures that rather than redesigns it. Ticket: none - instrumentation asked for in chat; no ticket exists for it --- positronic/eval/keys.py | 17 +++- positronic/policy/harness.py | 71 +++++++++++++- positronic/policy/tests/test_harness.py | 122 ++++++++++++++++++++++++ 3 files changed, 206 insertions(+), 4 deletions(-) diff --git a/positronic/eval/keys.py b/positronic/eval/keys.py index a063c7d37..0ea53c77b 100644 --- a/positronic/eval/keys.py +++ b/positronic/eval/keys.py @@ -1,4 +1,5 @@ -"""The keys a trial writes: what it readies, the conditions it runs under and the verdict it ends on.""" +"""The keys a trial writes: what it readies, the conditions it runs under, how well the loop kept its +schedule, and the verdict it ends on.""" # The names of what a trial readies before it opens. ``Embodiment.prepare_handlers`` is keyed by them, and so # is what a ``Task`` asks for. A rig with two arms names its arms ``arm.{side}``. ``SCENE`` means the world @@ -41,3 +42,17 @@ # The id of the task a trial runs, as the benchmark names it. The episode records it; positronic never reads # inside it. TASK = 'eval.task' + +# How well the loop kept the trajectory's schedule, per command channel, over the episode. A reader composes +# the prefix, a channel and a field: f'{SCHEDULE}.{keys.ROBOT_COMMAND}.{DROPPED}'. ``SCHEDULED`` counts every +# waypoint the channel's schedule took, ``EMITTED`` the rounds that sent one and ``DROPPED`` the waypoints a +# round overtook, so the rest is what a fresh chunk replaced before it came due. The lateness figures measure +# an emitted waypoint against its own due time, in milliseconds; the percentiles are binned to the whole +# millisecond, and only ``LATE_MAX_MS`` is exact. +SCHEDULE = 'eval.schedule' +SCHEDULED = 'scheduled' +EMITTED = 'emitted' +DROPPED = 'dropped' +LATE_P50_MS = 'late_p50_ms' +LATE_P90_MS = 'late_p90_ms' +LATE_MAX_MS = 'late_max_ms' diff --git a/positronic/policy/harness.py b/positronic/policy/harness.py index e40f3f482..645ce77ec 100644 --- a/positronic/policy/harness.py +++ b/positronic/policy/harness.py @@ -163,6 +163,62 @@ def _end_span(self) -> None: self._span = None +# Lateness is binned to the whole millisecond up to this bound. Anything later shares the last bin, where +# only the running maximum still separates it, so the bound must outlast the longest round worth resolving. +_LATE_BINS_MS = 1000 + + +class _ScheduleFidelity: + """How well one episode's loop played the waypoints scheduled on one command channel. + + A round emits only the newest waypoint that has come due, so a round longer than the control period + discards every earlier one it overtook. This counts those against what the schedule took, and bins how + far past its own due time each emitted waypoint went out. + """ + + def __init__(self) -> None: + self._scheduled = 0 + self._emitted = 0 + self._dropped = 0 + self._max_late_ns = 0 + self._late_bins = [0] * (_LATE_BINS_MS + 1) + + def count_scheduled(self, waypoints: int) -> None: + self._scheduled += waypoints + + def count_played(self, popped: int, late_ns: int) -> None: + """Record a round that emitted the newest of ``popped`` due waypoints, ``late_ns`` past its due time.""" + self._emitted += 1 + self._dropped += popped - 1 + self._max_late_ns = max(self._max_late_ns, late_ns) + self._late_bins[min(late_ns // 1_000_000, _LATE_BINS_MS)] += 1 + + def meta(self, prefix: str) -> dict[str, Any]: + """The episode's account under ``prefix``, empty for a channel the trajectory never named.""" + if self._scheduled == 0: + return {} + meta: dict[str, Any] = { + f'{prefix}.{eval_keys.SCHEDULED}': self._scheduled, + f'{prefix}.{eval_keys.EMITTED}': self._emitted, + f'{prefix}.{eval_keys.DROPPED}': self._dropped, + } + if self._emitted > 0: + meta[f'{prefix}.{eval_keys.LATE_P50_MS}'] = self._late_percentile_ms(50) + meta[f'{prefix}.{eval_keys.LATE_P90_MS}'] = self._late_percentile_ms(90) + meta[f'{prefix}.{eval_keys.LATE_MAX_MS}'] = self._max_late_ns / 1e6 + return meta + + def _late_percentile_ms(self, percent: int) -> float: + """The whole millisecond at or under which ``percent`` of the emitted waypoints went out.""" + rank = -(-self._emitted * percent // 100) # nearest-rank, in integers, so no float rounds it off by one + seen = 0 + for ms, count in enumerate(self._late_bins): + seen += count + if seen >= rank: + return float(ms) + raise AssertionError('every emitted waypoint is binned, so a rank within the count is always reached') + + class Harness(pimm.ControlSystem): """Control system that runs the episode lifecycle and plays the policy's trajectory to the drivers. @@ -195,6 +251,8 @@ def __init__(self, embodiment: Embodiment, *, static_meta: dict[str, Any] | None self.prepare = pimm.calls.CallerDict[Any, None](self, names=embodiment.prepare_handlers) # Each channel's waypoints not yet played, stamped with absolute clock ns and ascending. self._schedules: dict[str, deque[tuple[int, Any]]] = {name: deque() for name in embodiment.commands} + # How the live episode has played those schedules, rebuilt per episode and stamped into its statics. + self._fidelity = {name: _ScheduleFidelity() for name in embodiment.commands} # One episode per call, answered with the terminal payload it ended on. self.perform_task = pimm.calls.ControlSystemHandler[Rollout, dict[str, Any]](self) @@ -231,6 +289,8 @@ def _build_episode_meta(self) -> dict[str, Any]: assert self._inference is not None, 'only a live episode has meta' for k, v in flatten_dict(self._inference.meta).items(): meta[f'{policy_keys.POLICY_META}.{k}'] = v + for name, fidelity in self._fidelity.items(): + meta.update(fidelity.meta(f'{eval_keys.SCHEDULE}.{name}')) meta.update(self._task.meta) meta[keys.TASK] = self._task.instruction return meta @@ -294,6 +354,7 @@ def _begin_episode( # it, and still closes the session it was handed. self._call = call self._inference = _EpisodeInference(call.request, self._charges_wall_time, clock) + self._fidelity = {name: _ScheduleFidelity() for name in self._embodiment.commands} # The episode span opens first, so the prepare and the rollout's other phase spans parent to it. self._telemetry.begin(self._task.meta) with telemetry.span(telemetry_keys.SPAN_RESET): @@ -379,16 +440,20 @@ def _reschedule(self, trajectory: list[dict[str, Any]], clock: pimm.Clock) -> No for name, schedule in self._schedules.items(): schedule.clear() schedule.extend((int(a[keys.ACTION_TIMESTAMP] * 1e9), a[name]) for a in trajectory if name in a) + self._fidelity[name].count_scheduled(len(schedule)) def _issue_due_commands(self, clock: pimm.Clock) -> None: - """Emit each channel's due command. Nothing on a channel with none. Last on a channel with multiple.""" + """Emit each channel's due command. Nothing on a channel with none. Last on a channel with multiple, + counting the ones it overtakes as dropped and how late the one it sends is.""" now_ns = clock.now_ns() for name, schedule in self._schedules.items(): - value = None + due_ns, value, popped = now_ns, None, 0 while schedule and schedule[0][0] <= now_ns: - value = schedule.popleft()[1] + due_ns, value = schedule.popleft() + popped += 1 if value is not None: self.commands[name].emit(value) + self._fidelity[name].count_played(popped, now_ns - due_ns) def _trial_terminal(self, done: pimm.Message[dict] | None, clock: pimm.Clock) -> dict[str, Any] | None: """The terminal static payload if the live trial has ended this round, else ``None``. diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py index b05bed4c8..13b59851f 100644 --- a/positronic/policy/tests/test_harness.py +++ b/positronic/policy/tests/test_harness.py @@ -1689,6 +1689,128 @@ def test_harness_clears_trajectory_on_run(world): assert _last_grip(p) >= 200.0, 'Expected chunk 2; trajectory clearing on a new episode failed' +class _FrozenClock(pimm.Clock): + """A clock stopped at an exact nanosecond, so a waypoint scheduled on a whole millisecond is not moved off + it by a float.""" + + def __init__(self, now_ns: int): + self._now_ns = now_ns + + def now(self) -> float: + return self._now_ns / 1e9 + + def now_ns(self) -> int: + return self._now_ns + + +def _schedule_key(channel: str, field: str) -> str: + """One field of one channel's schedule account, keyed as the episode's statics carry it.""" + return f'{eval_keys.SCHEDULE}.{channel}.{field}' + + +def _schedule_account(harness: Harness, channel: str) -> dict[str, Any]: + """What the harness would stamp for ``channel``, without ending an episode to read it.""" + return harness._fidelity[channel].meta(f'{eval_keys.SCHEDULE}.{channel}') + + +def _play_round(harness: Harness, due_ms: list[int], now_ms: int, channel: str = keys.ROBOT_COMMAND) -> None: + """Schedule a waypoint on ``channel`` for each of ``due_ms``, then run one round of the loop at ``now_ms``.""" + harness._schedules[channel].extend((ms * 1_000_000, f'waypoint@{ms}ms') for ms in due_ms) + harness._fidelity[channel].count_scheduled(len(due_ms)) + harness._issue_due_commands(_FrozenClock(now_ms * 1_000_000)) + + +def _harness_recording_commands() -> tuple[Harness, RecordingEmitter]: + harness = Harness(make_embodiment()) + recorder = RecordingEmitter() + harness.commands[keys.ROBOT_COMMAND]._bind(recorder) + return harness, recorder + + +@pytest.mark.timeout(3.0) +def test_a_round_that_overtakes_waypoints_counts_them_dropped(): + """A round late enough to find three waypoints due sends the newest and counts the two it overtook.""" + harness, recorder = _harness_recording_commands() + + _play_round(harness, due_ms=[0, 10, 20], now_ms=25) + + assert _emitted_commands(recorder) == ['waypoint@20ms'] + account = _schedule_account(harness, keys.ROBOT_COMMAND) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.SCHEDULED)] == 3 + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.EMITTED)] == 1 + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.DROPPED)] == 2 + + +@pytest.mark.timeout(3.0) +def test_a_round_that_finds_one_waypoint_due_counts_no_drop(): + """A round that keeps up overtakes nothing: the two waypoints still ahead of it are not dropped.""" + harness, recorder = _harness_recording_commands() + + _play_round(harness, due_ms=[0, 10, 20], now_ms=5) + + assert _emitted_commands(recorder) == ['waypoint@0ms'] + account = _schedule_account(harness, keys.ROBOT_COMMAND) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.EMITTED)] == 1 + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.DROPPED)] == 0 + + +@pytest.mark.timeout(3.0) +def test_lateness_is_measured_against_the_waypoint_that_went_out(): + """The waypoint sent at 25 ms was due at 20, so it is 5 ms late — not the 25 the round is past the oldest + waypoint it overtook.""" + harness, _ = _harness_recording_commands() + + _play_round(harness, due_ms=[0, 10, 20], now_ms=25) + + account = _schedule_account(harness, keys.ROBOT_COMMAND) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.LATE_MAX_MS)] == pytest.approx(5.0) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.LATE_P50_MS)] == pytest.approx(5.0) + + +@pytest.mark.timeout(3.0) +def test_the_lateness_percentiles_read_the_spread_of_the_rounds(): + """Ten rounds, one waypoint each, 0 to 9 ms late: the percentiles rank them rather than report the worst.""" + harness, _ = _harness_recording_commands() + + for late_ms in range(10): + _play_round(harness, due_ms=[100 * late_ms], now_ms=100 * late_ms + late_ms) + + account = _schedule_account(harness, keys.ROBOT_COMMAND) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.LATE_P50_MS)] == pytest.approx(4.0) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.LATE_P90_MS)] == pytest.approx(8.0) + assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.LATE_MAX_MS)] == pytest.approx(9.0) + + +@pytest.mark.timeout(3.0) +def test_the_schedule_account_reaches_the_episode_meta(world): + """A finished episode's statics say, per command channel, what its loop scheduled, sent and dropped.""" + policy = ChunkPolicy() + harness = Harness(make_embodiment()) + p = _pair_all(world, harness, policy) + robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6]) + + driver = ManualDriver([ + (partial(p['perform_task'], Task(instruction_source='test', timeout_sec=None)), 0.0), + (partial(emit_ready_payload, p['frame_em'], p['robot_em'], p['grip_em'], robot_state), 0.01), + (None, 0.05), + (partial(p['done_em'].emit, OPERATOR_DONE), 0.0), + (None, 0.02), + ]) + scheduler = world.start([harness, driver]) + drive_scheduler(scheduler, steps=200) + + stops = [c for c in _ds_commands(p) if c.type == DsWriterCommandType.STOP_EPISODE] + assert len(stops) == 1 + meta = stops[0].static_data + for channel in (keys.ROBOT_COMMAND, keys.TARGET_GRIP): + scheduled = meta[_schedule_key(channel, eval_keys.SCHEDULED)] + emitted = meta[_schedule_key(channel, eval_keys.EMITTED)] + assert emitted > 0, f'{channel} played no waypoint' + assert emitted + meta[_schedule_key(channel, eval_keys.DROPPED)] <= scheduled + assert meta[_schedule_key(channel, eval_keys.LATE_MAX_MS)] >= 0.0 + assert meta[_schedule_key(channel, eval_keys.LATE_P90_MS)] >= 0.0 + + @pytest.mark.timeout(3.0) @pytest.mark.parametrize('unavailable', [RobotStatus.BUSY, RobotStatus.ERROR]) def test_the_stack_keeps_the_model_away_from_an_unavailable_arm(world, unavailable): From e61cdd50f5e867c8a1c2ce083fb470c10da23eb3 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Tue, 8 Sep 2026 21:54:58 +0000 Subject: [PATCH 2/5] Stamp the waypoint account on the episode span, and report it The episode statics carry the per-channel account. A timing report reads spans, not statics, and that is where a reader already goes to ask where the loop's time went. The episode span now carries the same account totalled over its command channels, and 'positronic eval timing-report' prints what the pass scheduled, sent and dropped, the drop share of what came due, and the mean and worst lateness. Lateness rides on the span as a sum and a maximum, because those total across episodes exactly; a percentile does not, so the distribution stays per channel in the episode's statics. The two sinks carry one measurement. Telemetry is opt-in - 'telemetry.bind_from_env' is inert unless POSITRONIC_ENV_TELEMETRY_DIR is set - so a customer rollout writes no span at all, and the episode record is what makes the account unconditional. Ticket: none - instrumentation asked for in chat; no ticket exists for it --- .../cli/eval/tests/test_timing_report.py | 84 +++++++++++++++++++ positronic/cli/eval/timing_report.py | 64 ++++++++++++++ positronic/policy/harness.py | 49 ++++++++--- positronic/policy/tests/test_harness.py | 39 +++++++++ positronic/telemetry_keys.py | 10 +++ 5 files changed, 236 insertions(+), 10 deletions(-) diff --git a/positronic/cli/eval/tests/test_timing_report.py b/positronic/cli/eval/tests/test_timing_report.py index d577b25ee..1527f9349 100644 --- a/positronic/cli/eval/tests/test_timing_report.py +++ b/positronic/cli/eval/tests/test_timing_report.py @@ -28,6 +28,11 @@ ) from positronic.telemetry_keys import ( ATTR_EPISODE_VIRTUAL_S, + ATTR_WAYPOINTS_DROPPED, + ATTR_WAYPOINTS_EMITTED, + ATTR_WAYPOINTS_LATE_MAX_MS, + ATTR_WAYPOINTS_LATE_SUM_MS, + ATTR_WAYPOINTS_SCHEDULED, HARNESS_PROCESS, SPAN_ENV_STEP, SPAN_EPISODE, @@ -685,3 +690,82 @@ def test_a_telemetry_directory_holding_no_spans_is_not_reported_as_untimed(tmp_p (tmp_path / TELEMETRY_SUBDIR).mkdir() with pytest.raises(ValueError, match='carries no spans'): timing_report(run_dir=str(tmp_path)) + + +def _waypoint_fixture(telemetry_dir): + """One pass, two episodes carrying waypoint accounts that differ in every field, so a sum, a share, an + emission-weighted mean and a maximum are each told apart from the others.""" + telemetry_dir.mkdir() + _write_lines( + telemetry_dir / f'{HARNESS_PROCESS}{SPANS_SUFFIX}', + [ + _span(SPAN_EVAL_PASS, 0, 100, 'pass0'), + _span( + SPAN_EPISODE, + 0, + 40, + 'ep0', + 'pass0', + { + ATTR_EPISODE_VIRTUAL_S: 20.0, + ATTR_WAYPOINTS_SCHEDULED: 100, + ATTR_WAYPOINTS_EMITTED: 10, + ATTR_WAYPOINTS_DROPPED: 30, + ATTR_WAYPOINTS_LATE_SUM_MS: 50.0, + ATTR_WAYPOINTS_LATE_MAX_MS: 20.0, + }, + ), + _span( + SPAN_EPISODE, + 50, + 90, + 'ep1', + 'pass0', + { + ATTR_EPISODE_VIRTUAL_S: 20.0, + ATTR_WAYPOINTS_SCHEDULED: 100, + ATTR_WAYPOINTS_EMITTED: 30, + ATTR_WAYPOINTS_DROPPED: 10, + ATTR_WAYPOINTS_LATE_SUM_MS: 30.0, + ATTR_WAYPOINTS_LATE_MAX_MS: 12.0, + }, + ), + ], + ) + + +def test_the_waypoint_account_sums_over_the_pass(tmp_path): + """Counts add, the drop share is of what came due rather than of what was scheduled, the lateness mean is + the pass's own emissions divided into the pass's own sum, and the maximum is the worse episode's.""" + _waypoint_fixture(tmp_path / TELEMETRY_SUBDIR) + report = _build_report(_read_spans_dir(tmp_path / TELEMETRY_SUBDIR), [], policy_gpu=None) + + assert report.waypoints is not None + waypoints = report.waypoints + assert waypoints.scheduled == 200 + assert waypoints.emitted == 40 + assert waypoints.dropped == 40 + assert waypoints.dropped_share == pytest.approx(0.5) # 40 of the 80 that came due, not of the 200 scheduled + assert waypoints.mean_late_ms == pytest.approx(2.0) # 80 ms over 40 emissions, so the busier episode weighs more + assert waypoints.max_late_ms == pytest.approx(20.0) + + +def test_render_shows_the_waypoint_account(tmp_path): + _waypoint_fixture(tmp_path / TELEMETRY_SUBDIR) + report = _build_report(_read_spans_dir(tmp_path / TELEMETRY_SUBDIR), [], policy_gpu=None) + + rendered = _render(report).splitlines() + + assert 'waypoints: 200 scheduled, 40 emitted, 40 dropped' in rendered + assert 'waypoint drops: 50.0% of the waypoints that came due' in rendered + assert 'waypoint late mean: 2.0 ms (max 20.0)' in rendered + + +def test_a_pass_whose_episodes_carry_no_waypoint_account_reports_none(tmp_path): + """A run that scheduled no waypoint reduces to no waypoint block at all, rather than to a row of zeros + that reads as a loop keeping perfect time.""" + _fixture(tmp_path / TELEMETRY_SUBDIR) + report = _build_report(_read_spans_dir(tmp_path / TELEMETRY_SUBDIR), [], policy_gpu=None) + + assert report.waypoints is None + assert not any(line.startswith('waypoint') for line in _render(report).splitlines()) diff --git a/positronic/cli/eval/timing_report.py b/positronic/cli/eval/timing_report.py index 78f138c52..14af921f3 100644 --- a/positronic/cli/eval/timing_report.py +++ b/positronic/cli/eval/timing_report.py @@ -40,6 +40,11 @@ ATTR_EPISODE_PARTIAL, ATTR_EPISODE_VIRTUAL_S, ATTR_PASS_FAILED, + ATTR_WAYPOINTS_DROPPED, + ATTR_WAYPOINTS_EMITTED, + ATTR_WAYPOINTS_LATE_MAX_MS, + ATTR_WAYPOINTS_LATE_SUM_MS, + ATTR_WAYPOINTS_SCHEDULED, HARNESS_PROCESS, SPAN_ENV_STEP, SPAN_EPISODE, @@ -146,6 +151,27 @@ class EnvStepSplit: materialize: float +@dataclass +class WaypointReport: + """How well the loop kept the trajectory's schedule, summed over the pass's episodes and their command + channels. + + A round emits only the newest waypoint that has come due, so ``dropped`` counts the ones it overtook and + ``emitted + dropped`` is what came due; ``dropped_share`` is a fraction of that, not of ``scheduled``, + which also covers what a fresh chunk replaced before its time. The lateness figures measure an emitted + waypoint against its own due time, and both are exact over the pass because each episode carries a sum + and a maximum rather than a percentile. The distribution behind them is per channel in the episode's own + statics; a percentile of the pass is not recoverable from here. + """ + + scheduled: int + emitted: int + dropped: int + dropped_share: float + mean_late_ms: float + max_late_ms: float + + @dataclass class PassReport: """Pass-level wall-clock roll-up reduced from the recorded telemetry spans and stats. @@ -164,6 +190,7 @@ class PassReport: infer_p95_ms: float wall_split: WallSplit env_step_split: EnvStepSplit | None + waypoints: WaypointReport | None gpu: GpuReport @@ -377,6 +404,11 @@ class _EpisodeTiming: policy_wait_s: float overhead_s: float infer_ms: list[float] + waypoints_scheduled: int + waypoints_emitted: int + waypoints_dropped: int + late_sum_ms: float + late_max_ms: float def _episode_timing(episode: SpanRec, children: dict[str, list[SpanRec]]) -> _EpisodeTiming: @@ -405,6 +437,30 @@ def _episode_timing(episode: SpanRec, children: dict[str, list[SpanRec]]) -> _Ep policy_wait_s=policy_wait_s, overhead_s=max(wall_s - measured, 0.0), infer_ms=infer_ms, + waypoints_scheduled=int(episode.attrs.get(ATTR_WAYPOINTS_SCHEDULED, 0)), + waypoints_emitted=int(episode.attrs.get(ATTR_WAYPOINTS_EMITTED, 0)), + waypoints_dropped=int(episode.attrs.get(ATTR_WAYPOINTS_DROPPED, 0)), + late_sum_ms=float(episode.attrs.get(ATTR_WAYPOINTS_LATE_SUM_MS, 0.0)), + late_max_ms=float(episode.attrs.get(ATTR_WAYPOINTS_LATE_MAX_MS, 0.0)), + ) + + +def _waypoint_report(timings: list[_EpisodeTiming]) -> WaypointReport | None: + """The pass's waypoint account, or ``None`` where no episode carries one — a run that played no + trajectory, and a sidecar whose episodes hold no waypoint attributes, reduce the same way.""" + scheduled = sum(t.waypoints_scheduled for t in timings) + if not scheduled: + return None + emitted = sum(t.waypoints_emitted for t in timings) + dropped = sum(t.waypoints_dropped for t in timings) + due = emitted + dropped + return WaypointReport( + scheduled=scheduled, + emitted=emitted, + dropped=dropped, + dropped_share=(dropped / due) if due else 0.0, + mean_late_ms=(sum(t.late_sum_ms for t in timings) / emitted) if emitted else 0.0, + max_late_ms=max((t.late_max_ms for t in timings), default=0.0), ) @@ -535,6 +591,7 @@ def phase_fraction(total: float) -> float: infer_p95_ms=float(np.percentile(all_infer_ms, 95)) if all_infer_ms.size else 0.0, wall_split=wall_split, env_step_split=_env_step_split(spans, episodes, env_step_sum, materialize_sum), + waypoints=_waypoint_report(timings), gpu=GpuReport(sim=_gpu_summary_from_stats(stats, list(windows.values())), policy=policy_gpu), ) @@ -580,6 +637,13 @@ def _render(report: PassReport) -> str: lines += [_share_row(name, frac) for name, frac in split.phases.items()] lines.append(_share_row('wire', split.wire)) lines.append(_share_row('materialize', split.materialize)) + if report.waypoints is not None: + way = report.waypoints + lines += [ + f'waypoints: {way.scheduled} scheduled, {way.emitted} emitted, {way.dropped} dropped', + f'waypoint drops: {way.dropped_share * 100:>6.1f}% of the waypoints that came due', + f'waypoint late mean: {way.mean_late_ms:.1f} ms (max {way.max_late_ms:.1f})', + ] for f in fields(GpuReport): summary = getattr(report.gpu, f.name) if summary is not None: diff --git a/positronic/policy/harness.py b/positronic/policy/harness.py index 645ce77ec..8c1a0bf55 100644 --- a/positronic/policy/harness.py +++ b/positronic/policy/harness.py @@ -131,29 +131,29 @@ def start_rollout(self, virtual_now: float) -> None: def step(self) -> None: self._steps += 1 - def end(self, virtual_now: float) -> None: - """Close the rollout, stamped with its step count and its virtual duration up to ``virtual_now`` — - captured when the rollout ended, before the flush round advances the sim clock.""" + def end(self, virtual_now: float, waypoint_attrs: dict[str, Any]) -> None: + """Close the rollout, stamped with its step count, its waypoint account and its virtual duration up to + ``virtual_now`` — captured when the rollout ended, before the flush round advances the sim clock.""" if self._span is None: return - self._close(virtual_now) + self._close(virtual_now, waypoint_attrs) telemetry.force_flush() - def seal(self, virtual_now: float) -> None: + def seal(self, virtual_now: float, waypoint_attrs: dict[str, Any]) -> None: """Close a rollout abandoned mid-flight by a raising ``reset`` / ``new_session`` / session call, marked ``episode.partial`` so the reduce keeps it. Ending it is what exports it: the batch processor drops an unended span, orphaning the finished children and losing their phases.""" if self._span is None: return telemetry.set_attrs(self._span, **{telemetry_keys.ATTR_EPISODE_PARTIAL: True}) - self.end(virtual_now) + self.end(virtual_now, waypoint_attrs) - def _close(self, virtual_now: float) -> None: + def _close(self, virtual_now: float, waypoint_attrs: dict[str, Any]) -> None: # A rollout that never started — a prepare that raised — has zero virtual duration. virtual_s = max(virtual_now - self._virtual_start, 0.0) if self._virtual_start is not None else 0.0 assert self._span is not None attrs = {telemetry_keys.ATTR_EPISODE_STEPS: self._steps, telemetry_keys.ATTR_EPISODE_VIRTUAL_S: virtual_s} - telemetry.set_attrs(self._span, **attrs) + telemetry.set_attrs(self._span, **attrs, **waypoint_attrs) self._end_span() def _end_span(self) -> None: @@ -180,6 +180,7 @@ def __init__(self) -> None: self._scheduled = 0 self._emitted = 0 self._dropped = 0 + self._late_sum_ns = 0 self._max_late_ns = 0 self._late_bins = [0] * (_LATE_BINS_MS + 1) @@ -190,9 +191,30 @@ def count_played(self, popped: int, late_ns: int) -> None: """Record a round that emitted the newest of ``popped`` due waypoints, ``late_ns`` past its due time.""" self._emitted += 1 self._dropped += popped - 1 + self._late_sum_ns += late_ns self._max_late_ns = max(self._max_late_ns, late_ns) self._late_bins[min(late_ns // 1_000_000, _LATE_BINS_MS)] += 1 + def merge(self, other: '_ScheduleFidelity') -> None: + """Fold another channel's account into this one, so an episode's totals read off a single account.""" + self._scheduled += other._scheduled + self._emitted += other._emitted + self._dropped += other._dropped + self._late_sum_ns += other._late_sum_ns + self._max_late_ns = max(self._max_late_ns, other._max_late_ns) + self._late_bins = [own + theirs for own, theirs in zip(self._late_bins, other._late_bins, strict=True)] + + def span_attrs(self) -> dict[str, Any]: + """The account as episode-span attributes. Lateness rides as a sum and a maximum, which the offline + reduce totals across episodes exactly, where a percentile would not survive being averaged.""" + return { + telemetry_keys.ATTR_WAYPOINTS_SCHEDULED: self._scheduled, + telemetry_keys.ATTR_WAYPOINTS_EMITTED: self._emitted, + telemetry_keys.ATTR_WAYPOINTS_DROPPED: self._dropped, + telemetry_keys.ATTR_WAYPOINTS_LATE_SUM_MS: self._late_sum_ns / 1e6, + telemetry_keys.ATTR_WAYPOINTS_LATE_MAX_MS: self._max_late_ns / 1e6, + } + def meta(self, prefix: str) -> dict[str, Any]: """The episode's account under ``prefix``, empty for a channel the trajectory never named.""" if self._scheduled == 0: @@ -295,6 +317,13 @@ def _build_episode_meta(self) -> dict[str, Any]: meta[keys.TASK] = self._task.instruction return meta + def _episode_waypoints(self) -> _ScheduleFidelity: + """Every command channel's account for this episode, in one.""" + total = _ScheduleFidelity() + for fidelity in self._fidelity.values(): + total.merge(fidelity) + return total + def _ready( self, should_stop: pimm.SignalReceiver, clock: pimm.Clock, args: dict[str, Any] ) -> Generator[pimm.Command, None, None]: @@ -338,7 +367,7 @@ def _finalize_recording( # The episode span must still be open while the recorder writes the STOP, so that write is timed # inside the episode. The other control systems run in that round as well, and the episode is timed # with their work too. The error is not more than one control period. - self._telemetry.end(virtual_now) + self._telemetry.end(virtual_now, self._episode_waypoints().span_attrs()) def _set_deadline(self, deadline_ns: int | None) -> None: """Arm the live episode's deadline and publish it, so the enforced one and the published one agree.""" @@ -471,7 +500,7 @@ def run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[p try: yield from self._run(should_stop, clock) except BaseException as exc: - self._telemetry.seal(clock.now()) + self._telemetry.seal(clock.now(), self._episode_waypoints().span_attrs()) if self._call is not None: self._call.set_exception(exc) self._call = None diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py index 13b59851f..93a63c05e 100644 --- a/positronic/policy/tests/test_harness.py +++ b/positronic/policy/tests/test_harness.py @@ -2658,3 +2658,42 @@ def test_finishing_discards_a_call_that_is_still_in_flight(world): drive_scheduler(world.start([harness, driver, _Pacer()]), steps=2000) assert not _emitted_commands(cmd_recorder) + + +@pytest.mark.timeout(5.0) +def test_the_episode_span_carries_the_same_waypoint_account_as_the_meta(world, tmp_path): + """Under ``telemetry.bind`` the episode span carries the waypoint totals over every command channel, and + they are the episode statics' own per-channel figures added up: one measurement, two sinks.""" + policy = ChunkPolicy() + harness = Harness(make_embodiment()) + p = _pair_all(world, harness, policy) + robot_state = make_robot_state([0.1, 0.2, 0.3], [0.4, 0.5, 0.6]) + driver = ManualDriver([ + (partial(p['perform_task'], Task(instruction_source='t', timeout_sec=None)), 0.0), + (partial(emit_ready_payload, p['frame_em'], p['robot_em'], p['grip_em'], robot_state), 0.01), + (None, 0.05), + (partial(p['done_em'].emit, OPERATOR_DONE), 0.0), + (None, 0.02), + ]) + + with telemetry.bind(tmp_path, telemetry_keys.HARNESS_PROCESS, 'run-waypoints'), _eval_pass('run-waypoints'): + drive_scheduler(world.start([harness, driver]), steps=200) + + spans = list(telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS))) + episodes = [s for s in spans if s.name == telemetry_keys.SPAN_EPISODE] + assert len(episodes) == 1 + attrs = episodes[0].attrs + assert attrs[telemetry_keys.ATTR_WAYPOINTS_EMITTED] > 0, 'the episode played no waypoint' + assert attrs[telemetry_keys.ATTR_WAYPOINTS_LATE_SUM_MS] >= 0.0 + assert attrs[telemetry_keys.ATTR_WAYPOINTS_LATE_MAX_MS] >= 0.0 + + stops = [c for c in _ds_commands(p) if c.type == DsWriterCommandType.STOP_EPISODE] + assert len(stops) == 1 + meta = stops[0].static_data + channels = (keys.ROBOT_COMMAND, keys.TARGET_GRIP) + for attr, field in ( + (telemetry_keys.ATTR_WAYPOINTS_SCHEDULED, eval_keys.SCHEDULED), + (telemetry_keys.ATTR_WAYPOINTS_EMITTED, eval_keys.EMITTED), + (telemetry_keys.ATTR_WAYPOINTS_DROPPED, eval_keys.DROPPED), + ): + assert attrs[attr] == sum(meta[_schedule_key(channel, field)] for channel in channels) diff --git a/positronic/telemetry_keys.py b/positronic/telemetry_keys.py index 08deed8b2..bb44e9b85 100644 --- a/positronic/telemetry_keys.py +++ b/positronic/telemetry_keys.py @@ -32,6 +32,16 @@ ATTR_EPISODE_PARTIAL = 'episode.partial' ATTR_PASS_FAILED = 'pass.failed' +# One episode's waypoint account, totalled over its command channels. A round emits only the newest waypoint +# that has come due, so ``DROPPED`` counts the ones it overtook and ``EMITTED + DROPPED`` is what came due; +# ``SCHEDULED`` also covers what a fresh chunk replaced before its time. Lateness rides as a sum and a maximum +# rather than a percentile, because those are what the reduce can total across episodes exactly. +ATTR_WAYPOINTS_SCHEDULED = 'episode.waypoints.scheduled' +ATTR_WAYPOINTS_EMITTED = 'episode.waypoints.emitted' +ATTR_WAYPOINTS_DROPPED = 'episode.waypoints.dropped' +ATTR_WAYPOINTS_LATE_SUM_MS = 'episode.waypoints.late_sum_ms' +ATTR_WAYPOINTS_LATE_MAX_MS = 'episode.waypoints.late_max_ms' + # The harness process's sidecar name — the discriminator between client-side spans (episode, client env.step) # and an env server's own file, which reduces rely on. HARNESS_PROCESS = 'harness' From dfc934b7cae8f55b72547548f0f585d2af023e33 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Tue, 8 Sep 2026 22:05:40 +0000 Subject: [PATCH 3/5] Trim the waypoint-account comments to one home each The sum-and-maximum rationale belongs with the span rendering that chooses it, and the eval and telemetry key comments carry only what a reader of the record needs. Ticket: none - a comment trim on the change above --- positronic/cli/eval/timing_report.py | 10 ++++------ positronic/eval/keys.py | 13 +++++-------- positronic/telemetry_keys.py | 7 +++---- 3 files changed, 12 insertions(+), 18 deletions(-) diff --git a/positronic/cli/eval/timing_report.py b/positronic/cli/eval/timing_report.py index 14af921f3..43fd89cf0 100644 --- a/positronic/cli/eval/timing_report.py +++ b/positronic/cli/eval/timing_report.py @@ -156,12 +156,10 @@ class WaypointReport: """How well the loop kept the trajectory's schedule, summed over the pass's episodes and their command channels. - A round emits only the newest waypoint that has come due, so ``dropped`` counts the ones it overtook and - ``emitted + dropped`` is what came due; ``dropped_share`` is a fraction of that, not of ``scheduled``, - which also covers what a fresh chunk replaced before its time. The lateness figures measure an emitted - waypoint against its own due time, and both are exact over the pass because each episode carries a sum - and a maximum rather than a percentile. The distribution behind them is per channel in the episode's own - statics; a percentile of the pass is not recoverable from here. + ``dropped`` counts a waypoint a round overtook, so ``dropped_share`` is a fraction of + ``emitted + dropped``, not of ``scheduled``, which also covers what a fresh chunk replaced first. The + distribution behind the lateness figures is per channel in the episode's own statics; a percentile of + the pass is not recoverable from here. """ scheduled: int diff --git a/positronic/eval/keys.py b/positronic/eval/keys.py index 0ea53c77b..0dff1675a 100644 --- a/positronic/eval/keys.py +++ b/positronic/eval/keys.py @@ -1,5 +1,4 @@ -"""The keys a trial writes: what it readies, the conditions it runs under, how well the loop kept its -schedule, and the verdict it ends on.""" +"""The keys a trial writes: what it readies, the conditions it runs under and the verdict it ends on.""" # The names of what a trial readies before it opens. ``Embodiment.prepare_handlers`` is keyed by them, and so # is what a ``Task`` asks for. A rig with two arms names its arms ``arm.{side}``. ``SCENE`` means the world @@ -43,12 +42,10 @@ # inside it. TASK = 'eval.task' -# How well the loop kept the trajectory's schedule, per command channel, over the episode. A reader composes -# the prefix, a channel and a field: f'{SCHEDULE}.{keys.ROBOT_COMMAND}.{DROPPED}'. ``SCHEDULED`` counts every -# waypoint the channel's schedule took, ``EMITTED`` the rounds that sent one and ``DROPPED`` the waypoints a -# round overtook, so the rest is what a fresh chunk replaced before it came due. The lateness figures measure -# an emitted waypoint against its own due time, in milliseconds; the percentiles are binned to the whole -# millisecond, and only ``LATE_MAX_MS`` is exact. +# How well the loop kept the trajectory's schedule, per command channel: a reader composes the prefix, a +# channel and a field, as f'{SCHEDULE}.{keys.ROBOT_COMMAND}.{DROPPED}'. ``DROPPED`` counts a waypoint a round +# overtook, so ``EMITTED + DROPPED`` is what came due and ``SCHEDULED`` also covers what a fresh chunk replaced +# first. The lateness figures are milliseconds against an emitted waypoint's own due time. SCHEDULE = 'eval.schedule' SCHEDULED = 'scheduled' EMITTED = 'emitted' diff --git a/positronic/telemetry_keys.py b/positronic/telemetry_keys.py index bb44e9b85..96b4b303d 100644 --- a/positronic/telemetry_keys.py +++ b/positronic/telemetry_keys.py @@ -32,10 +32,9 @@ ATTR_EPISODE_PARTIAL = 'episode.partial' ATTR_PASS_FAILED = 'pass.failed' -# One episode's waypoint account, totalled over its command channels. A round emits only the newest waypoint -# that has come due, so ``DROPPED`` counts the ones it overtook and ``EMITTED + DROPPED`` is what came due; -# ``SCHEDULED`` also covers what a fresh chunk replaced before its time. Lateness rides as a sum and a maximum -# rather than a percentile, because those are what the reduce can total across episodes exactly. +# One episode's waypoint account, totalled over its command channels. ``DROPPED`` counts a waypoint a round +# overtook, so ``EMITTED + DROPPED`` is what came due and ``SCHEDULED`` also covers what a fresh chunk +# replaced first. ATTR_WAYPOINTS_SCHEDULED = 'episode.waypoints.scheduled' ATTR_WAYPOINTS_EMITTED = 'episode.waypoints.emitted' ATTR_WAYPOINTS_DROPPED = 'episode.waypoints.dropped' From d28237171e1491ecb6f172823c4f9452ae62152d Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Tue, 8 Sep 2026 22:08:44 +0000 Subject: [PATCH 4/5] State the waypoint facts without denying a reading Ticket: none - a comment trim on the change above --- positronic/cli/eval/tests/test_timing_report.py | 8 ++++---- positronic/cli/eval/timing_report.py | 6 +++--- positronic/policy/tests/test_harness.py | 6 +++--- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/positronic/cli/eval/tests/test_timing_report.py b/positronic/cli/eval/tests/test_timing_report.py index 1527f9349..20f5f10b3 100644 --- a/positronic/cli/eval/tests/test_timing_report.py +++ b/positronic/cli/eval/tests/test_timing_report.py @@ -735,8 +735,8 @@ def _waypoint_fixture(telemetry_dir): def test_the_waypoint_account_sums_over_the_pass(tmp_path): - """Counts add, the drop share is of what came due rather than of what was scheduled, the lateness mean is - the pass's own emissions divided into the pass's own sum, and the maximum is the worse episode's.""" + """Counts add, the drop share is of what came due, the lateness mean is the pass's own emissions divided + into the pass's own sum, and the maximum is the worse episode's.""" _waypoint_fixture(tmp_path / TELEMETRY_SUBDIR) report = _build_report(_read_spans_dir(tmp_path / TELEMETRY_SUBDIR), [], policy_gpu=None) @@ -745,8 +745,8 @@ def test_the_waypoint_account_sums_over_the_pass(tmp_path): assert waypoints.scheduled == 200 assert waypoints.emitted == 40 assert waypoints.dropped == 40 - assert waypoints.dropped_share == pytest.approx(0.5) # 40 of the 80 that came due, not of the 200 scheduled - assert waypoints.mean_late_ms == pytest.approx(2.0) # 80 ms over 40 emissions, so the busier episode weighs more + assert waypoints.dropped_share == pytest.approx(0.5) # 40 of the 80 that came due + assert waypoints.mean_late_ms == pytest.approx(2.0) # 80 ms over the pass's own 40 emissions assert waypoints.max_late_ms == pytest.approx(20.0) diff --git a/positronic/cli/eval/timing_report.py b/positronic/cli/eval/timing_report.py index 43fd89cf0..4c9662494 100644 --- a/positronic/cli/eval/timing_report.py +++ b/positronic/cli/eval/timing_report.py @@ -157,9 +157,9 @@ class WaypointReport: channels. ``dropped`` counts a waypoint a round overtook, so ``dropped_share`` is a fraction of - ``emitted + dropped``, not of ``scheduled``, which also covers what a fresh chunk replaced first. The - distribution behind the lateness figures is per channel in the episode's own statics; a percentile of - the pass is not recoverable from here. + ``emitted + dropped``; ``scheduled`` also covers what a fresh chunk replaced first. The distribution + behind the lateness figures is per channel in the episode's own statics; a percentile of the pass is + not recoverable from here. """ scheduled: int diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py index 93a63c05e..464f643b0 100644 --- a/positronic/policy/tests/test_harness.py +++ b/positronic/policy/tests/test_harness.py @@ -1756,8 +1756,8 @@ def test_a_round_that_finds_one_waypoint_due_counts_no_drop(): @pytest.mark.timeout(3.0) def test_lateness_is_measured_against_the_waypoint_that_went_out(): - """The waypoint sent at 25 ms was due at 20, so it is 5 ms late — not the 25 the round is past the oldest - waypoint it overtook.""" + """The waypoint sent at 25 ms was due at 20, so it is 5 ms late. A round measured against the oldest + waypoint it overtook would read 25.""" harness, _ = _harness_recording_commands() _play_round(harness, due_ms=[0, 10, 20], now_ms=25) @@ -2663,7 +2663,7 @@ def test_finishing_discards_a_call_that_is_still_in_flight(world): @pytest.mark.timeout(5.0) def test_the_episode_span_carries_the_same_waypoint_account_as_the_meta(world, tmp_path): """Under ``telemetry.bind`` the episode span carries the waypoint totals over every command channel, and - they are the episode statics' own per-channel figures added up: one measurement, two sinks.""" + they are the episode statics' own per-channel figures added up.""" policy = ChunkPolicy() harness = Harness(make_embodiment()) p = _pair_all(world, harness, policy) From 097902c0d729e889cac2203b476f37ed00c00b87 Mon Sep 17 00:00:00 2001 From: Vladimir Yakunin Date: Wed, 9 Sep 2026 00:05:16 +0000 Subject: [PATCH 5/5] Count a due waypoint a fresh chunk replaces, and say what the report covers A chunk arriving on a late round replaces waypoints that had already come due. The round then plays the new schedule, so those go out nowhere and were counted neither emitted nor dropped - understating the drop rate exactly when the loop runs slow. '_reschedule' now counts the due leading run of each schedule as dropped before it clears it. The reduce reads only the episodes whose span carries an account, and reports how many did. A directory holding passes from either side of the account's arrival had been counting an episode that measured nothing as one that dropped nothing. '_waypoint_report' moves beside the caller it has. Ticket: none - review findings on the change above --- .../cli/eval/tests/test_timing_report.py | 22 +++++ positronic/cli/eval/timing_report.py | 92 ++++++++++++------- positronic/eval/keys.py | 5 +- positronic/policy/harness.py | 15 +++ positronic/policy/tests/test_harness.py | 17 ++++ positronic/telemetry_keys.py | 4 +- 6 files changed, 117 insertions(+), 38 deletions(-) diff --git a/positronic/cli/eval/tests/test_timing_report.py b/positronic/cli/eval/tests/test_timing_report.py index 20f5f10b3..6aecf7a4b 100644 --- a/positronic/cli/eval/tests/test_timing_report.py +++ b/positronic/cli/eval/tests/test_timing_report.py @@ -769,3 +769,25 @@ def test_a_pass_whose_episodes_carry_no_waypoint_account_reports_none(tmp_path): assert report.waypoints is None assert not any(line.startswith('waypoint') for line in _render(report).splitlines()) + + +def test_an_episode_carrying_no_waypoint_account_is_left_out_and_said_so(tmp_path): + """A directory holding passes from either side of the account's arrival reduces over the episodes that + carry one, and reports how many that was — an episode that measured nothing is not one that dropped + nothing.""" + telemetry_dir = tmp_path / TELEMETRY_SUBDIR + _waypoint_fixture(telemetry_dir) + _write_lines( + telemetry_dir / f'older{SPANS_SUFFIX}', + [ + _span(SPAN_EVAL_PASS, 200, 300, 'pass1'), + _span(SPAN_EPISODE, 210, 250, 'ep2', 'pass1', {ATTR_EPISODE_VIRTUAL_S: 20.0}), + ], + ) + report = _build_report(_read_spans_dir(telemetry_dir), [], policy_gpu=None) + + assert report.episodes == 3 + assert report.waypoints is not None + assert report.waypoints.episodes == 2 + assert report.waypoints.scheduled == 200 # the uninstrumented episode adds no zero of its own + assert 'waypoints: 200 scheduled, 40 emitted, 40 dropped over 2 of 3 episodes' in _render(report) diff --git a/positronic/cli/eval/timing_report.py b/positronic/cli/eval/timing_report.py index 4c9662494..2e288f3a5 100644 --- a/positronic/cli/eval/timing_report.py +++ b/positronic/cli/eval/timing_report.py @@ -156,12 +156,17 @@ class WaypointReport: """How well the loop kept the trajectory's schedule, summed over the pass's episodes and their command channels. - ``dropped`` counts a waypoint a round overtook, so ``dropped_share`` is a fraction of - ``emitted + dropped``; ``scheduled`` also covers what a fresh chunk replaced first. The distribution + ``dropped_share`` is a fraction of ``emitted + dropped``, the waypoints that came due. The distribution behind the lateness figures is per channel in the episode's own statics; a percentile of the pass is not recoverable from here. + + ``episodes`` is how many of the pass's episodes carried an account, which is what every figure here + covers. A directory holding passes from either side of this account's arrival reduces to fewer than the + pass's episodes, and the report says so rather than counting an episode that measured nothing as one + that dropped nothing. """ + episodes: int scheduled: int emitted: int dropped: int @@ -387,6 +392,17 @@ def flush_cycle() -> None: ) +@dataclass +class _EpisodeWaypoints: + """One episode's waypoint totals over its command channels, as its span carries them.""" + + scheduled: int + emitted: int + dropped: int + late_sum_ms: float + late_max_ms: float + + @dataclass class _EpisodeTiming: """One episode's wall aggregate reduced from its span subtree; fields are seconds unless named otherwise. @@ -402,11 +418,20 @@ class _EpisodeTiming: policy_wait_s: float overhead_s: float infer_ms: list[float] - waypoints_scheduled: int - waypoints_emitted: int - waypoints_dropped: int - late_sum_ms: float - late_max_ms: float + waypoints: _EpisodeWaypoints | None + + +def _episode_waypoints(episode: SpanRec) -> _EpisodeWaypoints | None: + """One episode's waypoint account, or ``None`` for a span that carries none.""" + if ATTR_WAYPOINTS_SCHEDULED not in episode.attrs: + return None + return _EpisodeWaypoints( + scheduled=int(episode.attrs[ATTR_WAYPOINTS_SCHEDULED]), + emitted=int(episode.attrs[ATTR_WAYPOINTS_EMITTED]), + dropped=int(episode.attrs[ATTR_WAYPOINTS_DROPPED]), + late_sum_ms=float(episode.attrs[ATTR_WAYPOINTS_LATE_SUM_MS]), + late_max_ms=float(episode.attrs[ATTR_WAYPOINTS_LATE_MAX_MS]), + ) def _episode_timing(episode: SpanRec, children: dict[str, list[SpanRec]]) -> _EpisodeTiming: @@ -435,30 +460,7 @@ def _episode_timing(episode: SpanRec, children: dict[str, list[SpanRec]]) -> _Ep policy_wait_s=policy_wait_s, overhead_s=max(wall_s - measured, 0.0), infer_ms=infer_ms, - waypoints_scheduled=int(episode.attrs.get(ATTR_WAYPOINTS_SCHEDULED, 0)), - waypoints_emitted=int(episode.attrs.get(ATTR_WAYPOINTS_EMITTED, 0)), - waypoints_dropped=int(episode.attrs.get(ATTR_WAYPOINTS_DROPPED, 0)), - late_sum_ms=float(episode.attrs.get(ATTR_WAYPOINTS_LATE_SUM_MS, 0.0)), - late_max_ms=float(episode.attrs.get(ATTR_WAYPOINTS_LATE_MAX_MS, 0.0)), - ) - - -def _waypoint_report(timings: list[_EpisodeTiming]) -> WaypointReport | None: - """The pass's waypoint account, or ``None`` where no episode carries one — a run that played no - trajectory, and a sidecar whose episodes hold no waypoint attributes, reduce the same way.""" - scheduled = sum(t.waypoints_scheduled for t in timings) - if not scheduled: - return None - emitted = sum(t.waypoints_emitted for t in timings) - dropped = sum(t.waypoints_dropped for t in timings) - due = emitted + dropped - return WaypointReport( - scheduled=scheduled, - emitted=emitted, - dropped=dropped, - dropped_share=(dropped / due) if due else 0.0, - mean_late_ms=(sum(t.late_sum_ms for t in timings) / emitted) if emitted else 0.0, - max_late_ms=max((t.late_max_ms for t in timings), default=0.0), + waypoints=_episode_waypoints(episode), ) @@ -509,6 +511,31 @@ def _episode_windows(episodes: list[SpanRec]) -> dict[str | None, tuple[int, int } +def _waypoint_report(timings: list[_EpisodeTiming]) -> WaypointReport | None: + """The pass's waypoint account over the episodes that carried one, or ``None`` where none did.""" + accounts = [t.waypoints for t in timings if t.waypoints is not None] + if not accounts: + return None + if len(accounts) < len(timings): + logger.warning( + '%d of %d episode(s) carry no waypoint account; every waypoint figure covers the rest', + len(timings) - len(accounts), + len(timings), + ) + emitted = sum(a.emitted for a in accounts) + dropped = sum(a.dropped for a in accounts) + due = emitted + dropped + return WaypointReport( + episodes=len(accounts), + scheduled=sum(a.scheduled for a in accounts), + emitted=emitted, + dropped=dropped, + dropped_share=(dropped / due) if due else 0.0, + mean_late_ms=(sum(a.late_sum_ms for a in accounts) / emitted) if emitted else 0.0, + max_late_ms=max((a.late_max_ms for a in accounts), default=0.0), + ) + + def _build_report(spans: list[SpanRec], stats: list[dict], policy_gpu: GpuSummary | None) -> PassReport: children: dict[str, list[SpanRec]] = defaultdict(list) for span in spans: @@ -638,7 +665,8 @@ def _render(report: PassReport) -> str: if report.waypoints is not None: way = report.waypoints lines += [ - f'waypoints: {way.scheduled} scheduled, {way.emitted} emitted, {way.dropped} dropped', + f'waypoints: {way.scheduled} scheduled, {way.emitted} emitted, {way.dropped} dropped' + + (f' over {way.episodes} of {report.episodes} episodes' if way.episodes < report.episodes else ''), f'waypoint drops: {way.dropped_share * 100:>6.1f}% of the waypoints that came due', f'waypoint late mean: {way.mean_late_ms:.1f} ms (max {way.max_late_ms:.1f})', ] diff --git a/positronic/eval/keys.py b/positronic/eval/keys.py index 0dff1675a..3ac2a5778 100644 --- a/positronic/eval/keys.py +++ b/positronic/eval/keys.py @@ -43,9 +43,8 @@ TASK = 'eval.task' # How well the loop kept the trajectory's schedule, per command channel: a reader composes the prefix, a -# channel and a field, as f'{SCHEDULE}.{keys.ROBOT_COMMAND}.{DROPPED}'. ``DROPPED`` counts a waypoint a round -# overtook, so ``EMITTED + DROPPED`` is what came due and ``SCHEDULED`` also covers what a fresh chunk replaced -# first. The lateness figures are milliseconds against an emitted waypoint's own due time. +# channel and a field, as f'{SCHEDULE}.{keys.ROBOT_COMMAND}.{DROPPED}'. ``DROPPED`` is a waypoint a later one +# overtook before it could go out; the lateness figures are milliseconds against a waypoint's own due time. SCHEDULE = 'eval.schedule' SCHEDULED = 'scheduled' EMITTED = 'emitted' diff --git a/positronic/policy/harness.py b/positronic/policy/harness.py index 8c1a0bf55..f676e100a 100644 --- a/positronic/policy/harness.py +++ b/positronic/policy/harness.py @@ -187,6 +187,10 @@ def __init__(self) -> None: def count_scheduled(self, waypoints: int) -> None: self._scheduled += waypoints + def count_dropped(self, waypoints: int) -> None: + """Record waypoints discarded without going out, a fresh chunk having replaced them after their time.""" + self._dropped += waypoints + def count_played(self, popped: int, late_ns: int) -> None: """Record a round that emitted the newest of ``popped`` due waypoints, ``late_ns`` past its due time.""" self._emitted += 1 @@ -458,6 +462,14 @@ def _assert_anchored(trajectory: list[dict[str, Any]], now: float) -> None: f'rig-side stack is not anchoring chunks to the harness clock' ) + @staticmethod + def _due_count(schedule: deque[tuple[int, Any]], now_ns: int) -> int: + """How many of a schedule's waypoints have come due. A schedule ascends, so they are its leading run.""" + for index, (due_ns, _) in enumerate(schedule): + if due_ns > now_ns: + return index + return len(schedule) + def _reschedule(self, trajectory: list[dict[str, Any]], clock: pimm.Clock) -> None: """Replace the schedule being played with the session's trajectory. Every channel it names gets that channel's waypoints; one it omits is cleared and holds. The timestamps are already absolute, stamped @@ -465,8 +477,11 @@ def _reschedule(self, trajectory: list[dict[str, Any]], clock: pimm.Clock) -> No """ self._assert_anchored(trajectory, clock.now()) self._telemetry.step() + now_ns = clock.now_ns() # Layers time actions in float seconds; the schedules and every pimm channel are in ns. for name, schedule in self._schedules.items(): + # A chunk landing on a late round replaces waypoints already due, which then go out on no round. + self._fidelity[name].count_dropped(self._due_count(schedule, now_ns)) schedule.clear() schedule.extend((int(a[keys.ACTION_TIMESTAMP] * 1e9), a[name]) for a in trajectory if name in a) self._fidelity[name].count_scheduled(len(schedule)) diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py index 464f643b0..720132125 100644 --- a/positronic/policy/tests/test_harness.py +++ b/positronic/policy/tests/test_harness.py @@ -1754,6 +1754,23 @@ def test_a_round_that_finds_one_waypoint_due_counts_no_drop(): assert account[_schedule_key(keys.ROBOT_COMMAND, eval_keys.DROPPED)] == 0 +@pytest.mark.timeout(3.0) +def test_a_fresh_chunk_counts_the_waypoints_it_replaced_after_their_time(): + """A chunk landing on a late round replaces waypoints that had already come due. They go out on no + round, so they are drops; the ones still ahead of the round are not.""" + harness, recorder = _harness_recording_commands() + channel = keys.ROBOT_COMMAND + harness._schedules[channel].extend((ms * 1_000_000, f'waypoint@{ms}ms') for ms in (0, 10, 20, 30)) + harness._fidelity[channel].count_scheduled(4) + + harness._reschedule([{keys.ACTION_TIMESTAMP: 0.03, channel: 'fresh'}], _FrozenClock(25 * 1_000_000)) + + account = _schedule_account(harness, channel) + assert account[_schedule_key(channel, eval_keys.DROPPED)] == 3 # 0, 10 and 20 ms; the 30 ms one was early + assert account[_schedule_key(channel, eval_keys.SCHEDULED)] == 5 + assert not _emitted_commands(recorder) + + @pytest.mark.timeout(3.0) def test_lateness_is_measured_against_the_waypoint_that_went_out(): """The waypoint sent at 25 ms was due at 20, so it is 5 ms late. A round measured against the oldest diff --git a/positronic/telemetry_keys.py b/positronic/telemetry_keys.py index 96b4b303d..d0906ba4b 100644 --- a/positronic/telemetry_keys.py +++ b/positronic/telemetry_keys.py @@ -32,9 +32,7 @@ ATTR_EPISODE_PARTIAL = 'episode.partial' ATTR_PASS_FAILED = 'pass.failed' -# One episode's waypoint account, totalled over its command channels. ``DROPPED`` counts a waypoint a round -# overtook, so ``EMITTED + DROPPED`` is what came due and ``SCHEDULED`` also covers what a fresh chunk -# replaced first. +# One episode's waypoint account, totalled over its command channels. ATTR_WAYPOINTS_SCHEDULED = 'episode.waypoints.scheduled' ATTR_WAYPOINTS_EMITTED = 'episode.waypoints.emitted' ATTR_WAYPOINTS_DROPPED = 'episode.waypoints.dropped'