Add metric comparison helpers for FT tests

[fzyzcjy]

Add the metric comparison helpers used by fault-tolerance tests to compare logged
training metrics offline.

• miles/utils/test_utils/comparisons/metrics.py.

[gemini-code-assist]

Code Review

This pull request introduces metric comparison utilities and associated unit tests to compare baseline and target metric events, handling fault-tolerance retries by filtering out earlier failed attempts. The feedback highlights two key improvements: optimizing the event filtering logic from O(N^2) to O(N) to prevent performance degradation on large logs, and removing a fragile dependency on an internal module of the external sglang library by rendering the Polars DataFrame directly.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The current implementation of _keep_only_final_attempt has an $O(N^2)$ time complexity because it performs a nested loop over all events for each unique rollout_id. For large training logs with thousands of events, this can significantly slow down test execution. We can optimize this to $O(N)$ by using a single pass to find the maximum attempt for each rollout_id.

def _keep_only_final_attempt(events: list[MetricEvent]) -> list[MetricEvent]:
    """Keep only events from the highest-attempt for each rollout_id.

    During FT healing, a crashed rollout is retried at attempt+1; events from
    the failed attempt are partial and should be discarded for comparison.

    Rollout-side metrics (e.g. RolloutManager log_rollout_metrics) have
    attempt=None — they are not part of the FT retry stream, so we treat them
    as a single attempt (normalized to 0).
    """

    def _attempt(e: MetricEvent) -> int:
        return e.attempt if e.attempt is not None else 0

    max_attempt_by_rollout: dict[int | None, int] = {}
    for e in events:
        rid = e.rollout_id
        att = _attempt(e)
        if rid not in max_attempt_by_rollout or att > max_attempt_by_rollout[rid]:
            max_attempt_by_rollout[rid] = att

    return [e for e in events if _attempt(e) == max_attempt_by_rollout[e.rollout_id]]

[Shi-Dong]

This is worth fixing.

[Shi-Dong]

Please hoist this import to the beginning

[Shi-Dong]

LGTM with nits.