From b50b5cc1a377492aa45569df9ab4ae7acacab5c8 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 23 Oct 2026 22:29:52 -0500 Subject: [PATCH] Allow slow source tracking experiments from measured native sensitivity --- openpilot/starpilot/flm/live_tracking.py | 14 +- .../starpilot/flm/live_tracking_actor.py | 1 + .../starpilot/flm/live_tracking_episodes.py | 13 +- .../starpilot/flm/live_tracking_ingress.py | 39 ++++- .../starpilot/flm/source_plant_runtime.py | 14 +- .../flm/tests/test_tracking_pose_interval.py | 140 ++++++++++++++++++ prebuilt.json | 2 +- 7 files changed, 208 insertions(+), 15 deletions(-) create mode 100644 openpilot/starpilot/flm/tests/test_tracking_pose_interval.py diff --git a/openpilot/starpilot/flm/live_tracking.py b/openpilot/starpilot/flm/live_tracking.py index 9e1b41f96f..02e1ca97eb 100644 --- a/openpilot/starpilot/flm/live_tracking.py +++ b/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: diff --git a/openpilot/starpilot/flm/live_tracking_actor.py b/openpilot/starpilot/flm/live_tracking_actor.py index cb2c979226..fe41a1ec5d 100644 --- a/openpilot/starpilot/flm/live_tracking_actor.py +++ b/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) diff --git a/openpilot/starpilot/flm/live_tracking_episodes.py b/openpilot/starpilot/flm/live_tracking_episodes.py index 0f001e14eb..b0ca3a1cd0 100644 --- a/openpilot/starpilot/flm/live_tracking_episodes.py +++ b/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) diff --git a/openpilot/starpilot/flm/live_tracking_ingress.py b/openpilot/starpilot/flm/live_tracking_ingress.py index 48f9d3869c..dff00131f0 100644 --- a/openpilot/starpilot/flm/live_tracking_ingress.py +++ b/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 diff --git a/openpilot/starpilot/flm/source_plant_runtime.py b/openpilot/starpilot/flm/source_plant_runtime.py index 72b72baaf6..619f0c2a30 100644 --- a/openpilot/starpilot/flm/source_plant_runtime.py +++ b/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 diff --git a/openpilot/starpilot/flm/tests/test_tracking_pose_interval.py b/openpilot/starpilot/flm/tests/test_tracking_pose_interval.py new file mode 100644 index 0000000000..2eaa4b779a --- /dev/null +++ b/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) diff --git a/prebuilt.json b/prebuilt.json index bc0c773f6f..673bd059cf 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "3b7cf8d9825b2b10cb3097344954b4b332e57e3f3c9cf3a9ab839fbb23f40d26", "schema_version": 1} +{"protected_inventory_sha256": "0d75026b4495a25fef19288fbb9687970cee858fd6abf5799370b414c824890b", "schema_version": 1}