fix(checks): fail closed on unmeasurable joint-motion steps (#546)

src/hflow/checks.py

@@ -79,11 +79,19 @@ def timestamps(ep: hflow.Episode) -> hflow.CheckResult:
 @dataclass(frozen=True)
 class _JointMotionProfile:
     """Finite-difference motion facts of one state channel, computed once for
-    every check that reasons about joint speed."""
+    every check that reasons about joint speed.
+
+    ``per_step_max_speed`` is max over joints of |dq/dt|, one per step, and is
+    NaN exactly where the step is not measurable; ``measurable`` is that mask
+    spelled out -- strictly positive duration AND finite positions on every
+    joint (#546). A NaN step is not a clean step and not a moving step: it is
+    unmeasured, and every consumer must compute over ``measurable`` only.
+    """
 
     stamps_ns: np.ndarray
     deltas_s: np.ndarray
-    per_step_max_speed: np.ndarray  # max over joints of |dq/dt|, one per step
+    per_step_max_speed: np.ndarray  # max over joints of |dq/dt|, NaN if unmeasurable
+    measurable: np.ndarray  # bool, one per step: finite speed on a positive dt
     nonpositive_dt_count: int
 
 
@@ -100,12 +108,25 @@ def _joint_motion_profile(
     if len(stamps_ns) < 2:
         return None
     deltas_s = np.diff(stamps_ns) / 1e9
+    position_jumps = np.diff(positions, axis=0)
+    finite_jump = np.all(np.isfinite(position_jumps), axis=1)
+    measurable = (deltas_s > 0) & finite_jump
     safe_deltas_s = np.where(deltas_s > 0, deltas_s, np.nan)
-    velocities = np.abs(np.diff(positions, axis=0)) / safe_deltas_s[:, np.newaxis]
+    velocities = np.abs(position_jumps) / safe_deltas_s[:, np.newaxis]
+    # The mask picks the measurable rows, and every measurable row is finite,
+    # so the per-step maximum is an ordinary max over those rows; unmeasurable
+    # rows stay NaN because they are exactly the rows not filled. np.nanmax
+    # here computed the same values but warned "All-NaN slice encountered" on
+    # every duplicate-stamped or NaN stream -- numpy noise aimed at precisely
+    # the users this profile exists to serve (#549 review).
+    per_step_max_speed = np.full(len(deltas_s), np.nan)
+    if measurable.any():
+        per_step_max_speed[measurable] = np.max(velocities[measurable], axis=1)
     return _JointMotionProfile(
         stamps_ns=stamps_ns,
         deltas_s=deltas_s,
-        per_step_max_speed=np.nanmax(velocities, axis=1),
+        per_step_max_speed=per_step_max_speed,
+        measurable=measurable,
         nonpositive_dt_count=int(np.sum(deltas_s <= 0)),
     )
 
@@ -372,19 +393,39 @@ def joint_discontinuity(
     Ships as measurements and intervals only -- never a default reject rule:
     motion-smoothness heuristics are known to invert on real defects (the
     Voxel51 result), so the threshold and any verdict stay user-owned.
+
+    A duplicate-stamped or NaN-positioned step has no velocity at all, and a
+    NaN comparison is False: counting such steps as compliant (or letting
+    them dilute the percentage denominator) fails the gate open (#546).
+    Every percentage and maximum here covers measurable steps only, and a
+    stream with nothing measurable reports its counts and withholds the
+    verdict-shaped keys rather than emitting 0% or NaN.
     """
     profile = _joint_motion_profile(episode, topic, field)
     if profile is None:
         return CheckResult(
             measurements={f"{topic}/velocity_sample_count": len(episode.channel(topic).timestamps)}
         )
-    violation_mask = profile.per_step_max_speed > velocity_limit
+    measurable = profile.measurable
+    measurable_step_count = int(np.count_nonzero(measurable))
+    if measurable_step_count == 0:
+        return CheckResult(
+            measurements={
+                f"{topic}/velocity_sample_count": len(profile.stamps_ns),
+                f"{topic}/velocity_measurable_step_count": 0,
+                f"{topic}/nonpositive_dt_count": profile.nonpositive_dt_count,
+            }
+        )
+    speed = profile.per_step_max_speed
+    violation_mask = np.zeros(measurable.shape, dtype=bool)
+    violation_mask[measurable] = speed[measurable] > velocity_limit
     return CheckResult(
         measurements={
-            f"{topic}/max_abs_velocity": float(np.nanmax(profile.per_step_max_speed)),
+            f"{topic}/max_abs_velocity": float(np.max(speed[measurable])),
             f"{topic}/velocity_limit": velocity_limit,
-            f"{topic}/violation_count": int(np.sum(violation_mask)),
-            f"{topic}/violation_pct": float(np.mean(violation_mask) * 100.0),
+            f"{topic}/violation_count": int(np.count_nonzero(violation_mask)),
+            f"{topic}/violation_pct": float(np.mean(violation_mask[measurable]) * 100.0),
+            f"{topic}/velocity_measurable_step_count": measurable_step_count,
             f"{topic}/nonpositive_dt_count": profile.nonpositive_dt_count,
         },
         intervals=_mask_run_intervals(
@@ -651,29 +692,48 @@ def idle_fraction(
     velocity_epsilon: float = 0.05,
     min_interval_s: float = 1.0,
 ) -> CheckResult:
-    """Time-weighted fraction of the episode spent with no joint moving.
+    """Time-weighted fraction of the MEASURED episode spent with no joint moving.
 
     A step is idle when every joint's finite-difference speed is below
     ``velocity_epsilon``; the fraction weights each step by its own duration,
     so irregular sampling does not skew it. Idle runs at least
     ``min_interval_s`` long become labeled ``idle:<topic>`` intervals.
     Evidence for curation cuts over mostly-stationary demonstrations -- the
     keep/drop policy (and any verdict) stays user-owned.
+
+    Unmeasurable steps (duplicate stamps, NaN positions) belong in neither the
+    numerator nor the denominator: they had no velocity to be idle under, and
+    counting their time as "moving" understates the fraction (#546). A stream
+    with nothing measurable reports its counts and withholds
+    ``idle_fraction`` rather than storing 0.0 for a dead channel.
     """
     profile = _joint_motion_profile(episode, topic, field)
     if profile is None:
         return CheckResult(
             measurements={f"{topic}/idle_sample_count": len(episode.channel(topic).timestamps)}
         )
-    idle_mask = profile.per_step_max_speed < velocity_epsilon
-    positive_deltas_s = np.where(profile.deltas_s > 0, profile.deltas_s, 0.0)
-    total_span_s = float(np.sum(positive_deltas_s))
-    idle_total_s = float(np.sum(positive_deltas_s[idle_mask]))
+    measurable = profile.measurable
+    measurable_step_count = int(np.count_nonzero(measurable))
+    if measurable_step_count == 0:
+        return CheckResult(
+            measurements={
+                f"{topic}/idle_sample_count": len(profile.stamps_ns),
+                f"{topic}/idle_measurable_step_count": 0,
+                f"{topic}/idle_nonpositive_dt_count": profile.nonpositive_dt_count,
+            }
+        )
+    speed = profile.per_step_max_speed
+    idle_mask = np.zeros(measurable.shape, dtype=bool)
+    idle_mask[measurable] = speed[measurable] < velocity_epsilon
+    total_span_s = float(np.sum(profile.deltas_s[measurable]))
+    idle_total_s = float(np.sum(profile.deltas_s[idle_mask]))
     return CheckResult(
         measurements={
-            f"{topic}/idle_fraction": idle_total_s / total_span_s if total_span_s else 0.0,
+            f"{topic}/idle_fraction": idle_total_s / total_span_s,
             f"{topic}/idle_total_s": idle_total_s,
             f"{topic}/velocity_epsilon": velocity_epsilon,
+            f"{topic}/idle_measurable_step_count": measurable_step_count,
+            f"{topic}/idle_nonpositive_dt_count": profile.nonpositive_dt_count,
         },
         intervals=_mask_run_intervals(
             profile.stamps_ns, idle_mask, f"idle:{topic}", min_duration_s=min_interval_s

tests/test_checks_nan_failopen.py

@@ -0,0 +1,168 @@
+"""#546: unmeasurable joint steps must fail CLOSED.
+
+``_joint_motion_profile`` yields NaN velocity where a step has a duplicate
+timestamp or a NaN position. A NaN comparison is False, so the old code
+counted those steps as compliant and not-idle while their duration stayed in
+the percentage denominators: a stream whose every step duplicates its
+timestamp reported 0.0 violations, and an all-NaN channel reported
+``idle_fraction = 0.0``. These tests pin the measurable-mask discipline:
+metrics over measurable steps only, counts and refusal when nothing is
+measurable, and no NaN ever leaving a check.
+"""
+
+from __future__ import annotations
+
+import math
+from pathlib import Path
+
+import pytest
+from mcap.writer import Writer
+
+from hflow import Episode
+from hflow.checks import idle_fraction, joint_discontinuity
+from hflow.format import METADATA_RECORD_EPISODE
+from hflow.steps import CheckResult
+from hflow.testing import (
+    _JOINT_STATE_SCHEMA_TEXT,
+    JOINT_STATE_SCHEMA_NAME,
+    _encode_joint_state,
+)
+
+TOPIC = "/joint_states"
+NS_PER_S = 1_000_000_000
+
+
+def _joint_states_episode(path: Path, stamps_ns: list[int], positions: list[float]) -> Episode:
+    with path.open("wb") as stream:
+        writer = Writer(stream)
+        writer.start(profile="ros2", library="test-546")
+        schema_id = writer.register_schema(
+            name=JOINT_STATE_SCHEMA_NAME,
+            encoding="ros2msg",
+            data=_JOINT_STATE_SCHEMA_TEXT.encode("utf-8"),
+        )
+        channel_id = writer.register_channel(
+            topic=TOPIC, message_encoding="cdr", schema_id=schema_id
+        )
+        writer.add_metadata(
+            METADATA_RECORD_EPISODE, {"task": "nan-failopen", "embodiment": "probe"}
+        )
+        for sequence, (stamp_ns, position) in enumerate(zip(stamps_ns, positions, strict=True)):
+            payload = _encode_joint_state(
+                stamp_ns,
+                "probe",
+                ["j0"],
+                [position],
+                [],
+                [],
+            )
+            writer.add_message(
+                channel_id=channel_id,
+                log_time=stamp_ns,
+                publish_time=stamp_ns,
+                data=payload,
+                sequence=sequence,
+            )
+        writer.finish()
+    return Episode(path)
+
+
+def _assert_all_measurements_finite(*results: CheckResult) -> None:
+    for result in results:
+        for key, value in result.measurements.items():
+            if isinstance(value, float):
+                assert math.isfinite(value), f"{key} left the check non-finite"
+
+
+def test_duplicate_stamp_stream_refuses_instead_of_reporting_zero_violations(
+    tmp_path: Path,
+) -> None:
+    """Stream A (all duplicate stamps, big position hops) must abstain;
+    Stream B (same positions, advancing stamps) reports 100.0. Old code:
+    A reported 0.0 -- a gate-wide fail-open."""
+    stamps = [0, 0, 0, 0, 0]
+    positions = [0.0, 5.0, 10.0, 15.0, 20.0]
+    with (
+        _joint_states_episode(tmp_path / "a.mcap", stamps, positions) as stream_a,
+        _joint_states_episode(
+            tmp_path / "b.mcap",
+            [i * NS_PER_S for i in range(5)],
+            positions,
+        ) as stream_b,
+    ):
+        refused = joint_discontinuity(stream_a, velocity_limit=3.0)
+        measured = joint_discontinuity(stream_b, velocity_limit=3.0)
+        _assert_all_measurements_finite(refused, measured)
+        assert f"{TOPIC}/violation_pct" not in refused.measurements
+        assert refused.measurements[f"{TOPIC}/velocity_measurable_step_count"] == 0
+        assert refused.measurements[f"{TOPIC}/velocity_sample_count"] == 5
+        assert refused.measurements[f"{TOPIC}/nonpositive_dt_count"] == 4
+        assert measured.measurements[f"{TOPIC}/violation_pct"] == pytest.approx(100.0)
+        assert measured.measurements[f"{TOPIC}/nonpositive_dt_count"] == 0
+
+
+def test_all_nan_channel_withholds_every_percentage(tmp_path: Path) -> None:
+    """An all-NaN position channel is not a 0%-violation, 0%-idle stream;
+    both checks must report counts and abstain (#546's dead-channel case).
+    Note idle_nonpositive_dt_count == 0: the timestamps advanced, only the
+    positions were unmeasurable."""
+    stamps = [i * NS_PER_S for i in range(4)]
+    with _joint_states_episode(tmp_path / "nan.mcap", stamps, [float("nan")] * 4) as episode:
+        velocity = joint_discontinuity(episode, velocity_limit=3.0)
+        idle = idle_fraction(episode, velocity_epsilon=0.05)
+        _assert_all_measurements_finite(velocity, idle)
+        for key in ("violation_pct", "violation_count", "max_abs_velocity"):
+            assert f"{TOPIC}/{key}" not in velocity.measurements
+        assert velocity.measurements[f"{TOPIC}/velocity_measurable_step_count"] == 0
+        assert f"{TOPIC}/idle_fraction" not in idle.measurements
+        assert idle.measurements[f"{TOPIC}/idle_sample_count"] == 4
+        assert idle.measurements[f"{TOPIC}/idle_measurable_step_count"] == 0
+        assert idle.measurements[f"{TOPIC}/idle_nonpositive_dt_count"] == 0
+
+
+def test_unmeasurable_steps_do_not_dilate_percentage_denominators(
+    tmp_path: Path,
+) -> None:
+    """1 violation and 2 hidden violations among 3 measurable steps must be
+    33.3%, not 20% (a revert of the mask in the denominator flips this red).
+    The idle case needs no NaN at all to dilute: a step with positive dt but
+    NaN positions left the old denominator, under-reporting idle."""
+    with (
+        _joint_states_episode(
+            tmp_path / "mixed_v.mcap",
+            [0, 0, NS_PER_S, 2 * NS_PER_S, 2 * NS_PER_S, 3 * NS_PER_S],
+            [0.0, 5.0, 10.0, 10.0, 15.0, 15.0],
+        ) as velocity_episode,
+        _joint_states_episode(
+            tmp_path / "mixed_i.mcap",
+            [0, NS_PER_S, 2 * NS_PER_S],
+            [5.0, 5.0, float("nan")],
+        ) as idle_episode,
+    ):
+        velocity = joint_discontinuity(velocity_episode, velocity_limit=3.0)
+        assert velocity.measurements[f"{TOPIC}/violation_pct"] == pytest.approx(100.0 / 3.0)
+        assert velocity.measurements[f"{TOPIC}/velocity_measurable_step_count"] == 3
+        assert velocity.measurements[f"{TOPIC}/nonpositive_dt_count"] == 2
+        idle = idle_fraction(idle_episode, velocity_epsilon=0.05)
+        assert idle.measurements[f"{TOPIC}/idle_fraction"] == pytest.approx(1.0)
+        assert idle.measurements[f"{TOPIC}/idle_measurable_step_count"] == 1
+        _assert_all_measurements_finite(velocity, idle)
+
+
+def test_every_percentage_names_the_time_it_could_not_measure(
+    tmp_path: Path,
+) -> None:
+    """#546's visibility rule: wherever this profile feeds a percentage, the
+    unmeasurable-step counts ride along on the same row."""
+    with _joint_states_episode(
+        tmp_path / "mixed.mcap",
+        [0, 0, NS_PER_S],
+        [0.0, 5.0, 5.0],
+    ) as episode:
+        velocity = joint_discontinuity(episode, velocity_limit=3.0)
+        idle = idle_fraction(episode, velocity_epsilon=0.05)
+        assert velocity.measurements[f"{TOPIC}/nonpositive_dt_count"] == 1
+        assert f"{TOPIC}/violation_pct" in velocity.measurements
+        assert idle.measurements[f"{TOPIC}/idle_nonpositive_dt_count"] == 1
+        assert f"{TOPIC}/idle_fraction" in idle.measurements
+        _assert_all_measurements_finite(velocity, idle)