Allow slow source tracking experiments from measured native sensitivity

openpilot/starpilot/flm/live_tracking.py

@@ -213,6 +213,12 @@ def _discovery_support(rows, kind, model_rmse):
   # Maneuver geometry and miss severity are separate. A modest tracking miss
   # in a substantial curve needs the longer small-error evidence path, rather
   # than being excluded by the large-error floor forever.
+  if model_rmse is None:
+    # Native visibility selects a bounded experiment, not permission to keep it.
+    # One complete original window is enough to try a slow single-coordinate
+    # step; three NEW uncensored baseline windows and later feedback still
+    # decide its outcome. Quantized/missing probes never become sensitivity.
+    return (tuple(rows[-1:]), 1) if rows else ((), None)
   large = tuple(rows[-3:])
   if kind == 'large' and len(large) == 3 and all(row.error_rms >= (0.10 if model_rmse is None else max(0.10, 6 * model_rmse)) for row in large):
     return large, 3
@@ -587,9 +593,9 @@ class LiveTrackingOptimizer:
   def _comparison_allocation(
     self, episode: TrackingEpisode, expected: float, target: dict[str, float], *, minimum_episodes: int | None = None
   ) -> tuple[int, int] | None:
-    if minimum_episodes is not None and minimum_episodes not in (3, 9):
+    if minimum_episodes is not None and minimum_episodes not in ((1, 3, 9) if self.rmse is None else (3, 9)):
       return None
-    minimum = max(_required(episode.kind), minimum_episodes or 0)
+    minimum = max(3 if self.rmse is None else _required(episode.kind), minimum_episodes or 0)
     if type(expected) not in (int, float) or not math.isfinite(expected) or expected <= 0.0001:
       return None
     # Freeze the NEW baseline size before observing any of its scores. More
@@ -855,9 +861,9 @@ class LiveTrackingOptimizer:
   def _trial_plan(
     self, rows: tuple[TrackingEpisode, ...], expected: float, target: dict[str, float], *, minimum_episodes: int | None = None
   ) -> TrialPlan | None:
-    if minimum_episodes is not None and minimum_episodes not in (3, 9):
+    if minimum_episodes is not None and minimum_episodes not in ((1, 3, 9) if self.rmse is None else (3, 9)):
       return None
-    minimum = max(_required(rows[0].kind), minimum_episodes or 0)
+    minimum = max(3 if self.rmse is None else _required(rows[0].kind), minimum_episodes or 0)
     if type(expected) not in (int, float) or not math.isfinite(expected) or expected <= 0.0001:
       return None
     if self.rmse is None:

openpilot/starpilot/flm/live_tracking_actor.py

@@ -371,6 +371,7 @@ class LiveTrackingActor:
     if self.process is not None and self.process.child is not None:
       return  # At most one optional reconstruction/identification child.
     optimizer = self.ingress.optimizer
+    self.reason = optimizer.reason
     accepted = tuple((name, optimizer.accepted[name]) for name in NAMES)
     profile_hash = full_profile_sha256(self.certificate.binding.baseline_revision, dict(accepted))
     token = (self.token, self.certificate.model_revision, profile_hash)

openpilot/starpilot/flm/live_tracking_episodes.py

