Require native effect in the selected FLM trial direction

Symmetric native probes could qualify a coordinate even when only the opposite direction changed its encoded command. Carry each direction’s measured visibility and physical effect into selection and interrupted-study recovery so a quantized direction cannot borrow the other direction’s evidence.

Keep existing trial bounds, slew, deadlines and independent acceptance criteria. Validate the previous selector failure, full FLM suite and recorded d5 producer/learner replay. Real-drive improvement requires post-update recorded evidence.

openpilot/starpilot/flm/live_tracking.py

@@ -11,7 +11,7 @@ from collections import deque
 from collections.abc import Sequence
 from dataclasses import dataclass
 from functools import lru_cache
-from typing import TYPE_CHECKING, Literal, NotRequired, TypedDict
+from typing import TYPE_CHECKING, Literal, NotRequired, TypedDict, cast
 import hashlib
 import math
 import statistics
@@ -101,6 +101,21 @@ class HookSensitivity:
   native_endpoints: int
   max_step_physical_rms: float  # RMS physical acceleration DELTA at .005, after native encoding and retained slope.
   native_gradient_complete: bool = False
+  minus_native_separated: int | None = None
+  plus_native_separated: int | None = None
+  minus_max_step_physical_rms: float | None = None
+  plus_max_step_physical_rms: float | None = None
+
+  def native_effect(self, step: float) -> tuple[int, float]:
+    if step == 0:
+      return 0, 0.0
+    if self.minus_native_separated is None:
+      # Older numerical episodes retain their original secant contract.
+      # Production probe episodes record both actual directions below.
+      return self.native_separated, self.max_step_physical_rms
+    if step < 0:
+      return self.minus_native_separated, cast(float, self.minus_max_step_physical_rms)
+    return cast(int, self.plus_native_separated), cast(float, self.plus_max_step_physical_rms)
 
   def validate(self):
     if (
@@ -117,6 +132,14 @@ class HookSensitivity:
       and self.native_endpoints < 4
     ):
       raise ValueError('actual finite hook/native probe support required')
+    directional = (self.minus_native_separated, self.plus_native_separated,
+                   self.minus_max_step_physical_rms, self.plus_max_step_physical_rms)
+    if any(value is not None for value in directional) and (
+      any(type(value) is not int or not 0 <= value <= self.native_endpoints for value in directional[:2])
+      or any(value is None or type(value) not in (int, float) or not math.isfinite(value) or value < 0 for value in directional[2:])
+      or any(count == 0 and effect != 0 for count, effect in zip(directional[:2], directional[2:], strict=True))
+    ):
+      raise ValueError('complete finite directional native probe support required')
     return self
 
 
@@ -589,7 +612,11 @@ class LiveTrackingOptimizer:
         return None
       # No error/error_feature_sum/sign is inspected: the earlier coordinate
       # and bounded direction were fixed before these observations existed.
