Finish pending completed-drive reviews after newer trips

openpilot/starpilot/flm/offroad_cache.py

@@ -10,6 +10,7 @@ import tempfile
 MAX_DOCUMENT = 1024 * 1024
 MAX_FILES = 128
 MAX_TOTAL = 64 * 1024 * 1024
+MAX_CHECKPOINTS = 8
 
 
 def _pairs(rows):
@@ -106,7 +107,7 @@ class AssessmentCache:
   def _capacity(self, size):
     total = count = 0
     for p in self.directory.iterdir():
-      if p.name == 'latest.json':
+      if p.name in ('latest.json', 'inputs.json'):
         continue
       if not p.name.endswith('.json'):
         raise ValueError('Unexpected assessment directory contents')
@@ -149,11 +150,8 @@ class AssessmentCache:
       os.close(fd)
     return self._read(name) == raw
 
-  def latest(self):
-    try:
-      value = json.loads(self._read('latest.json', 32768), object_pairs_hook=_pairs)
-    except FileNotFoundError:
-      return None
+  @staticmethod
+  def _checkpoint(value):
     if (
       type(value) is not dict
       or set(value) != {'input_fingerprint', 'result'}
@@ -166,8 +164,33 @@ class AssessmentCache:
       raise ValueError('Invalid completed-drive checkpoint')
     return value
 
+  def latest(self):
+    try:
+      return self._checkpoint(json.loads(self._read('latest.json', 32768), object_pairs_hook=_pairs))
+    except FileNotFoundError:
+      return None
+
+  def checkpoints(self):
+    try:
+      rows = json.loads(self._read('inputs.json', MAX_CHECKPOINTS * 32768 + 32), object_pairs_hook=_pairs)
+    except FileNotFoundError:
+      rows = []
+    if type(rows) is not list or len(rows) > MAX_CHECKPOINTS:
+      raise ValueError('Bounded completed-drive history required')
+    rows = [self._checkpoint(row) for row in rows]
+    if len({row['input_fingerprint'] for row in rows}) != len(rows):
+      raise ValueError('Duplicate completed-drive fingerprint')
+    latest = self.latest()
+    if latest is not None:
+      rows = [row for row in rows if row['input_fingerprint'] != latest['input_fingerprint']] + [latest]
+    return rows[-MAX_CHECKPOINTS:]
+
   def checkpoint(self, value):
+    self._checkpoint(value)
     raw = json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()
     if len(raw) > 32768:
       raise ValueError('Assessment checkpoint bound')
+    rows = [row for row in self.checkpoints() if row['input_fingerprint'] != value['input_fingerprint']] + [value]
+    self._write('inputs.json', json.dumps(rows[-MAX_CHECKPOINTS:], sort_keys=True, separators=(',', ':'), allow_nan=False).encode())
     self._write('latest.json', raw)
+

openpilot/starpilot/flm/offroad_daemon.py

@@ -36,7 +36,7 @@ class StatusReport:
     # Historical findings survive a new fit. Never carry a saved/current map,
     # enabled flag or application evidence forward from a previous report.
     fields = ('completed_route', 'vehicle', 'analysis_counts', 'recorded_tracking', 'source_sha256', 'assessment_sha256',
-              'recommendations', 'suggestions', 'suggestion_count')
+              'recommendations', 'suggestions', 'suggestion_count', 'source_is_latest')
     previous = {key: self.previous[key] for key in fields if key in self.previous}
     if previous:
       previous['retained_decision'] = self.previous.get('reason', 'waiting_for_completed_drive')
@@ -95,6 +95,7 @@ class StatusReport:
     if 'completed_route' in value:
       details['Earlier analysis shown while updating'] = value.get('retained_analysis') is True
       details['Completed drive analyzed'] = value['completed_route']
+      details['Earlier completed drive reviewed'] = value.get('source_is_latest') is False
       details['Recorded steering observations'] = count('recorded_tracking')
       reviewed, available = count('original_segments_reviewed'), count('original_segments_available')
       if reviewed and available:
@@ -337,3 +338,4 @@ def main():
 
 if __name__ == '__main__':
   main()
+

openpilot/starpilot/flm/offroad_processor.py

