Count the waypoints a late round drops, and how late they go out

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)

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:

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'