perf(perception): require exact critical-range persistence

tests/test_m48_low_step_occupancy.py

@@ -23,10 +23,7 @@ from k1link.perception.providers import SourcePacket
 from k1link.perception.recorded_source import RecordedFrameReference
 
 REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
-PROFILE_PATH = (
-    REPOSITORY_ROOT
-    / "config/perception/m48r3-additive-low-step-occupancy-v1.json"
-)
+PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48r3-additive-low-step-occupancy-v1.json"
 R2_CASES_PATH = (
     REPOSITORY_ROOT
     / ".runtime/compute-experiments/m48/static-occupancy-qualification-results"
@@ -126,9 +123,7 @@ def test_wide_operator_region_cannot_bridge_two_spatial_components() -> None:
 
     observations = provider.associate(_packet(), ())
     additive = tuple(
-        item
-        for item in observations
-        if "additive-low-step-current-component" in item.reason_codes
+        item for item in observations if "additive-low-step-current-component" in item.reason_codes
     )
 
     assert len(additive) == 2
@@ -155,15 +150,29 @@ def test_sparse_component_requires_bounded_causal_persistence() -> None:
         profile=load_m48_low_step_occupancy_profile(PROFILE_PATH),
     )
 
-    first = provider.associate(_packet(0), ())
-    second = provider.associate(_packet(1), ())
-    third = provider.associate(_packet(2), ())
+    first_five = tuple(provider.associate(_packet(sequence), ()) for sequence in range(5))
+    sixth = provider.associate(_packet(5), ())
 
-    assert first == ()
-    assert second == ()
-    assert len(third) == 1
-    assert third[0].source_point_ids == (0, 1)
-    assert "additive-low-step-current-component" in third[0].reason_codes
+    assert first_five == ((), (), (), (), ())
+    assert len(sixth) == 1
+    assert sixth[0].source_point_ids == (0, 1)
+    assert "additive-low-step-current-component" in sixth[0].reason_codes
+
+
+def test_persistent_sparse_component_is_bounded_to_critical_range() -> None:
+    points = np.asarray(
+        ((0.00, 0.0, 8.20), (0.04, 0.0, 8.20)),
+        dtype=np.float64,
+    )
+    store = _Store(_frame(points), np.ones(2, dtype=np.uint8))
+    provider = M48AdditiveLowStepGeometryProvider(  # type: ignore[arg-type]
+        store=store,
+        profile=load_m48_low_step_occupancy_profile(PROFILE_PATH),
+    )
+
+    observations = tuple(provider.associate(_packet(sequence), ()) for sequence in range(6))
+
+    assert observations == ((), (), (), (), (), ())
 
 
 def test_frame_1856_preserves_baseline_posts_and_splits_low_hemisphere_support() -> None:
@@ -185,8 +194,7 @@ def test_frame_1856_preserves_baseline_posts_and_splits_low_hemisphere_support()
         profile=frame.projection,
     )
     source_rows = {
-        int(source_index): row
-        for row, source_index in enumerate(projected.source_indices)
+        int(source_index): row for row, source_index in enumerate(projected.source_indices)
     }
     cases = [
         json.loads(line)
@@ -198,14 +206,62 @@ def test_frame_1856_preserves_baseline_posts_and_splits_low_hemisphere_support()
     hemispheres = by_anchor["anchor-924a4623077fe5df18816b47"]
 
     assert posts["accepted_graph"]["component_count"] >= 2
-    assert _component_hits(
-        observations,
-        hemispheres["extent_xyxy"],
-        projected.pixels_xy,
-        source_rows,
-        width=frame.projection.width,
-        height=frame.projection.height,
-    ) >= 2
+    assert (
+        _component_hits(
+            observations,
+            hemispheres["extent_xyxy"],
+            projected.pixels_xy,
+            source_rows,
+            width=frame.projection.width,
+            height=frame.projection.height,
+        )
+        >= 2
+    )
+
+
+def test_exact_six_frame_persistence_recovers_critical_near_anchor() -> None:
+    store = RecordedGeometryStore.from_repository(REPOSITORY_ROOT)
+    provider = M48AdditiveLowStepGeometryProvider(
+        store=store,
+        profile=load_m48_low_step_occupancy_profile(PROFILE_PATH),
+    )
+    observations: tuple[ObstacleObservation, ...] = ()
+    for source_sequence in range(1084, 1093):
+        observations = provider.associate(_packet(source_sequence), ())
+
+    evidence_frame = store.frame_for_index(1092)
+    target_frame = store.frame_for_index(1093)
+    assert evidence_frame is not None
+    assert target_frame is not None
+    projected = project_map_points_kb4(
+        evidence_frame.points_map,
+        position_map_xyz=target_frame.sensor_position_map,
+        orientation_map_from_lidar_xyzw=target_frame.sensor_orientation_xyzw,
+        profile=target_frame.projection,
+    )
+    source_rows = {
+        int(source_index): row for row, source_index in enumerate(projected.source_indices)
+    }
+    critical = next(
+        json.loads(line)
+        for line in R2_CASES_PATH.read_text("utf-8").splitlines()
+        if "anchor-0df056d9c565b74a25d3cca3" in line
+    )
+    additive = tuple(
+        item for item in observations if "additive-low-step-current-component" in item.reason_codes
+    )
+
+    assert (
+        _component_hits(
+            additive,
+            critical["extent_xyxy"],
+            projected.pixels_xy,
+            source_rows,
+            width=target_frame.projection.width,
+            height=target_frame.projection.height,
+        )
+        >= 1
+    )
 
 
 def _component_hits(
@@ -229,9 +285,6 @@ def _component_hits(
         if not indices:
             continue
         rows = [source_rows[index] for index in indices if index in source_rows]
-        if any(
-            x1 <= pixels[row, 0] <= x2 and y1 <= pixels[row, 1] <= y2
-            for row in rows
-        ):
+        if any(x1 <= pixels[row, 0] <= x2 and y1 <= pixels[row, 1] <= y2 for row in rows):
             count += 1
     return count