@@ -387,8 +387,8 @@ 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 collect_latest_steering_inputs(params, root, *, permitted):
-  route = latest_route(params, root, permitted=permitted)
+def collect_latest_steering_inputs(params, root, *, permitted, selected_route=None):
+  route = latest_route(params, root, permitted=permitted) if selected_route is None else selected_route
   import time
   from openpilot.starpilot.flm.offroad_route import MAX_ANALYSIS_SECONDS, MAX_SEGMENTS, LOCAL_ID
   from openpilot.starpilot.galaxy.drive_history import DriveHistory
@@ -415,7 +415,7 @@ def collect_latest_steering_inputs(params, root, *, permitted):
       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']):
+  if selected_route is None and 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')
@@ -438,8 +438,11 @@ def collect_latest_steering_inputs(params, root, *, permitted):
   return route, collected
 
 
-def analyze_latest(params, root, previous, *, permitted):
-  route, collected = collect_latest_steering_inputs(params, root, permitted=permitted)
+def analyze_latest(params, root, previous, *, permitted, selected_route=None):
+  if selected_route is None:
+    route, collected = collect_latest_steering_inputs(params, root, permitted=permitted)
+  else:
+    route, collected = collect_latest_steering_inputs(params, root, permitted=permitted, selected_route=selected_route)
   receipt, state = collected.receipt, collected.state
   if type(receipt) not in (RouteReceipt, SelectedSourceReceipt):
     raise ValueError('An original whole-drive or explicit input-selected receipt is required')
@@ -548,3 +551,4 @@ def analyze_latest(params, root, previous, *, permitted):
   return CompletedAssessment(
     receipt, reports, counts, state['cp'], (), route, tuple(quality_stops), tuple(diagnostic), tuple(stable), _recorded_tracking(proposal_windows)
   )
+

openpilot/starpilot/flm/offroad_worker.py

@@ -17,7 +17,7 @@ from openpilot.starpilot.flm.learned_lifecycle import decode
 from openpilot.starpilot.flm.offroad_cache import AssessmentCache
 from openpilot.starpilot.flm.offroad_checks import qualify
 from openpilot.starpilot.flm.offroad_processor import analyze_latest, CompletedAssessment
-from openpilot.starpilot.flm.offroad_route import latest_route
+from openpilot.starpilot.flm.offroad_route import latest_route, LOCAL_ID, MAX_SEGMENTS
 from openpilot.starpilot.flm.offroad_runtime import configuration_scope
 from openpilot.starpilot.flm.offroad_store import RouteStorage, KEY, MAX_BYTES, MAX_ROUTE_STEP, NAMESPACE
 from openpilot.starpilot.flm.offroad_surface import REVISION