-      effects.append(feature.max_step_physical_rms * abs(target[name] - self.accepted[name]) / MAX_STEP / _score_scale(row))
+      delta = target[name] - self.accepted[name]
+      separated, effect = feature.native_effect(delta)
+      if separated / feature.native_endpoints < 0.09 or effect <= 0:
+        return None
+      effects.append(effect * abs(delta) / MAX_STEP / _score_scale(row))
     return min(effects) if effects else None
 
   def _comparison_allocation(
@@ -1060,7 +1087,11 @@ class LiveTrackingOptimizer:
           if self.rmse is None and self._same_inconclusive_trial(name, sign, rows[-1], now_ns):
             continue
           delta = sign * step_limit
-          predicted = min(features[id(row)][name].max_step_physical_rms for row in rows)
+          directed = [features[id(row)][name].native_effect(delta) for row in rows]
+          if any(separated / features[id(row)][name].native_endpoints < 0.09
+                 for row, (separated, _) in zip(rows, directed, strict=True)):
+            continue
+          predicted = min(effect for _, effect in directed)
           expected = predicted * abs(delta) / MAX_STEP / max(_score_scale(row) for row in rows)
           if expected <= 0.0001:
             continue  # The original native effect must be visible.

openpilot/starpilot/flm/live_tracking_episodes.py

@@ -14,6 +14,7 @@ import struct
 from openpilot.common.constants import ACCELERATION_DUE_TO_GRAVITY
 from openpilot.starpilot.flm.canonical_surface_adapter import NAMES, PROBE_GROUPS, probe_names, full_profile_sha256
 from openpilot.starpilot.flm.fullrate_collector import ControlEvidence, AppliedEvidence
+from openpilot.starpilot.flm.live_trial_contract import MAX_STEP
 from openpilot.starpilot.flm.live_tracking import TrackingEpisode, HookSensitivity
 from openpilot.starpilot.flm.live_tracking_response import validate_certificate, validate_response_witness, physical_calibration_generation
 from openpilot.starpilot.flm.original_source_capture import OriginalSourcePose, validate_pose
@@ -562,6 +563,18 @@ class TrackingEpisodeBuilder:
   def _sensitivities(self, points, probes, *, requested=None):
     rows = self._gradient_rows(points, probes)
     coverage = self._gradient_coverage(points, probes, set(probes) if requested is None else requested, rows)
+    def directed_effect(row, side):
+      coordinate = getattr(row, side + '_coordinate') - row.baseline_coordinate
+      native = getattr(row, side + '_delivered') - row.baseline_delivered
+      native_width = row.plus_delivered - row.minus_delivered
+      if coordinate == 0 or native_width == 0:
+        return 0.0
+      # Recover the same measured/native conversion used by the secant,
+      # then measure THIS direction against the encoded baseline. A visible
+      # opposite probe cannot make a quantized selected direction visible.
+      gain = abs(row.physical_derivative * (row.plus_coordinate - row.minus_coordinate) / native_width)
+      return abs(native) * gain * MAX_STEP / abs(coordinate)
+
     return tuple(
       HookSensitivity(
         name,
@@ -571,6 +584,10 @@ class TrackingEpisodeBuilder:
         len(values),
         math.sqrt(sum(row.max_step_physical_delta**2 for _, row in values) / len(values)),
         coverage.get(name, False),
+        sum(row.minus_delivered != row.baseline_delivered for _, row in values),
+        sum(row.plus_delivered != row.baseline_delivered for _, row in values),
+        math.sqrt(sum(directed_effect(row, 'minus')**2 for _, row in values) / len(values)),
+        math.sqrt(sum(directed_effect(row, 'plus')**2 for _, row in values) / len(values)),
       )
       for name, values in rows.items()
       if values

openpilot/starpilot/flm/tests/test_directional_native_fallback.py

@@ -0,0 +1,23 @@
+"""A dead native direction must not monopolize another measurable correction."""
+from dataclasses import replace
+from openpilot.starpilot.flm.live_trial_contract import MAX_STEP
+from openpilot.starpilot.flm.tests.test_live_tracking import episode
+from openpilot.starpilot.flm.tests.test_tracking_source_context import owner, refresh
+from openpilot.starpilot.flm.tests.test_directional_native_visibility import feature
+
+def test_quantized_preference_allows_a_different_measurable_calming_hook():
+    o = owner()
+    for i in range(40):
+        e = episode(o, i, 'center_taper_max', kind='subtle')
+        dead = feature(error=-0.5)
+        center = feature(name='center_taper_max', error=0.05)
+        e = replace(e, sensitivity=(dead, center), normalization_accel=max(0.1, e.goal_rms))
+        refresh(o, e.end_ns)
+        o.observe_episode(e, now_ns=e.end_ns)
+        if o.offer_state is not None:
+            break
+    assert o.offer_state is not None
+    assert o.offer_state['names'] == ('center_taper_max',)
+    assert o.offer_state['target']['center_taper_max'] == MAX_STEP
+    assert o.offer_state['target']['ff_gain_right'] == 0.0
+    assert not o.offer_state.get('accepted', False) and (not any(o.accepted.values()))

openpilot/starpilot/flm/tests/test_directional_native_visibility.py

@@ -0,0 +1,74 @@
+"""Numerical selection regressions; no fabricated hardware/plant admission."""
+from dataclasses import replace
+from types import SimpleNamespace
+import pytest
+from openpilot.starpilot.flm.live_tracking import HookSensitivity
+from openpilot.starpilot.flm.live_tracking_episodes import TrackingEpisodeBuilder
+from openpilot.starpilot.flm.live_tracking_probe import _physical_row
+from openpilot.starpilot.flm.live_trial_contract import MAX_STEP
+from openpilot.starpilot.flm.tests.test_tracking_source_context import owner, refresh
+from openpilot.starpilot.flm.tests.test_live_tracking import episode
+
+@pytest.mark.parametrize('minus,plus,low,high,negative,positive', [
+    (-MAX_STEP, MAX_STEP, 0.1, 0.102, 0.0, 0.002),
+    (-MAX_STEP, MAX_STEP, 0.098, 0.1, 0.002, 0.0),
+    (-MAX_STEP, MAX_STEP, 0.098, 0.102, 0.002, 0.002),
+    (0.0, MAX_STEP, 0.1, 0.102, 0.0, 0.002),
+])
+def test_directional_visibility_uses_actual_encoded_baseline(minus, plus, low, high, negative, positive):
+    probe = _physical_row('ff_gain_right', 0.0, minus, plus, 0.1, low, high, 1.0, 'separated')
+    points = [SimpleNamespace(control_ns=100 + i * 100, apply_ns=110 + i * 100, error=0.1) for i in range(4)]
+    probes = {point.control_ns: SimpleNamespace(apply_ns=point.apply_ns, rows=(probe,)) for point in points}
+    builder = object.__new__(TrackingEpisodeBuilder)
+    feature = builder._sensitivities(points, probes)[0].validate()
+    assert feature.native_gradient_complete
+    count, effect = feature.native_effect(-MAX_STEP)
+    assert count == (4 if negative else 0) and effect == pytest.approx(negative)
+    count, effect = feature.native_effect(MAX_STEP)
+    assert count == (4 if positive else 0) and effect == pytest.approx(positive)
+
+def feature(name='ff_gain_right', error=-0.05):
+    return HookSensitivity(name, error, 1.0, 20, 20, 0.01, True, 0, 20, 0.0, 0.02).validate()
+
+def test_selected_direction_cannot_borrow_opposite_native_visibility():
+    o = owner()
+    for i in range(9):
+        e = episode(o, i, 'ff_gain_right', kind='large')
+        e = replace(e, sensitivity=(feature(),), normalization_accel=max(0.1, e.goal_rms))
+        refresh(o, e.end_ns)
+        o.observe_episode(e, now_ns=e.end_ns)
+    assert o.offer_state is None and o.discovery is None
+    assert not any(o.accepted.values())
+
+def test_visible_direction_still_gets_a_bounded_trial():
+    o = owner()
+    for i in range(20):
+        e = episode(o, i, 'ff_gain_right', kind='large')
+        e = replace(e, sensitivity=(feature(error=0.05),), normalization_accel=max(0.1, e.goal_rms))
+        refresh(o, e.end_ns)
+        o.observe_episode(e, now_ns=e.end_ns)
+        if o.offer_state is not None:
+            break
+    assert o.offer_state is not None
+    assert o.offer_state['target']['ff_gain_right'] == MAX_STEP
+    assert not o.offer_state.get('accepted', False) and (not any(o.accepted.values()))
+
+def test_recovery_requires_visibility_in_the_already_selected_direction():
+    o = owner()
+    e = episode(o, 0, 'ff_gain_right', kind='large')
+    e = replace(e, sensitivity=(feature(),))
+    target = dict(o.accepted)
+    target['ff_gain_right'] = -MAX_STEP
+    assert o._fresh_selected_effect((e,), 'ff_gain_right', target) is None
+    target['ff_gain_right'] = MAX_STEP
+    assert o._fresh_selected_effect((e,), 'ff_gain_right', target) > 0
+
+@pytest.mark.parametrize('changes', [
+    {'minus_native_separated': None},
+    {'plus_max_step_physical_rms': float('nan')},
+    {'minus_max_step_physical_rms': 0.01},
+    {'plus_native_separated': 21},
+])
+def test_partial_or_impossible_directional_support_is_rejected(changes):
+    with pytest.raises(ValueError):
+        replace(feature(), **changes).validate()

openpilot/starpilot/flm/tests/test_live_tracking_episodes.py

@@ -173,7 +173,11 @@ def test_physical_probe_aggregates_have_signed_derivative_and_independent_native
   b = completed_builder(count=4, error=0.2)
   name = NAMES[0]
   for index, point in enumerate(b.endpoints):
-    row = SimpleNamespace(name=name, physical_derivative=-2.0, native_separated=index % 2 == 0, max_step_physical_delta=0.01, status='separated')
+    row = SimpleNamespace(
+      name=name, physical_derivative=-2.0, native_separated=index % 2 == 0, max_step_physical_delta=0.01, status='separated',
+      baseline_coordinate=0.0, minus_coordinate=-0.05, plus_coordinate=0.05,
+      baseline_delivered=0.0, minus_delivered=0.01 if index % 2 == 0 else 0.0, plus_delivered=-0.01 if index % 2 == 0 else 0.0,
+    )
     b.probes[point.control_ns] = SimpleNamespace(
       rows=(row,), control_ns=point.control_ns, apply_ns=point.apply_ns, profile_sha256=b.pending[0][4], model_revision=b.certificate.model_revision
     )
@@ -505,7 +509,11 @@ def numerical_probe(b, point, group, name):
     profile_sha256=next(iter(b.completed.values()))['episode'].profile_sha256,
     model_revision=b.certificate.model_revision,
     group=group,
-    rows=(SimpleNamespace(name=name, physical_derivative=-2.0, native_separated=True, max_step_physical_delta=0.01, status='separated'),),
+    rows=(SimpleNamespace(
+      name=name, physical_derivative=-2.0, native_separated=True, max_step_physical_delta=0.01, status='separated',
+      baseline_coordinate=0.0, minus_coordinate=-0.05, plus_coordinate=0.05,
+      baseline_delivered=0.0, minus_delivered=0.01, plus_delivered=-0.01,
+    ),),
   )
 
 
@@ -535,7 +543,11 @@ def numerical_probe_from_saved(b, saved, point, group, name):
     profile_sha256=saved['episode'].profile_sha256,
     model_revision=b.certificate.model_revision,
     group=group,
-    rows=(SimpleNamespace(name=name, physical_derivative=-2.0, native_separated=True, max_step_physical_delta=0.01, status='separated'),),
+    rows=(SimpleNamespace(
+      name=name, physical_derivative=-2.0, native_separated=True, max_step_physical_delta=0.01, status='separated',
+      baseline_coordinate=0.0, minus_coordinate=-0.05, plus_coordinate=0.05,
+      baseline_delivered=0.0, minus_delivered=0.01, plus_delivered=-0.01,
+    ),),
   )
 
 

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "0226b7b111e3bf69ae1e54b94cd8d92fb57c1661a6ce3ec382778fc7425c5f8b", "schema_version": 1}
+{"protected_inventory_sha256": "4c1d246773e42dedf3c37ef869b403650a69d07ee24be98f493cd18d652177b0", "schema_version": 1}