Keep parked sessions from consuming untested steering tunes

openpilot/starpilot/flm/offroad_processor.py

@@ -189,6 +189,9 @@ def collect_route_inputs(root, route, *, permitted, walker=None):
   raw_windows = {}
   context_metadata = {}
   counts = {
+    'recorded_lateral_request_rows': 0,
+    'active_lateral_request_rows': 0,
+    'invalid_lateral_request_rows': 0,
     'recorded_tracking': 0,
     'joined_tracking': 0,
     'missing_join': 0,
@@ -226,6 +229,10 @@ def collect_route_inputs(root, route, *, permitted, walker=None):
       if index % 128 == 0 and not permitted():
         raise ValueError('Parked analysis cancelled')
       kind = event.which()
+      if kind == 'carControl':
+        counts['recorded_lateral_request_rows'] += 1
+        counts['active_lateral_request_rows'] += bool(event.carControl.latActive)
+        counts['invalid_lateral_request_rows'] += not event.valid
       if kind == 'carParams':
         # Own the actual CP reader independently of this segment's byte buffer.
         cp = event.carParams.as_builder().as_reader()
@@ -352,9 +359,52 @@ def collect_route_inputs(root, route, *, permitted, walker=None):
   return RecordedInputs(receipt, state, windows, stable_windows, baseline_tracking, bookends, raw_windows, context_metadata, counts, quality_stops)
 
 
-def analyze_latest(params, root, previous, *, permitted):
+def collect_latest_steering_inputs(params, root, *, permitted):
   route = latest_route(params, root, permitted=permitted)
-  collected = collect_route_inputs(root, route, permitted=permitted)
+  import time
+  from openpilot.starpilot.flm.offroad_route import MAX_ANALYSIS_SECONDS, MAX_SEGMENTS, LOCAL_ID
+  from openpilot.starpilot.galaxy.drive_history import DriveHistory
+
+  started = time.monotonic()
+
+  def allowed():
+    return permitted() and time.monotonic() - started <= MAX_ANALYSIS_SECONDS
+
+  collected = collect_route_inputs(root, route, permitted=allowed)
+  current_cp = collected.receipt.cp_sha256
+  skipped = 0
+
+  def inactive(inputs):
+    c = inputs.counts
+    return (
+      c['recorded_lateral_request_rows'] >= 40 and c['active_lateral_request_rows'] == 0 and c['invalid_lateral_request_rows'] == 0 and not inputs.quality_stops
+    )
+
+  if inactive(collected) and LOCAL_ID.fullmatch(route['routeId']):
+    snapshot = DriveHistory(root, max_segments_per_route=MAX_SEGMENTS).snapshot()
+    if snapshot['scanIncomplete']:
+      raise ValueError('Latest steering drive inventory incomplete')
+    counter = int(route['routeId'].split('--')[0], 16)
+    candidates = []
+    for candidate in snapshot['routes']:
+      if LOCAL_ID.fullmatch(candidate['routeId']):
+        distance = (counter - int(candidate['routeId'].split('--')[0], 16)) % 2**32
+        if 1 <= distance <= 3 and candidate['segments'] and all(s['files']['rlog'] for s in candidate['segments']):
+          candidates.append((distance, candidate['routeId'], candidate))
+    for _, _, candidate in sorted(candidates):
+      earlier = collect_route_inputs(root, candidate, permitted=allowed)
+      if earlier.receipt.cp_sha256 != current_cp:
+        continue
+      skipped += 1
+      if not inactive(earlier):
+        route, collected = candidate, earlier
+        collected.counts['inactive_recordings_skipped'] = skipped
+        break
+  return route, collected
+
+
+def analyze_latest(params, root, previous, *, permitted):
+  route, collected = collect_latest_steering_inputs(params, root, permitted=permitted)
   receipt, state = collected.receipt, collected.state
   if type(receipt) is not RouteReceipt:
     raise ValueError('A whole completed-route assessment requires all original segments')

openpilot/starpilot/flm/offroad_quality.py

@@ -6,7 +6,9 @@ Only complete, input-matched original windows can trigger a withdrawal.
 """
 
 from collections import deque
+from collections.abc import Callable
 import math
+import os
 
 from openpilot.starpilot.flm.offroad_surface import ROUTE_NAMES as NAMES
 from openpilot.starpilot.flm.offroad_analysis import cohort
@@ -38,6 +40,7 @@ class TrialGuard:
     self.last_ns = 0
     self.override_ns = 0
     self.stopped = None
+    self.start_trial: Callable[[], bool] | None = None
     self.cohort = None
     self.baseline = ()
     try:
@@ -106,6 +109,15 @@ class TrialGuard:
     ):
       self.rows.clear()
       return None
+    if self.start_trial is not None:
+      start, self.start_trial = self.start_trial, None
+      try:
+        claimed = start()
+      except (OSError, ValueError):
+        claimed = False
+      if not claimed:
+        self.stopped = 'baseline_unavailable'
+        return self.marker
     point = read_point(event, companions, route='current-route-revocation-only', segment=0, context_key=self.scope_key)
     if point is None:
       self.rows.clear()
@@ -167,8 +179,11 @@ def from_configuration(cp, params, cache_directory, *, vehicle_session):
     cache = AssessmentCache(cache_directory)
     trial = trial_document(cache, row['analysis_sha256'])
     guard = TrialGuard(scope.key(), row['values'], trial)
-    if row['state'] == 'ready_for_next_drive' and guard.stopped is None and not cache.claim_trial(row['analysis_sha256'], vehicle_session):
-      guard.stopped = 'baseline_unavailable'
+    if row['state'] == 'ready_for_next_drive' and guard.stopped is None:
+      if os.path.lexists(cache.directory / ('trial-' + row['analysis_sha256'] + '.json')):
+        guard.stopped = 'baseline_unavailable'
+      else:
+        guard.start_trial = lambda digest=row['analysis_sha256'], session=vehicle_session: cache.claim_trial(digest, session)
   except (OSError, KeyError, TypeError, ValueError, OverflowError):
     guard = TrialGuard(scope.key(), row['values'], {})
   return guard

openpilot/starpilot/flm/offroad_runtime.py

@@ -4,6 +4,7 @@ No permissions, response witnesses, ACKs or source clocks are loaded from disk.
 """
 
 import time
+import os
 
 from openpilot.starpilot.flm.offroad_surface import ROUTE_NAMES as NAMES
 from openpilot.starpilot.flm.offroad_store import RouteScope, startup_row
@@ -273,6 +274,11 @@ def configure(cp, ci, controller, params, *, manual_status):
     row, status = startup_row(params, scope)
     if row is None:
       return None, status
+    if row['state'] == 'ready_for_next_drive':
+      from openpilot.starpilot.flm.offroad_cache import directory_for_params
+
+      if os.path.lexists(directory_for_params(params) / ('trial-' + row['analysis_sha256'] + '.json')):
+        return None, 'provisional_trial_already_started'
     values = tuple((n, float(row['values'][n])) for n in NAMES)
     previous = tuple((n, float(row['previous'][n])) for n in NAMES)
     fixed = None if not fixed_manual else fixed_manual_numeric(controller, fixed_manual)

openpilot/starpilot/flm/offroad_worker.py

@@ -8,6 +8,7 @@ from dataclasses import replace
 import hashlib
 import json
 import math
+import os
 from pathlib import Path
 
 from openpilot.starpilot.flm.offroad_surface import ROUTE_NAMES as NAMES
@@ -148,7 +149,7 @@ def quality_withdrawn(assessment, scope, values):
   return any(stop_marker(scope.key(), values, r) in reasons for r in STOP_REASONS)
 
 
-def assess_next_drive(trial, assessment, scope):
+def assess_next_drive(trial, assessment, scope, *, trial_started=True):
   """Compare independently held-out actual observations; no raw-RMS reward."""
   if trial.get('policy') != POLICY or trial.get('kind') != 'next_drive_trial':
     raise ValueError('Exact provisional assessment provenance required')
@@ -200,6 +201,14 @@ def assess_next_drive(trial, assessment, scope):
         else:
           results.append(compare_observed(old, new))
       comparisons.append({'original_context_sha256': group['original_context_sha256'], 'comparison_contract_sha256': contract, 'independent': results})
+  if not comparisons and not trial_started:
+    return {
+      'outcome': 'pending',
+      'reason': 'trial_not_applied_waiting_for_eligible_drive',
+      'observed_only': True,
+      'causal_road_benefit_proven': False,
+      'comparisons': [],
+    }
   improved = bool(comparisons and all(all(r['improved'] for r in c['independent']) for c in comparisons))
   return {
     'outcome': 'accepted_after_drive' if improved else 'rolled_back',
@@ -327,11 +336,17 @@ class OffroadWorker:
     scope, row, expected = state['scope'], state['row'], state['raw']
 
     def authorized():
-      return (
-        self.permitted()
-        and self.scope_for(assessment.cp, self.params) == scope
-        and latest_route(self.params, self.root, permitted=self.permitted) == assessment.source_inventory
-      )
+      if not (
+        self.permitted() and self.scope_for(assessment.cp, self.params) == scope and latest_route(self.params, self.root, permitted=self.permitted) == route
+      ):
+        return False
+      if assessment.source_inventory == route:
+        return True
+      from openpilot.starpilot.flm.offroad_route import MAX_SEGMENTS
+      from openpilot.starpilot.galaxy.drive_history import DriveHistory
+
+      inventory = DriveHistory(self.root, max_segments_per_route=MAX_SEGMENTS).snapshot()
+      return not inventory['scanIncomplete'] and sum(r == assessment.source_inventory for r in inventory['routes']) == 1
 
     if not authorized():
       raise ValueError('Source or settings changed after fitting')
@@ -360,7 +375,7 @@ class OffroadWorker:
               observed_checks.append(
                 {'qualified': actual['qualified'] and contribution['qualified'], 'actual_recorded_map': actual, 'compared_with_prior_map': contribution}
               )
-      outcome = assess_next_drive(trial, assessment, scope)
+      outcome = assess_next_drive(trial, assessment, scope, trial_started=os.path.lexists(self.cache.directory / ('trial-' + row['analysis_sha256'] + '.json')))
       if outcome['outcome'] == 'accepted_after_drive' and (not observed_checks or not all(r['qualified'] for r in observed_checks)):
         outcome = {
           'outcome': 'rolled_back',
@@ -369,28 +384,33 @@ class OffroadWorker:
           'causal_road_benefit_proven': False,
           'comparisons': outcome['comparisons'],
         }
-      record = {
-        'kind': 'next_drive_outcome',
-        'policy': POLICY,
-        'trial_analysis_sha256': row['analysis_sha256'],
-        'assessment': report,
-        'all_actual_next_drive_checks': observed_checks,
-        **outcome,
-      }
-      digest = self.cache.put(record)
-      saved = RouteStorage(self.params).resolve(
-        scope=scope,
-        expected=expected,
-        outcome=outcome['outcome'],
-        route_sha256=assessment.receipt.source_sha256,
-        completed_wall_ns=assessment.receipt.completed_wall_ns,
-        analysis_sha256=digest,
-        authorized=authorized,
-      )
-      if not saved:
-        raise ValueError('Next-drive outcome compare-and-swap refused')
-      status = outcome['outcome']
-      reason = outcome['reason']
+      if outcome['outcome'] == 'pending':
+        status, reason = 'ready_for_next_drive', outcome['reason']
+        report['unused_recording_review'] = outcome
+        report['recorded_tracking'] = trial.get('assessment', {}).get('recorded_tracking')
+      else:
+        record = {
+          'kind': 'next_drive_outcome',
+          'policy': POLICY,
+          'trial_analysis_sha256': row['analysis_sha256'],
+          'assessment': report,
+          'all_actual_next_drive_checks': observed_checks,
+          **outcome,
+        }
+        digest = self.cache.put(record)
+        saved = RouteStorage(self.params).resolve(
+          scope=scope,
+          expected=expected,
+          outcome=outcome['outcome'],
+          route_sha256=assessment.receipt.source_sha256,
+          completed_wall_ns=assessment.receipt.completed_wall_ns,
+          analysis_sha256=digest,
+          authorized=authorized,
+        )
+        if not saved:
+          raise ValueError('Next-drive outcome compare-and-swap refused')
+        status = outcome['outcome']
+        reason = outcome['reason']
     else:
       current = dict(ZERO) if row is None else dict(row['values'])
       choices = []

openpilot/starpilot/flm/tests/test_offroad_backend.py

@@ -123,6 +123,34 @@ class TestOffroadCache(unittest.TestCase):
       first = from_configuration(None, None, self.cache.directory, vehicle_session='a' * 32)
       self.assertIsNotNone(first)
       self.assertIsNone(first.stopped)
+      claim = self.cache.directory / ('trial-' + row['analysis_sha256'] + '.json')
+      self.assertFalse(claim.exists(), 'A parked constructor does not consume an untested trial')
+      from types import SimpleNamespace
+
+      stamp = 1_000_000_000
+      control = SimpleNamespace(
+        valid=True,
+        enabled=True,
+        active=False,
+        manual=False,
+        override=False,
+        fault=False,
+        limited=False,
+        saturated=False,
+        control_ns=stamp,
+        route_effect_revision=REVISION,
+        route_configuration_scope=scope().key(),
+        route_applied_coordinates=tuple(values(0.01).items()),
+      )
+      event = SimpleNamespace(logMonoTime=stamp)
+      first.observe(event, {}, control, now_ns=stamp)
+      self.assertFalse(claim.exists())
+      stamp += 10_000_000
+      event.logMonoTime = control.control_ns = stamp
+      control.active = True
+      with patch('openpilot.starpilot.flm.offroad_quality.read_point', return_value=None):
+        first.observe(event, {}, control, now_ns=stamp)
+      self.assertTrue(claim.exists(), 'The first valid current active route frame consumes the trial once')
       restarted = from_configuration(None, None, self.cache.directory, vehicle_session='b' * 32)
       self.assertEqual(restarted.stopped, 'baseline_unavailable')
       owner = RouteEffects(scope(), tuple(values(0.01).items()), (), previous=tuple(values(0.005).items()))
@@ -133,6 +161,118 @@ class TestOffroadCache(unittest.TestCase):
       row['state'] = 'accepted_after_drive'
       self.assertIsNone(from_configuration(None, None, self.cache.directory, vehicle_session='c' * 32).stopped)
 
+  def test_unused_session_waits_but_started_or_withdrawn_trials_still_roll_back(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.flm.offroad_worker import assess_next_drive, POLICY
+    from openpilot.starpilot.flm.offroad_surface import stop_marker
+
+    declared = dict(trial(), policy=POLICY, values=values(0.01), original_metadata={})
+    empty = SimpleNamespace(groups=(), stable_groups=(), quality_stop_markers=())
+    self.assertEqual(assess_next_drive(declared, empty, scope(), trial_started=False)['outcome'], 'pending')
+    self.assertEqual(assess_next_drive(declared, empty, scope(), trial_started=True)['outcome'], 'rolled_back')
+    marker = stop_marker(scope().key(), declared['values'], 'gross_tracking_or_oscillation_worse')
+    stopped = SimpleNamespace(groups=(), stable_groups=(), quality_stop_markers=({'reason': marker},))
+    self.assertEqual(assess_next_drive(declared, stopped, scope(), trial_started=False)['outcome'], 'rolled_back')
+
+  def test_actual_start_is_blocked_before_controller_reuse_and_rollback_restores_prior_map(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.flm.offroad_runtime import configure
+
+    params = Params(str(self.root / 'startup'))
+    cache = AssessmentCache(directory_for_params(params))
+    row = {'state': 'ready_for_next_drive', 'values': values(0.01), 'previous': values(), 'analysis_sha256': cache.put(trial())}
+    with (
+      patch('openpilot.starpilot.flm.offroad_runtime.configuration_data', return_value=(scope(), (), ())),
+      patch('openpilot.starpilot.flm.offroad_runtime.startup_row', return_value=(row, 'loaded')),
+    ):
+      owner, status = configure(None, None, SimpleNamespace(), params, manual_status='not_configured')
+      self.assertIsNotNone(owner)
+      self.assertEqual(status, 'configured_pending_actual_stage')
+      self.assertTrue(cache.claim_trial(row['analysis_sha256'], 'a' * 32))
+      owner, status = configure(None, None, SimpleNamespace(), params, manual_status='not_configured')
+      self.assertIsNone(owner, 'A genuine consumed trial cannot apply even one frame after restart')
+      self.assertEqual(status, 'provisional_trial_already_started')
+      row['state'] = 'rolled_back'
+      row['values'] = values(0.005)
+      owner, status = configure(None, None, SimpleNamespace(), params, manual_status='not_configured')
+      self.assertIsNotNone(owner, 'Rollback must restore the previous known map, including a nonzero accepted tune')
+      assert owner is not None
+      self.assertEqual(dict(owner.values), row['values'])
+      self.assertEqual(status, 'configured_pending_actual_stage')
+
+  def test_last_steering_drive_selection_ignores_only_recorded_inactive_sessions(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.flm import offroad_processor as module
+
+    latest = {'routeId': '00000012--aaaaaaaaaa', 'segments': [{'files': {'rlog': True}}]}
+    earlier = {'routeId': '00000011--bbbbbbbbbb', 'segments': [{'files': {'rlog': True}}]}
+    counts = {'recorded_lateral_request_rows': 100, 'active_lateral_request_rows': 0, 'invalid_lateral_request_rows': 0}
+    parked = SimpleNamespace(receipt=SimpleNamespace(cp_sha256='a' * 64), counts=dict(counts), quality_stops=[])
+    driving = SimpleNamespace(receipt=SimpleNamespace(cp_sha256='a' * 64), counts=dict(counts, active_lateral_request_rows=70), quality_stops=[])
+    calls = []
+
+    def collect(root, route, *, permitted):
+      self.assertTrue(permitted())
+      calls.append(route['routeId'])
+      return parked if route is latest else driving
+
+    history = SimpleNamespace(snapshot=lambda: {'scanIncomplete': False, 'routes': [latest, earlier]})
+    with (
+      patch.object(module, 'latest_route', return_value=latest),
+      patch.object(module, 'collect_route_inputs', new=collect),
+      patch('openpilot.starpilot.galaxy.drive_history.DriveHistory', return_value=history),
+    ):
+      route, collected = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True)
+      self.assertIs(route, earlier)
+      self.assertIs(collected, driving)
+      self.assertEqual(calls, [latest['routeId'], earlier['routeId']])
+      calls.clear()
+      parked.counts['active_lateral_request_rows'] = 1
+      route, _ = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True)
+      self.assertIs(route, latest, 'A failed or weak active drive must never cause a search for favorable older results')
+      self.assertEqual(calls, [latest['routeId']])
+      calls.clear()
+      parked.counts['active_lateral_request_rows'] = 0
+      driving.receipt.cp_sha256 = 'b' * 64
+      route, _ = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True)
+      self.assertIs(route, latest, 'An older car cannot be substituted')
+
+  def test_selected_previous_drive_keeps_both_source_inventories_guarded(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.flm import offroad_worker as module
+    from openpilot.starpilot.flm.offroad_processor import CompletedAssessment
+    from openpilot.starpilot.flm.offroad_route import RouteReceipt
+
+    parked = {'routeId': '00000012--aaaaaaaaaa'}
+    selected = {'routeId': '00000011--bbbbbbbbbb'}
+    cp = SimpleNamespace(carFingerprint='declared-unit-only')
+    receipt = RouteReceipt(selected['routeId'], 'd' * 64, (), 'a' * 64, 1, 2, 3, ())
+    assessment = CompletedAssessment(receipt, (), {}, cp, (), selected)
+
+    def assessed(params, root, previous, *, permitted):
+      previous(cp)
+      return assessment
+
+    for intact in (True, False):
+      with self.subTest(selected_source_intact=intact), tempfile.TemporaryDirectory() as directory:
+        history = SimpleNamespace(snapshot=lambda intact=intact: {'scanIncomplete': False, 'routes': [parked, selected] if intact else [parked]})
+        with (
+          patch.object(module, 'latest_route', return_value=parked),
+          patch('openpilot.starpilot.galaxy.drive_history.DriveHistory', return_value=history),
+        ):
+          params = Params(directory)
+          worker = module.OffroadWorker(
+            params, directory, Path(directory) / 'cache', permitted=lambda: True, assessor=assessed, scope_for=lambda cp, params: scope()
+          )
+          if intact:
+            result = worker.run_once()
+            self.assertEqual(result['completed_route'], selected['routeId'])
+            self.assertFalse(result['configuration_changed'])
+          else:
+            with self.assertRaisesRegex(ValueError, 'Source or settings changed'):
+              worker.run_once()
+          self.assertEqual(read_saved(params, KEY, MAX_BYTES), (None, True))
+
 
 class TestOffroadSurface(unittest.TestCase):
   def test_real_controller_route_replay_preserves_limiter_and_requires_exact_hooks(self):

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "89a56ed49aa9b917f6284ab4d50ebeb52b659220a49bd13ec2f0b90a49ae0299", "schema_version": 1}
+{"protected_inventory_sha256": "fb95974ed48cf6ef0178e5090be20cde2e4a10104fd9d72abcbf71c493fb3c0f", "schema_version": 1}