@@ -780,7 +780,7 @@ class TrackingEpisodeBuilder:
       episode.validate()
       requests = {point.control_ns: self.probe_requests[point.control_ns] for point in points if point.control_ns in self.probe_requests}
       defer = (
-        kind == 'subtle'
+        (kind == 'subtle' or self.source_tracking)
         and getattr(self, 'defer_unprobed', False) is True
         and type(getattr(self, 'numerical_observed_ns', 0)) is int
         and self.numerical_observed_ns > 0
@@ -843,6 +843,17 @@ class TrackingEpisodeBuilder:
     # Preserve source order before the optimizer's strict chronological check;
     # never replace source time with numerical completion time. The original
     # probe deadline and bounded ready deque still bound this wait.
+    if self.source_tracking and self.ready:
+      # Fixed source windows are independently usable. An earlier incomplete
+      # numerical job must not hold every later complete window for 30 seconds.
+      # Drop it by source order alone, without inspecting tracking error or
+      # native sensitivity; late numerical results cannot resurrect it.
+      through = max(episode.end_ns for episode in self.ready)
+      for key, row in tuple(self.completed.items()):
+        if row['episode'].end_ns <= through:
+          for point in row['points']:
+            self.probe_requests.pop(point.control_ns, None)
+          del self.completed[key]
     barrier = min((row['episode'].start_ns for row in self.completed.values()), default=None)
     ordered = sorted(self.ready, key=lambda episode: episode.start_ns)
     result = tuple(episode for episode in ordered if barrier is None or episode.end_ns < barrier)

openpilot/starpilot/flm/live_tracking_ingress.py

@@ -97,13 +97,42 @@ class LiveTrackingIngress:
           self.optimizer.observe_episode(episode, now_ns=now_ns)
     return accepted
 
-  def observe_original_pose(self, pose, collector, *, now_ns, current_native):
+  def original_pose_pair(self, ready, pose):
     lag = self.optimizer.lag_ns
-    lower = [(c, o) for c, o in collector.ready if o.apply_ns <= pose.source_lower_ns-lag]
-    upper = [(c, o) for c, o in collector.ready if o.apply_ns <= pose.source_upper_ns-lag]
-    if not lower or not upper or lower[-1][1].apply_ns != upper[-1][1].apply_ns:
+    lower = [(c, o) for c, o in ready if o.apply_ns <= pose.source_lower_ns - lag]
+    upper = [(c, o) for c, o in ready if o.apply_ns <= pose.source_upper_ns - lag]
+    if not lower or not upper:
       return None
-    return self.observe_pose(pose, *lower[-1], now_ns=now_ns, current_native=current_native)
+    pair = lower[-1]
+    if not 0 <= pose.source_upper_ns - lag - pair[1].apply_ns <= 30_000_000:
+      return None
+    if pair[1].apply_ns != upper[-1][1].apply_ns:
+      if not self.episodes.source_tracking:
+        return None  # A plant response still requires one unambiguous input.
+      # Source-only tracking compares desired and measured acceleration at the
+      # pose time, not a fitted command->response. Retain the genuine earlier
+      # causal emission across clock uncertainty, only within one consumed map
+      # and source lineage. Never interpolate a command or replace its clocks.
+      binding = self.binding
+      lineage = (binding.identity, binding.source_epoch, binding.baseline_revision,
+                 binding.controller_revision, binding.native_identity, binding.native_required)
+      try:
+        profile = self.profile(pair[0])
+        for control, output in ready:
+          if pair[1].apply_ns <= output.apply_ns <= upper[-1][1].apply_ns:
+            if ((control.identity, control.source_epoch, control.baseline_revision,
+                 control.controller_revision, control.native_identity, control.native_required) != lineage
+                or control.generation != pair[0].generation or self.profile(control) != profile):
+              return None
+      except (AttributeError, TypeError, ValueError):
+        return None
+    return pair
+
+  def observe_original_pose(self, pose, collector, *, now_ns, current_native):
+    pair = self.original_pose_pair(collector.ready, pose)
+    if pair is None:
+      return None
+    return self.observe_pose(pose, *pair, now_ns=now_ns, current_native=current_native)
 
   def observe_pose(self, pose, control, output, *, now_ns, current_native):
     # The caller passes the collector's original lag-associated pair. Neither

openpilot/starpilot/flm/source_plant_runtime.py

@@ -70,10 +70,16 @@ def observe(sm, cp, collector, now_ns, current_native):
   live_lag = live.optimizer.lag_ns if live is not None else getattr(collector, 'live_response_lag_ns', None)
   collector.live_response_pair = None
   if type(live_lag) is int and 0 <= live_lag <= 500_000_000:
-    lower = [(c,o) for c,o in collector.ready if o.apply_ns <= row.source_lower_ns-live_lag]
-    upper = [(c,o) for c,o in collector.ready if o.apply_ns <= row.source_upper_ns-live_lag]
-    if lower and upper and lower[-1][1].apply_ns == upper[-1][1].apply_ns:
-      collector.live_response_pair = (*lower[-1], row)
+    if live is not None:
+      collector.live_response_pair = None
+      pair = live.original_pose_pair(collector.ready, row)
+      if pair is not None:
+        collector.live_response_pair = (*pair, row)
+    else:
+      lower = [(c,o) for c,o in collector.ready if o.apply_ns <= row.source_lower_ns-live_lag]
+      upper = [(c,o) for c,o in collector.ready if o.apply_ns <= row.source_upper_ns-live_lag]
+      if lower and upper and lower[-1][1].apply_ns == upper[-1][1].apply_ns:
+        collector.live_response_pair = (*lower[-1], row)
     if live is not None:
       live.observe_original_pose(row, collector, now_ns=now_ns, current_native=current_native)
   model, binding = adapter.model, adapter.source_plant_binding

openpilot/starpilot/flm/tests/test_tracking_pose_interval.py

@@ -0,0 +1,140 @@
+"""Pair selection only; these operands grant no native or application authority."""
+from types import SimpleNamespace
+import unittest
+
+from openpilot.starpilot.flm.live_tracking_ingress import LiveTrackingIngress
+
+
+def subject(source_tracking=True):
+  owner = object.__new__(LiveTrackingIngress)
+  owner.binding = SimpleNamespace(identity='1'*64, source_epoch='2'*64, baseline_revision='3'*64,
+                                  controller_revision='revision', native_identity='4'*64, native_required=True)
+  owner.episodes = SimpleNamespace(source_tracking=source_tracking)
+  owner.optimizer = SimpleNamespace(lag_ns=0)
+  owner.last_coordinates = owner.last_profile = None
+  return owner
+
+
+def pair(owner, ns):
+  control = SimpleNamespace(**vars(owner.binding), generation=1, live_applied_coordinates=(),
+                            effect_trace=None, applied_coordinates=(0., 0.))
+  return control, SimpleNamespace(apply_ns=ns)
+
+
+class TestTrackingPoseInterval(unittest.TestCase):
+  def test_same_original_emission_works_for_both_protocols(self):
+    for source_tracking in (False, True):
+      owner = subject(source_tracking)
+      row = pair(owner, 1_000_000_000)
+      pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_007_000_000)
+      selected = owner.original_pose_pair([row], pose)
+      assert selected[0] is row[0] and selected[1] is row[1]
+
+  def test_source_tracking_keeps_original_earlier_causal_emission(self):
+    owner = subject()
+    rows = [pair(owner, 1_000_000_000), pair(owner, 1_010_000_000)]
+    pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000)
+    selected = owner.original_pose_pair(rows, pose)
+    assert selected[0] is rows[0][0] and selected[1] is rows[0][1]
+    assert rows[0][1].apply_ns == 1_000_000_000
+
+  def test_plant_model_keeps_single_emission_requirement(self):
+    owner = subject(False)
+    rows = [pair(owner, 1_000_000_000), pair(owner, 1_010_000_000)]
+    pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000)
+    assert owner.original_pose_pair(rows, pose) is None
+
+  def test_crossing_consumed_profile_change_is_rejected(self):
+    owner = subject()
+    rows = [pair(owner, 1_000_000_000), pair(owner, 1_010_000_000)]
+    rows[1][0].live_applied_coordinates = (0.001,)+(0.,)*29
+    pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000)
+    assert owner.original_pose_pair(rows, pose) is None
+
+  def test_crossing_source_or_generation_change_is_rejected(self):
+    for name, value in (('source_epoch', '5'*64), ('generation', 2)):
+      owner = subject()
+      rows = [pair(owner, 1_000_000_000), pair(owner, 1_010_000_000)]
+      setattr(rows[1][0], name, value)
+      pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000)
+      assert owner.original_pose_pair(rows, pose) is None
+
+  def test_original_thirty_millisecond_join_limit_is_kept(self):
+    owner = subject()
+    row = pair(owner, 1_000_000_000)
+    pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_031_000_000)
+    assert owner.original_pose_pair([row], pose) is None
+
+  def test_no_preceding_original_emission_is_not_filled(self):
+    owner = subject()
+    row = pair(owner, 1_010_000_000)
+    pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000)
+    assert owner.original_pose_pair([row], pose) is None
+
+  def test_unknown_consumed_profile_is_not_assumed_neutral(self):
+    owner = subject()
+    rows = [pair(owner, 1_000_000_000), pair(owner, 1_010_000_000)]
+    rows[1][0].applied_coordinates = (0.001, 0.)
+    pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000)
+    assert owner.original_pose_pair(rows, pose) is None
+
+
+class TestSourceWindowProbeDeferral(unittest.TestCase):
+  def builder(self, source_tracking):
+    from dataclasses import replace
+    from openpilot.starpilot.flm.tests.test_live_tracking_episodes import completed_builder
+    builder = completed_builder(count=4, error=0.2)
+    builder.source_tracking = source_tracking
+    builder.defer_unprobed = True
+    builder.numerical_observed_ns = 2_000_000_000
+    if source_tracking:
+      builder.endpoints = type(builder.endpoints)(
+        (replace(point, reference_ns=point.control_ns) for point in builder.endpoints), maxlen=320)
+    return builder
+
+  def test_short_source_curve_waits_for_original_probe_support(self):
+    builder = self.builder(True)
+    builder._finish()
+    assert len(builder.completed) == 1
+    assert not builder.drain(now_ns=2_000_000_001)
+    row = next(iter(builder.completed.values()))
+    assert row['episode'].kind == 'large'
+    assert row['deadline'] == 32_000_000_000
+
+  def test_probe_deadline_is_unchanged_and_missing_support_stays_partial(self):
+    builder = self.builder(True)
+    builder._finish()
+    assert not builder.drain(now_ns=31_999_999_999)
+    episode, = builder.drain(now_ns=32_000_000_000)
+    assert not episode.native_gradient_complete
+    assert episode.sensitivity == ()
+
+  def test_non_tracking_large_curve_keeps_existing_behavior(self):
+    builder = self.builder(False)
+    builder._finish()
+    assert not builder.completed
+    episode, = builder.drain(now_ns=2_000_000_001)
+    assert episode.kind == 'large'
+
+class TestSourceExperimentEntry(unittest.TestCase):
+  def test_one_measured_window_can_select_source_experiment(self):
+    from openpilot.starpilot.flm.live_tracking import _discovery_support
+    row = SimpleNamespace(error_rms=0.02)
+    assert _discovery_support((row,), 'subtle', None) == ((row,), 1)
+    assert _discovery_support((), 'subtle', None) == ((), None)
+    assert _discovery_support((row,), 'subtle', 0.001) == ((), None)
+
+  def test_source_experiment_still_allocates_three_fresh_baseline_windows(self):
+    from openpilot.starpilot.flm.live_tracking import LiveTrackingOptimizer
+    import inspect
+    # Exercise the real allocation without manufacturing source/native authority.
+    function = next(value for name, value in vars(LiveTrackingOptimizer).items() if name == '_comparison_allocation')
+    scope = dict(function.__globals__)
+    scope['_matches'] = lambda a, b: True
+    from types import FunctionType
+    allocation = FunctionType(function.__code__, scope)
+    owner = SimpleNamespace(rmse=None, binding='binding', accepted=dict.fromkeys(scope['NAMES'], 0.), lag_ns=0)
+    owner.planning_history = [SimpleNamespace(binding='binding', profile_sha256='profile', end_ns=n*2_000_000_000) for n in (1,2,3)]
+    episode = SimpleNamespace(kind='subtle', profile_sha256='profile', end_ns=6_000_000_000)
+    target = dict(owner.accepted); target[next(iter(target))] = .005
+    assert allocation(owner, episode, .001, target, minimum_episodes=1) == (3,3)

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "3b7cf8d9825b2b10cb3097344954b4b332e57e3f3c9cf3a9ab839fbb23f40d26", "schema_version": 1}
+{"protected_inventory_sha256": "0d75026b4495a25fef19288fbb9687970cee858fd6abf5799370b414c824890b", "schema_version": 1}