@@ -151,7 +151,29 @@ 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, *, trial_started=True):
+def pending_routes(root, latest):
+  """Newest first, then two preceding logger inputs, oldest first; no fit selection."""
+  if not LOCAL_ID.fullmatch(latest['routeId']):
+    return [latest]
+  from openpilot.starpilot.galaxy.drive_history import DriveHistory
+
+  inventory = DriveHistory(root, max_segments_per_route=MAX_SEGMENTS).snapshot()
+  if inventory['scanIncomplete']:
+    raise ValueError('Completed-drive backlog inventory incomplete')
+  counter = int(latest['routeId'].split('--')[0], 16)
+  candidates = []
+  for route in inventory['routes']:
+    if not LOCAL_ID.fullmatch(route['routeId']):
+      continue
+    distance = (counter - int(route['routeId'].split('--')[0], 16)) % 2**32
+    if 1 <= distance <= 2 and route.get('segments') and all(s['files']['rlog'] for s in route['segments']):
+      candidates.append((distance, route['routeId'], route))
+  if len({distance for distance, _, _ in candidates}) != len(candidates):
+    raise ValueError('Ambiguous completed logger counter')
+  return [latest] + [route for _, _, route in sorted(candidates, reverse=True)]
+
+
+def assess_next_drive(trial, assessment, scope, *, trial_started=True, recorded_backlog=False):
   """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')
@@ -163,6 +185,22 @@ def assess_next_drive(trial, assessment, scope, *, trial_started=True):
       'causal_road_benefit_proven': False,
       'comparisons': [],
     }
+  boundary = trial.get('configured_after_route')
+  if boundary is None and recorded_backlog:
+    return {
+      'outcome': 'pending', 'reason': 'trial_configuration_chronology_unavailable',
+      'counts_as_new_drive': False, 'observed_only': True, 'causal_road_benefit_proven': False, 'comparisons': [],
+    }
+  if boundary is not None:
+    observed = assessment.receipt.route_id
+    if not LOCAL_ID.fullmatch(boundary) or not LOCAL_ID.fullmatch(observed):
+      raise ValueError('Exact local trial chronology required')
+    distance = (int(observed.split('--')[0], 16) - int(boundary.split('--')[0], 16)) % 2**32
+    if not 0 < distance < 2**31:
+      return {
+        'outcome': 'pending', 'reason': 'recorded_drive_precedes_trial_configuration',
+        'counts_as_new_drive': False, 'observed_only': True, 'causal_road_benefit_proven': False, 'comparisons': [],
+      }
   before = trial['original_metadata']
   comparisons = []
   stable = trial['before']['training'].get('kind') == 'uncommanded_residual_v1'
@@ -345,6 +383,7 @@ class OffroadWorker:
     self.permitted, self.assessor, self.checker, self.scope_for = permitted, assessor, checker, scope_for
     self.source_revision = _digest([(p.name, hashlib.sha256(p.read_bytes()).hexdigest()) for p in sorted(Path(__file__).parent.glob('*.py'))])
     self.pending_input = None
+    self.pending_latest = None
     self.pending_failure_data = None
     self.on_analysis = on_analysis
     self.on_assessment = on_assessment
@@ -369,7 +408,7 @@ class OffroadWorker:
     if self.pending_input is None or not self.permitted():
       return
     route, fingerprint = self.pending_input
-    if self._snapshot(route) != fingerprint or latest_route(self.params, self.root, permitted=self.permitted) != route:
+    if self._snapshot(route) != fingerprint or latest_route(self.params, self.root, permitted=self.permitted) != self.pending_latest:
       return
     result = {
       'status': 'unavailable',
@@ -377,6 +416,7 @@ class OffroadWorker:
       'configuration_changed': False,
       'actual_application_proven': False,
       'road_improvement_confirmed': False,
+      'source_is_latest': route == self.pending_latest,
     }
     if pending is not None:
       authorized, recorded = pending
@@ -388,14 +428,24 @@ class OffroadWorker:
 
   def run_once(self):
     self.pending_failure_data = None
+    self.pending_input = None
     if not self.permitted():
       raise ValueError('Offroad analysis not permitted')
-    route = latest_route(self.params, self.root, permitted=self.permitted)
-    fingerprint = self._snapshot(route)
+    latest = latest_route(self.params, self.root, permitted=self.permitted)
+    self.pending_latest = latest
+    checkpoints = {row['input_fingerprint']: row['result'] for row in self.cache.checkpoints()}
+    fingerprints = []
+    for route in pending_routes(self.root, latest):
+      fingerprint = self._snapshot(route)
+      fingerprints.append(fingerprint)
+      if fingerprint not in checkpoints:
+        break
+    else:
+      last = self.cache.latest()
+      result = checkpoints[last['input_fingerprint']] if last is not None and last['input_fingerprint'] in fingerprints else checkpoints[fingerprints[0]]
+      source_is_latest = result['completed_route'] == latest['routeId'] if 'completed_route' in result else result.get('source_is_latest', True)
+      return dict(result, source_is_latest=source_is_latest)
     self.pending_input = (route, fingerprint)
-    last = self.cache.latest()
-    if last is not None and last.get('input_fingerprint') == fingerprint:
-      return last['result']
     state = {}
 
     def prior(cp):
@@ -409,7 +459,10 @@ class OffroadWorker:
 
     if self.on_analysis is not None:
       self.on_analysis()
-    assessment = self.assessor(self.params, self.root, prior, permitted=self.permitted)
+    if route == latest:
+      assessment = self.assessor(self.params, self.root, prior, permitted=self.permitted)
+    else:
+      assessment = self.assessor(self.params, self.root, prior, permitted=self.permitted, selected_route=route)
     if self.on_assessment is not None:
       self.on_assessment(assessment)
     if not state:
@@ -418,12 +471,11 @@ class OffroadWorker:
 
     def authorized():
       if not (
-        self.permitted() and self.scope_for(assessment.cp, self.params) == scope and latest_route(self.params, self.root, permitted=self.permitted) == route
+        self.permitted() and self.scope_for(assessment.cp, self.params) == scope and latest_route(self.params, self.root, permitted=self.permitted) == latest
       ):
         return False
-      if assessment.source_inventory == route:
+      if route == latest and 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()
@@ -439,6 +491,7 @@ class OffroadWorker:
       'vehicle': str(assessment.cp.carFingerprint),
       'analysis_counts': dict(assessment.counts),
       'recorded_tracking': assessment.recorded_tracking,
+      'source_is_latest': assessment.receipt.route_id == latest['routeId'],
     }
     self.pending_failure_data = (authorized, recorded)
     report = assessment.report()
@@ -478,8 +531,9 @@ class OffroadWorker:
       attempts = self.cache.get(row['analysis_sha256']).get('inconclusive_drives', 0)
       if type(attempts) is not int or not 0 <= attempts < MAX_INCONCLUSIVE_DRIVES:
         raise ValueError('Bounded inconclusive trial history required')
-      outcome = assess_next_drive(trial, assessment, scope, trial_started=started)
-      if outcome['outcome'] == 'pending' and started:
+      outcome = assess_next_drive(trial, assessment, scope, trial_started=started, recorded_backlog=route != latest)
+      counts_as_new_drive = outcome.get('counts_as_new_drive', True)
+      if outcome['outcome'] == 'pending' and started and counts_as_new_drive:
         attempts += 1
         if attempts >= MAX_INCONCLUSIVE_DRIVES:
           outcome = dict(outcome, outcome='rolled_back', reason='trial_inconclusive_drive_limit')
@@ -491,7 +545,7 @@ class OffroadWorker:
           'causal_road_benefit_proven': False,
           'comparisons': outcome['comparisons'],
         }
-      if outcome['outcome'] == 'pending' and not started:
+      if outcome['outcome'] == 'pending' and (not started or not counts_as_new_drive):
         status, reason = 'ready_for_next_drive', outcome['reason']
         report['unused_recording_review'] = outcome
         report['recorded_tracking'] = trial.get('assessment', {}).get('recorded_tracking')
@@ -675,6 +729,7 @@ class OffroadWorker:
             if checked['qualified']:
               record = {
                 'kind': 'next_drive_trial',
+                'configured_after_route': latest['routeId'],
                 'policy': POLICY,
                 'assessment': report,
                 'parameter': proposal['parameter'],
@@ -762,6 +817,7 @@ class OffroadWorker:
       'configuration_changed': saved,
       'policy': POLICY,
       'completed_route': assessment.receipt.route_id,
+      'source_is_latest': assessment.receipt.route_id == latest['routeId'],
       'analysis_counts': dict(assessment.counts),
       'actual_application_proven': False,
       'road_improvement_confirmed': False,
@@ -798,3 +854,4 @@ class OffroadWorker:
     self.cache.checkpoint({'input_fingerprint': self._snapshot(route), 'result': result})
     self.pending_failure_data = None
     return result
+

openpilot/starpilot/flm/tests/test_offroad_coordinate_policy.py

@@ -16,6 +16,142 @@ from openpilot.starpilot.flm.torque_stages import StageFrame
 from openpilot.starpilot.saved_source import read_saved
 
 
+class TestCompletedDriveBacklog(unittest.TestCase):
+  @staticmethod
+  def route(counter):
+    return {'routeId': f'{counter:08x}--aaaaaaaaaa', 'segments': [{'files': {'rlog': True}}]}
+
+  def test_input_order_ignores_outcomes_and_refuses_ambiguous_or_partial_inventory(self):
+    from openpilot.starpilot.flm.offroad_worker import pending_routes
+
+    routes = [self.route(i) for i in (16, 18, 17, 14, 19)]
+    inventory = {'scanIncomplete': False, 'routes': routes}
+    with patch('openpilot.starpilot.galaxy.drive_history.DriveHistory.snapshot', return_value=inventory):
+      self.assertEqual(pending_routes('.', self.route(18)), [self.route(i) for i in (18, 16, 17)])
+      inventory['routes'].append(dict(self.route(17), routeId='00000011--bbbbbbbbbb'))
+      with self.assertRaisesRegex(ValueError, 'Ambiguous'):
+        pending_routes('.', self.route(18))
+      inventory['scanIncomplete'] = True
+      with self.assertRaisesRegex(ValueError, 'incomplete'):
+        pending_routes('.', self.route(18))
+
+  def test_checkpoints_survive_newest_then_backlog_without_repeating_failed_inputs(self):
+    from openpilot.starpilot.flm import offroad_worker as module
+
+    latest, older = self.route(18), self.route(17)
+    seen = []
+    with tempfile.TemporaryDirectory() as directory, patch.object(module, 'latest_route', return_value=latest), patch.object(
+      module, 'pending_routes', return_value=[latest, older]
+    ):
+      params = Params(directory)
+
+      def assess(params, root, previous, *, permitted, selected_route=None):
+        seen.append(latest if selected_route is None else selected_route)
+        raise ValueError('Declared incomplete source; no tune permitted')
+
+      worker = module.OffroadWorker(params, directory, Path(directory) / 'cache', permitted=lambda: True, assessor=assess)
+      for route in (latest, older):
+        with self.assertRaises(ValueError) as caught:
+          worker.run_once()
+        result = worker.record_failure(caught.exception)
+        self.assertIsNotNone(result)
+        assert result is not None
+        self.assertFalse(result['configuration_changed'])
+        self.assertEqual(result['source_is_latest'], route == latest)
+      fresh = module.OffroadWorker(params, directory, worker.cache, permitted=lambda: True, assessor=assess)
+      self.assertEqual(fresh.run_once()['reason'], 'Declared incomplete source; no tune permitted')
+      self.assertEqual(seen, [latest, older], 'Fresh readers must not repeat either completed input')
+      params.put('AutoSteeringLearningReset', '1', block=True)
+      with self.assertRaises(ValueError):
+        fresh.run_once()
+      self.assertEqual(seen[-1], latest, 'Changed settings require a new assessment, newest first')
+      self.assertEqual(read_saved(params, KEY, MAX_BYTES), (None, True))
+      changed = self.route(19)
+      with patch.object(module, 'latest_route', return_value=changed):
+        self.assertIsNone(fresh.record_failure(ValueError('Cancelled by new drive')))
+      self.assertIs(result['source_is_latest'], False)
+
+  def test_completed_backlog_analysis_keeps_its_own_source_and_is_not_repeated(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
+
+    latest, older = self.route(18), self.route(17)
+    calls = []
+
+    def assess(params, root, previous, *, permitted, selected_route=None):
+      route = latest if selected_route is None else selected_route
+      calls.append(route)
+      cp = SimpleNamespace(carFingerprint='declared-queue-car')
+      previous(cp)
+      receipt = RouteReceipt(route['routeId'], ('d' if route == latest else 'e') * 64, (), 'a' * 64, 1, 2, 3, ())
+      return CompletedAssessment(receipt, (), {}, cp, (), route)
+
+    with tempfile.TemporaryDirectory() as directory, patch.object(module, 'latest_route', return_value=latest), patch.object(
+      module, 'pending_routes', return_value=[latest, older]
+    ), patch('openpilot.starpilot.galaxy.drive_history.DriveHistory.snapshot', return_value={'scanIncomplete': False, 'routes': [latest, older]}):
+      params = Params(directory)
+      worker = module.OffroadWorker(params, directory, Path(directory) / 'cache', permitted=lambda: True, assessor=assess,
+                                    scope_for=lambda cp, params: scope())
+      for route in (latest, older):
+        result = worker.run_once()
+        self.assertEqual(result['completed_route'], route['routeId'])
+        self.assertEqual(result['source_is_latest'], route == latest)
+        self.assertFalse(result['configuration_changed'])
+      self.assertEqual(worker.run_once(), result)
+      self.assertEqual(calls, [latest, older])
+      self.assertEqual(read_saved(params, KEY, MAX_BYTES), (None, True))
+
+  def test_checkpoint_history_is_bounded_and_cannot_reset_a_consumed_trial(self):
+    from openpilot.starpilot.flm.offroad_cache import AssessmentCache, MAX_CHECKPOINTS
+
+    with tempfile.TemporaryDirectory() as directory:
+      cache = AssessmentCache(directory)
+      self.assertTrue(cache.claim_trial('a' * 64, 'b' * 32))
+      for i in range(MAX_CHECKPOINTS + 2):
+        cache.checkpoint({'input_fingerprint': f'{i:064x}', 'result': {'status': 'collecting', 'reason': 'declared'}})
+      fresh = AssessmentCache(directory)
+      self.assertEqual(len(fresh.checkpoints()), MAX_CHECKPOINTS)
+      self.assertEqual(fresh.checkpoints()[-1], fresh.latest())
+      self.assertFalse(fresh.claim_trial('a' * 64, 'c' * 32))
+      rows = fresh.checkpoints()
+      (Path(directory) / 'inputs.json').write_text(json.dumps([rows[0], rows[0]]))
+      with self.assertRaisesRegex(ValueError, 'Duplicate'):
+        fresh.checkpoints()
+
+  def test_selected_backlog_does_not_substitute_an_inactive_source(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.flm import offroad_processor as module
+
+    route = self.route(17)
+    inactive = SimpleNamespace(receipt=SimpleNamespace(cp_sha256='a' * 64), counts={
+      'recorded_lateral_request_rows': 100, 'active_lateral_request_rows': 0, 'invalid_lateral_request_rows': 0,
+    }, quality_stops=[])
+    with patch.object(module, 'latest_route', side_effect=AssertionError('Explicit selected input must stay frozen')), patch.object(
+      module, 'collect_route_inputs', return_value=inactive
+    ) as collect:
+      actual, result = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True, selected_route=route)
+    self.assertEqual(actual, route)
+    self.assertIs(result, inactive)
+    self.assertEqual(collect.call_count, 1)
+
+  def test_preconfiguration_recordings_cannot_consume_a_next_drive_attempt(self):
+    from types import SimpleNamespace
+    from openpilot.starpilot.flm.offroad_worker import assess_next_drive, POLICY
+
+    declared = dict(trial(), policy=POLICY, values=values(0.01), original_metadata={}, configured_after_route=self.route(18)['routeId'])
+    for counter in (17, 18):
+      assessment = SimpleNamespace(receipt=SimpleNamespace(route_id=self.route(counter)['routeId']), quality_stop_markers=())
+      outcome = assess_next_drive(declared, assessment, scope(), trial_started=True, recorded_backlog=True)
+      self.assertEqual(outcome['outcome'], 'pending')
+      self.assertFalse(outcome['counts_as_new_drive'])
+    newer = SimpleNamespace(receipt=SimpleNamespace(route_id=self.route(19)['routeId']), quality_stop_markers=(), groups=(), stable_groups=())
+    self.assertEqual(assess_next_drive(declared, newer, scope(), trial_started=True)['reason'], 'trial_started_waiting_for_comparable_drive')
+    del declared['configured_after_route']
+    self.assertFalse(assess_next_drive(declared, newer, scope(), recorded_backlog=True)['counts_as_new_drive'])
+
+
 class TestAllThirtyCoordinateProposals(unittest.TestCase):
   @staticmethod
   def observations(bucket, speed, side='left', severity=1.0):
@@ -869,3 +1005,4 @@ class TestSelectedCurrentDriveInputs(unittest.TestCase):
     self.assertEqual(actual, route)
     self.assertIs(result, collected)
     self.assertIsNone(selected.call_args.kwargs['cp_sha256'])
+

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "2fec230e89e99e049a0c7888fd333e6ceb7109f7bf9cabf64a438c59abc8156a", "schema_version": 1}
+{"protected_inventory_sha256": "05fb38c744310b1ab587699683f13d1673234ff934f4474b0a0f3f5149ac4f41", "schema_version": 1}