diff --git a/positronic/cli/eval/tests/test_timing_report.py b/positronic/cli/eval/tests/test_timing_report.py index d577b25ee..6aecf7a4b 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,104 @@ 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, 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 + 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) + + +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()) + + +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 78f138c52..2e288f3a5 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,30 @@ 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. + + ``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 + 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 +193,7 @@ class PassReport: infer_p95_ms: float wall_split: WallSplit env_step_split: EnvStepSplit | None + waypoints: WaypointReport | None gpu: GpuReport @@ -362,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. @@ -377,6 +418,20 @@ class _EpisodeTiming: policy_wait_s: float overhead_s: float infer_ms: list[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: @@ -405,6 +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=_episode_waypoints(episode), ) @@ -455,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: @@ -535,6 +616,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 +662,14 @@ 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' 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})', + ] for f in fields(GpuReport): summary = getattr(report.gpu, f.name) if summary is not None: diff --git a/positronic/eval/keys.py b/positronic/eval/keys.py index a063c7d37..3ac2a5778 100644 --- a/positronic/eval/keys.py +++ b/positronic/eval/keys.py @@ -41,3 +41,14 @@ # 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: a reader composes the prefix, a +# 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' +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..f676e100a 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: @@ -163,6 +163,88 @@ 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._late_sum_ns = 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_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 + 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: + 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 +277,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,10 +315,19 @@ 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 + 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]: @@ -278,7 +371,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.""" @@ -294,6 +387,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): @@ -368,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 @@ -375,20 +477,27 @@ 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)) 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``. @@ -406,7 +515,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 b05bed4c8..720132125 100644 --- a/positronic/policy/tests/test_harness.py +++ b/positronic/policy/tests/test_harness.py @@ -1689,6 +1689,145 @@ 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_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 + waypoint it overtook would read 25.""" + 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): @@ -2536,3 +2675,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.""" + 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..d0906ba4b 100644 --- a/positronic/telemetry_keys.py +++ b/positronic/telemetry_keys.py @@ -32,6 +32,13 @@ ATTR_EPISODE_PARTIAL = 'episode.partial' ATTR_PASS_FAILED = 'pass.failed' +# 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' +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'