Fix Laboratory startup and steering availability display

.github/workflows/flm-offroad.yaml

@@ -64,22 +64,16 @@ jobs:
           openpilot/starpilot/flm/tests/test_offroad_backend.py
           openpilot/starpilot/flm/tests/test_offroad_coordinate_policy.py
           openpilot/starpilot/flm/tests/test_offroad_equal_native_output.py
-          openpilot/starpilot/flm/tests/test_offroad_behavior.py
-          openpilot/starpilot/flm/tests/test_offroad_history_safety.py
           --json-output ci-results/flm-offroad.json
       - name: Test completed-drive and steering-result presentation
         timeout-minutes: 1
         run: |
           node openpilot/starpilot/galaxy/tests/test_flm_training_results.mjs
           node openpilot/starpilot/galaxy/tests/test_flm_tracking_results.mjs
-          node openpilot/starpilot/galaxy/tests/test_flm_review.mjs
-      - name: Test observed-behavior findings in Chromium
-        timeout-minutes: 2
-        run: node openpilot/starpilot/galaxy/tests/test_flm_ux_browser.mjs
       - name: Retain offroad contract results
         if: always()
         uses: actions/upload-artifact@v4
         with:
           name: flm-offroad-contracts
-          path: ci-results/flm-*.json
+          path: ci-results/flm-offroad.json
           if-no-files-found: warn

openpilot/selfdrive/modeld/modeld.py

@@ -554,7 +554,8 @@ def main(demo=False):
       meta_extra = meta_main
 
     sm.update(0)
-    turn_supported = bool(model.npy.get('desire') is not None and model.npy['desire'].shape[-1] == 8)
+    turn_models = (model.lateral, model.longitudinal) if isinstance(model, LaboratoryPair) else (model,)
+    turn_supported = all(state.npy.get('desire') is not None and state.npy['desire'].shape[-1] == 8 for state in turn_models)
     desire = navigation_turn.select(sm, time.monotonic_ns(), DH.desire,
                                      supported=turn_supported, model_stop=bool(prev_action.shouldStop))
     is_rhd = sm["driverMonitoringState"].isRHD

openpilot/starpilot/flm/offroad_analysis.py

@@ -46,7 +46,7 @@ def _spread(rows, count):
   return tuple(rows[i * (len(rows) - 1) // (count - 1)] for i in range(count))
 
 
-def analyze(windows, previous, *, permitted=lambda: True, route_order=None, comparison=None):
+def analyze(windows, previous, *, permitted=lambda: True):
   """Return source-bound suggestions; caller still owes controller qualification."""
   if (
     type(windows) is not tuple
@@ -55,7 +55,7 @@ def analyze(windows, previous, *, permitted=lambda: True, route_order=None, comp
     or set(previous) != set(NAMES)
     or any(type(w) is not RouteWindow or not w.valid() for w in windows)
     or len({w.identity for w in windows}) != len(windows)
-    or comparison is None and len({(w.domain, w.points[0].context_key) for w in windows}) > 1
+    or len({(w.domain, w.points[0].context_key) for w in windows}) > 1
   ):
     raise ValueError('Bounded, distinct, same-controller whole windows required')
   if not permitted():
@@ -67,14 +67,14 @@ def analyze(windows, previous, *, permitted=lambda: True, route_order=None, comp
       groups.setdefault(key, []).append(window)
   results = []
   for key, rows in sorted(groups.items()):
-    from openpilot.starpilot.flm.offroad_history import partition_windows
-
-    train, heldout = partition_windows(rows, route_order)
+    rows = sorted(rows, key=lambda w: (w.points[0].segment, w.points[0].mono_ns))
+    chosen = _spread(rows, MAX_COHORT_WINDOWS)
+    # Partition original whole windows before fitting and before selecting knobs.
+    train, heldout = chosen[::2], chosen[1::2]
     if len(train) < 3 or len(heldout) < 3:
       results.append({'cohort': key, 'whole_windows': len(rows), 'reason': 'more_independent_windows_required'})
       continue
-    first = estimate(tuple(train), cohort=key, permitted=permitted, comparison=comparison)
-    second = estimate(tuple(heldout), cohort=key, permitted=permitted, comparison=comparison)
+    first, second = estimate(tuple(train), cohort=key, permitted=permitted), estimate(tuple(heldout), cohort=key, permitted=permitted)
     if first is None or second is None:
       results.append({'cohort': key, 'whole_windows': len(rows), 'reason': 'reference_response_not_identified'})
       continue

openpilot/starpilot/flm/offroad_behavior.py

@@ -1,94 +0,0 @@
-"""Observed behavior recommendations, never an automatic trial or approval.
-
-The denominator is input-selected, nonoverlapping recorded sections. Adjacent
-windows join before counting; a section is explicitly NOT a distinct maneuver.
-Missing original inputs withhold the count rather than improve its hit rate.
-"""
-
-import hashlib
-from statistics import fmean
-
-from openpilot.starpilot.flm.offroad_coordinate_policy import coordinate_candidates, recorded_observation, _symptom_steps
-from openpilot.starpilot.flm.offroad_stable_policy import cadenced_runs, classify_current_stable
-
-SYMPTOMS = {
-  'understeer': 'steering response was weaker than requested',
-  'oversteer': 'steering response was stronger than requested',
-  'late_turn_in': 'steering built too slowly on entry',
-  'early_turn_in': 'steering built too quickly on entry',
-  'unwind_too_slow': 'steering unwound too slowly',
-  'unwind_too_fast': 'steering unwound too quickly',
-  'low_speed_unwillingness': 'low-speed steering lagged the request',
-  'center_chatter': 'repeated steering oscillation was measured',
-}
-PHASE_BUCKETS = {
-  'steady': ('understeer', 'oversteer'),
-  'turn_in': ('late_turn_in', 'early_turn_in', 'low_speed_unwillingness'),
-  'unwind': ('unwind_too_slow', 'unwind_too_fast'),
-  'center': (),
-}
-
-
-def behavior_candidates(windows, rows, contexts, previous, cohort, *, stable=False, permitted=lambda: True):
-  from openpilot.starpilot.flm.offroad_stable_bridge import profile_sections
-
-  if not 1 <= len(windows) <= 24 or len(set(windows)) != len(windows):
-    raise ValueError('Bounded distinct behavior observation windows required')
-  if rows is None or any(not rows.get(key) for key in windows):
-    return []
-  band, side, phase = cohort.split(':')
-  allowed = ('center_chatter',) if stable else PHASE_BUCKETS.get(phase, ())
-  if not allowed:
-    return []
-  sections, _ = profile_sections(windows, rows, contexts)
-  observations, measured, routes = [], [], set()
-  for section in sections:
-    if not permitted():
-      raise ValueError('Parked behavior analysis cancelled')
-    # One contiguous source section earns at most one vote per symptom, even
-    # if the classifier reports several bursts or the section spans windows.
-    key = hashlib.sha256(repr(tuple((s.route, s.segment, s.t) for s in section)).encode()).hexdigest()
-    measured.append(key)
-    routes.add(section[0].route)
-    found = {}
-    for run in cadenced_runs(section):
-      summaries, _ = classify_current_stable(list(run))
-      for summary in summaries:
-        if summary['bucket'] not in allowed:
-          continue
-        for event in summary['events']:
-          if event['direction'] != side and summary['bucket'] != 'center_chatter':
-            continue
-          samples = run[event['startIdx']:event['endIdx'] + 1]
-          observation = recorded_observation(summary['bucket'], event['direction'], event['peakScore'], fmean(s.v_ego for s in samples))
-          pair = (summary['bucket'], event['direction'])
-          if pair not in found or found[pair]['severity'] < observation['severity']:
-            found[pair] = observation
-    observations.extend((key, observation) for observation in found.values())
-  # Require repeated support across sections, not many correlated samples.
-  # Directional conflicts are retained in the vote set so they cannot create
-  # a deceptively precise coordinate recommendation.
-  candidates = coordinate_candidates(observations, previous)
-  result = []
-  for candidate in candidates:
-    votes = set(candidate['original_windows'])
-    if len(votes) * 5 < len(measured) * 3:
-      continue
-    name = candidate['parameter'].removeprefix('stable:')
-    supporting = [(key, row) for key, row in observations if key in votes and name in _symptom_steps(row, previous)]
-    buckets = {row['bucket'] for _, row in supporting}
-    # Each displayed ratio describes one symptom, not a sum of unlike events.
-    for bucket in sorted(buckets):
-      matched = {key for key, row in supporting if row['bucket'] == bucket}
-      if len(matched) < 3 or len(matched) * 5 < len(measured) * 3:
-        continue
-      speed = {'0': 'low-speed', '1': 'medium-speed', '2': 'higher-speed'}[band]
-      condition = f'{speed} {side} ' + {'turn_in': 'turn-entry', 'unwind': 'turn-exit', 'steady': 'curve', 'center': 'center'}[phase]
-      evidence = (
-        f'{len(matched)} of {len(measured)} retained {condition} sections across {len(routes)} drive(s): {SYMPTOMS[bucket]}. '
-        + 'These are recorded sections, not a count of distinct turns. This is a possible adjustment, not a confirmed cause or improvement.'
-      )
-      result.append(dict(candidate, evidence=evidence, measured_sections=len(measured), matched_sections=len(matched),
-                         evidence_drives=len(routes), behavior=bucket, cohort=cohort, whole_windows=len(windows)))
-      break
-  return result

openpilot/starpilot/flm/offroad_cache.py

@@ -71,9 +71,9 @@ class AssessmentCache:
       raise ValueError('Assessment object required')
     return value
 
-  def _write(self, name, raw, *, maximum=MAX_DOCUMENT):
+  def _write(self, name, raw):
     self._owned()
-    if not 0 < len(raw) <= maximum:
+    if not 0 < len(raw) <= MAX_DOCUMENT:
       raise ValueError('Assessment document bound')
     temporary = None
     try:

openpilot/starpilot/flm/offroad_checks.py

@@ -5,7 +5,7 @@ from typing import Any
 from openpilot.starpilot.flm.offroad_controller_checks import check_candidate, check_original_native
 from openpilot.starpilot.flm.offroad_route import RETAIN
 from openpilot.starpilot.flm.log_decode import decode_full_rlog_segment
-from openpilot.starpilot.galaxy.drive_analysis import _closed_source, _Incomplete
+from openpilot.starpilot.galaxy.drive_analysis import _closed_source
 
 
 TRAINING_FRACTIONS = (1.0, 0.8, 0.6, 0.4, 0.2, 0.1, 0.05)
@@ -73,12 +73,12 @@ def fit_independent_route(assessment, group, cohort, prior, proposal, previous,
   train = tuple(roles['training']['windows']) + tuple(roles['heldout']['windows'])
   original = group['qualification_original_windows'] if 'qualification_roles' in cohort else group['selected_original_windows']
   validation = tuple(row['identity'] for row in prior.windows)
-  source_receipts = (assessment.receipt, *(source.receipt for source in getattr(assessment, 'evidence_sources', {}).values()))
   if (
     not all(3 <= len(roles[role]['windows']) <= 12 for role in ('training', 'heldout'))
     or not 6 <= len(validation) <= 12
     or len(set(train + validation)) != len(train + validation)
-    or any(receipt.source_sha256 == prior.receipt.source_sha256 or receipt.route_id == prior.receipt.route_id for receipt in source_receipts)
+    or assessment.receipt.source_sha256 == prior.receipt.source_sha256
+    or assessment.receipt.route_id == prior.receipt.route_id
     or prior.receipt.completed_wall_ns >= assessment.receipt.completed_wall_ns
     or prior.receipt.cp_sha256 != assessment.receipt.cp_sha256
     or car_params_digest(prior.cp) != assessment.receipt.cp_sha256
@@ -217,27 +217,12 @@ def _release_unused_analysis_heap():
       trim(0)
 
 
-def _original_results(assessment, original, ids, values, root, *, permitted, source_batch=False):
+def _original_results(assessment, original, ids, values, root, *, permitted):
   spans = {w['identity']: w for w in original if w['identity'] in ids}
-  if not (1 if source_batch else 3) <= len(ids) <= 24 or len(set(ids)) != len(ids) or set(spans) != set(ids):
+  if not 3 <= len(ids) <= 24 or len(set(ids)) != len(ids) or set(spans) != set(ids):
     raise ValueError('Every preselected original window required')
   if not permitted():
     raise ValueError('Parked checks cancelled')
-  if getattr(assessment, 'evidence_sources', None):
-    results = []
-    if any(w.get('route') not in assessment.evidence_sources for w in spans.values()):
-      raise ValueError('Every historical window needs its own original source')
-    for route, source in sorted(assessment.evidence_sources.items()):
-      selected = tuple(key for key in ids if spans[key]['route'] == route)
-      if selected:
-        try:
-          checked = _original_results(source, original, selected, values, root, permitted=permitted, source_batch=True)
-        except (FileNotFoundError, _Incomplete) as error:
-          if not permitted() or isinstance(error, _Incomplete) and str(error) == 'Analysis cancelled':
-            raise
-          checked = [{'window': key, 'qualified': False, 'reason': 'historical_dense_source_unavailable'} for key in selected]
-        results.extend(dict(row, route=route, source_sha256=source.receipt.source_sha256) for row in checked)
-    return results
   _release_unused_analysis_heap()
   from openpilot.starpilot.flm.learning_wire import read_control, read_applied
   import hashlib
@@ -316,7 +301,7 @@ def _original_results(assessment, original, ids, values, root, *, permitted, sou
       controls = []
       outputs = []
       missing = []
-  return [dict(row, route=receipt.route_id, source_sha256=receipt.source_sha256) for row in results]
+  return results
 
 
 def qualified_originals(results, cohort, *, require_change):

openpilot/starpilot/flm/offroad_daemon.py

@@ -103,8 +103,6 @@ class StatusReport:
       details['Joined steering observations'] = count('joined_tracking')
       details['Retained control windows'] = count('qualified_whole_windows')
       details['Retained tracking windows'] = count('stable_qualified_whole_windows')
-      details['Drives in steering evidence history'] = count('evidence_drives')
-      details['Control windows in steering evidence history'] = count('evidence_windows')
       details['Original controller observations'] = count('original_control_metadata_rows')
       details['Controller eligibility refusals'] = count('controller_fit_refusals')
       details['Automatic configuration changed'] = bool(value.get('configuration_changed', False))
@@ -122,7 +120,7 @@ class StatusReport:
       if 'recommendations' in value:
         recommendations = value['recommendations'][:35]
         details['Recorded coordinate recommendations'] = [
-          [s['parameter'], s['baseline'], s['target'], s['whole_windows'], s['reason'], s.get('evidence', '')]
+          [s['parameter'], s['baseline'], s['target'], s['whole_windows'], s['reason']]
           for s in recommendations
         ]
         details['Recorded route suggestions'] = [

openpilot/starpilot/flm/offroad_history.py

@@ -1,267 +0,0 @@
-"""Bounded cross-drive observations; original logs still supply native checks.
-
-Only small, input-selected tracking windows are retained. This is neither a log
-archive nor permission to use an old tune with a different controller/baseline.
-"""
-
-from dataclasses import asdict, dataclass, fields
-import hashlib
-import json
-import math
-from pathlib import Path
-import zlib
-
-from openpilot.starpilot.flm.offroad_analysis import cohort, _spread
-from openpilot.starpilot.flm.offroad_cache import AssessmentCache, _pairs
-from openpilot.starpilot.flm.offroad_evidence import RoutePoint, RouteWindow
-from openpilot.starpilot.flm.offroad_route import RouteReceipt, SelectedSourceReceipt
-
-MAX_DRIVES = 8
-MAX_WINDOWS_PER_DRIVE = 64
-MAX_PACKED = 8 * 1024 * 1024
-MAX_UNPACKED = 64 * 1024 * 1024
-POINT_FIELDS = tuple(f.name for f in fields(RoutePoint))
-
-
-@dataclass(frozen=True)
-class TrackingComparison:
-  originals: tuple
-
-  def original_context(self, window):
-    for key, metadata in self.originals:
-      recorded = {k: v for k, v in metadata.items() if k not in ('route_values', 'route_revision')}
-      if window.points[0].context_key == hashlib.sha256(repr(recorded).encode()).hexdigest():
-        return key, metadata
-    raise ValueError('Original window context missing')
-
-  def context(self, windows):
-    common = None
-    original_keys = set()
-    anchor_size = None
-    for original_key, metadata in self.originals:
-      if hashlib.sha256(repr(metadata).encode()).hexdigest() != original_key:
-        raise ValueError('Original numerical comparison binding changed')
-      anchor = metadata['vehicle_model_anchor']
-      if (len(anchor) not in (0, 3) or anchor_size is not None and len(anchor) != anchor_size
-          or any(type(v) not in (int, float) or not math.isfinite(v) or v <= 0 for v in anchor)):
-        raise ValueError('Original vehicle-model observation required')
-      anchor_size = len(anchor)
-      baseline = {k: v for k, v in metadata.items() if k != 'vehicle_model_anchor'}
-      if common is not None and baseline != common:
-        raise ValueError('Cross-drive vehicle/controller/tune baseline differs')
-      common = baseline
-      recorded = {k: v for k, v in metadata.items() if k not in ('route_values', 'route_revision')}
-      original_keys.add(hashlib.sha256(repr(recorded).encode()).hexdigest())
-    if common is None or any(w.points[0].context_key not in original_keys or w.domain != common['domain'] for w in windows):
-      raise ValueError('Every observed window needs an exact original context')
-    # Only the derived statistic shares a comparison key. Points, original
-    # identities and each controller/native replay retain their actual anchor.
-    return hashlib.sha256(repr(('cross_drive_observed_tracking_v1', common)).encode()).hexdigest()
-
-
-def statistical_inputs(collected):
-  from copy import copy
-
-  merged = copy(collected)
-  contexts, references, originals = {}, {}, {}
-  for name in ('windows', 'stable_windows', 'baseline_tracking'):
-    destination = {}
-    for key, windows in getattr(collected, name).items():
-      metadata = collected.context_metadata[key]
-      baseline = {k: v for k, v in metadata.items() if k != 'vehicle_model_anchor'}
-      comparison_key = hashlib.sha256(repr(baseline).encode()).hexdigest()
-      contexts.setdefault(comparison_key, metadata)
-      references.setdefault(comparison_key, {})[key] = metadata
-      for window in windows:
-        if window.identity in originals and originals[window.identity] != metadata:
-          raise ValueError('Conflicting original observation metadata')
-        originals[window.identity] = metadata
-      destination.setdefault(comparison_key, []).extend(windows)
-    setattr(merged, name, destination)
-  merged.context_metadata = contexts
-  comparisons = {key: TrackingComparison(tuple(rows.items())) for key, rows in references.items()}
-  return merged, comparisons, originals
-
-
-def bind_original_contexts(report, windows, comparison):
-  ids = {row['identity'] for name in ('selected_original_windows', 'qualification_original_windows', 'recommendation_original_windows')
-         for row in report.get(name, ())}
-  originals, bindings = {}, {}
-  for window in windows:
-    if window.identity in ids:
-      key, metadata = comparison.original_context(window)
-      originals[key] = metadata
-      bindings[window.identity] = key
-  report.update(original_contexts=originals, original_window_contexts=bindings)
-
-
-def _tuples(value):
-  return tuple(_tuples(v) for v in value) if type(value) is list else value
-
-
-def _metadata(rows):
-  # Preserve the original field order and tuple types used by source identities.
-  return {name: _tuples(value) for name, value in rows}
-
-
-def _window(row):
-  window = RouteWindow(row['identity'], tuple(RoutePoint(*point) for point in row['points']))
-  if not window.valid():
-    raise ValueError('Historical original tracking window changed')
-  return window
-
-
-def _receipt(row):
-  values = dict(row['receipt'])
-  values['segments'] = _tuples(values['segments'])
-  values['original_controller_contexts'] = _tuples(values['original_controller_contexts'])
-  if row['selected_sections']:
-    values['decoded_segments'] = _tuples(values['decoded_segments'])
-    return SelectedSourceReceipt(**values)
-  return RouteReceipt(**values)
-
-
-def partition_windows(rows, route_order=None, *, contiguous=False):
-  """Freeze source roles before fitting; different drives never share a role."""
-  if route_order is None:
-    chosen = _spread(sorted(rows, key=lambda w: (w.points[0].segment, w.points[0].mono_ns)), 24)
-    split = len(chosen) // 2
-    return (chosen[:split], chosen[split:]) if contiguous else (chosen[::2], chosen[1::2])
-  routes = sorted({w.points[0].route for w in rows}, key=lambda route: (route_order[route], route))
-  if len(routes) < 2:
-    return partition_windows(rows, contiguous=contiguous)
-  result = []
-  for sources in (set(routes[::2]), set(routes[1::2])):
-    ordered = sorted(
-      (w for w in rows if w.points[0].route in sources),
-      key=lambda w: (route_order[w.points[0].route], w.points[0].route, w.points[0].segment, w.points[0].mono_ns),
-    )
-    result.append(_spread(ordered, 12))
-  return tuple(result)
-
-
-class EvidenceBank:
-  def __init__(self, directory):
-    self.cache = AssessmentCache(Path(directory).with_name(Path(directory).name + '-evidence'))
-
-  def read(self, identity):
-    try:
-      packed = self.cache._read('history.z', MAX_PACKED)
-    except FileNotFoundError:
-      return []
-    decoder = zlib.decompressobj()
-    raw = decoder.decompress(packed, MAX_UNPACKED + 1)
-    if len(raw) > MAX_UNPACKED or not decoder.eof or decoder.unused_data:
-      raise ValueError('Historical steering evidence bound or framing changed')
-    value = json.loads(raw, object_pairs_hook=_pairs)
-    if value.get('version') != 1 or type(value.get('drives')) is not list or len(value['drives']) > MAX_DRIVES:
-      raise ValueError('Bounded historical steering evidence required')
-    if value.get('identity') != identity:
-      return []
-    rows = value['drives']
-    if len({row['receipt']['route_id'] for row in rows}) != len(rows):
-      raise ValueError('Historical drives cannot be counted twice')
-    return rows
-
-  def retain(self, collected, inventory, identity, *, permitted):
-    """Selection uses speed, side, phase and source order, never fit outcomes."""
-    buckets = []
-    for kind in ('windows', 'stable_windows', 'baseline_tracking'):
-      groups = {}
-      for context, windows in getattr(collected, kind).items():
-        for window in windows:
-          condition = cohort(window)
-          if condition is not None:
-            groups.setdefault((context, condition), []).append(window)
-      for (context, condition), windows in sorted(groups.items()):
-        ordered = sorted(windows, key=lambda w: (w.points[0].segment, w.points[0].mono_ns))
-        buckets.append((kind, context, condition, _spread(ordered, 4)))
-    retained = []
-    for offset in range(4):
-      for kind, context, condition, windows in buckets:
-        if offset < len(windows) and len(retained) < MAX_WINDOWS_PER_DRIVE:
-          window = windows[offset]
-          retained.append({
-            'kind': kind, 'context': context, 'condition': condition, 'identity': window.identity,
-            'points': [[getattr(point, name) for name in POINT_FIELDS] for point in window.points],
-            'bookends': collected.bookends.get(window.identity, ()),
-          })
-    row = {
-      'receipt': asdict(collected.receipt), 'selected_sections': type(collected.receipt) is SelectedSourceReceipt,
-      'inventory': inventory, 'windows': retained,
-      'metadata': {key: list(value.items()) for key, value in collected.context_metadata.items() if key in {w['context'] for w in retained}},
-      'quality_stops': collected.quality_stops,
-    }
-    history = self.read(identity)
-    previous = next((r for r in history if r['receipt']['route_id'] == collected.receipt.route_id), None)
-    if previous is not None and previous['receipt']['source_sha256'] != collected.receipt.source_sha256:
-      raise ValueError('A retained original drive changed')
-    history = [r for r in history if r['receipt']['route_id'] != collected.receipt.route_id] + [row]
-    history.sort(key=lambda r: (r['receipt']['completed_wall_ns'], r['receipt']['route_id']))
-    history = history[-MAX_DRIVES:]
-    raw = json.dumps({'version': 1, 'identity': identity, 'drives': history}, separators=(',', ':'), allow_nan=False).encode()
-    if len(raw) > MAX_UNPACKED:
-      raise ValueError('Historical steering evidence size bound')
-    packed = zlib.compress(raw, 6)
-    if len(packed) > MAX_PACKED or not permitted():
-      raise ValueError('Historical steering evidence bound or parked authority lost')
-    self.cache._write('history.z', packed, maximum=MAX_PACKED)
-    return history
-
-
-def combine(collected, inventory, history):
-  """Add exact-compatible original windows without rebinding a source identity."""
-  from copy import copy
-  from openpilot.starpilot.flm.offroad_processor import CompletedAssessment
-
-  pooled = copy(collected)
-  for name in ('windows', 'stable_windows', 'baseline_tracking'):
-    setattr(pooled, name, {key: list(values) for key, values in getattr(collected, name).items()})
-  pooled.context_metadata = dict(collected.context_metadata)
-  pooled.bookends = dict(collected.bookends)
-  sources = {collected.receipt.route_id: CompletedAssessment(collected.receipt, (), {}, collected.state['cp'], (), inventory)}
-  order = {collected.receipt.route_id: collected.receipt.completed_wall_ns}
-  for row in history:
-    receipt = _receipt(row)
-    if receipt.cp_sha256 != collected.receipt.cp_sha256 or receipt.route_id in sources or row['quality_stops']:
-      continue
-    if row['inventory'].get('routeId') != receipt.route_id or len(row['windows']) > MAX_WINDOWS_PER_DRIVE:
-      raise ValueError('Historical route inventory or window bound changed')
-    metadata = {key: _metadata(value) for key, value in row['metadata'].items()}
-    for key, value in metadata.items():
-      if hashlib.sha256(repr(value).encode()).hexdigest() != key:
-        raise ValueError('Historical original context changed')
-      if key in pooled.context_metadata and pooled.context_metadata[key] != value:
-        raise ValueError('Conflicting historical controller context')
-      pooled.context_metadata[key] = value
-    sources[receipt.route_id] = CompletedAssessment(receipt, (), {}, collected.state['cp'], (), row['inventory'])
-    order[receipt.route_id] = receipt.completed_wall_ns
-    seen = set()
-    for record in row['windows']:
-      if not permitted_window(record, metadata, receipt):
-        raise ValueError('Historical source window binding changed')
-      window = _window(record)
-      binding = (record['kind'], window.identity)
-      if binding in seen:
-        raise ValueError('Historical observations cannot be counted twice')
-      seen.add(binding)
-      getattr(pooled, record['kind']).setdefault(record['context'], []).append(window)
-      pooled.bookends[window.identity] = tuple(record['bookends'])
-  return pooled, sources, order
-
-
-def permitted_window(row, metadata, receipt):
-  if row['kind'] not in ('windows', 'stable_windows', 'baseline_tracking') or row['context'] not in metadata:
-    return False
-  window = _window(row)
-  actual = metadata[row['context']]
-  comparison = dict(actual)
-  comparison.pop('route_values')
-  comparison.pop('route_revision')
-  return bool(
-    actual['cp_sha256'] == receipt.cp_sha256
-    and window.points[0].route == receipt.route_id
-    and 0 <= window.points[0].segment < len(receipt.segments)
-    and window.points[0].context_key == hashlib.sha256(repr(comparison).encode()).hexdigest()
-    and cohort(window) == row['condition']
-  )

openpilot/starpilot/flm/offroad_prior_inputs.py

@@ -118,11 +118,9 @@ def _prior_originals(assessment, group, selected_cohort, root, *, permitted, sta
   if not LOCAL_ID.fullmatch(current):
     raise ValueError('Original local latest route required')
   counter = int(current.split('--')[0], 16)
-  used_routes = {current, *(source.receipt.route_id for source in getattr(assessment, 'evidence_sources', {}).values())}
-  used_digests = {assessment.receipt.source_sha256, *(source.receipt.source_sha256 for source in getattr(assessment, 'evidence_sources', {}).values())}
   candidates = []
   for row in inventory['routes']:
-    if row['routeId'] not in used_routes and LOCAL_ID.fullmatch(row['routeId']):
+    if LOCAL_ID.fullmatch(row['routeId']):
       distance = (counter - int(row['routeId'].split('--')[0], 16)) % 2**32
       if 1 <= distance <= 3 and row['segments'] and all(s['files']['rlog'] for s in row['segments']):
         candidates.append((distance, row['routeId'], row))
@@ -141,8 +139,6 @@ def _prior_originals(assessment, group, selected_cohort, root, *, permitted, sta
       collected = collect_route_inputs(root, route, permitted=allowed, walker=walk)
     except PriorIdentityMismatch:
       continue
-    if collected.receipt.route_id in used_routes or collected.receipt.source_sha256 in used_digests:
-      continue
     if collected.receipt.completed_wall_ns >= assessment.receipt.completed_wall_ns:
       raise ValueError('Independent prior chronology invalid')
     compatible = []
@@ -176,7 +172,6 @@ def _prior_originals(assessment, group, selected_cohort, root, *, permitted, sta
         chosen.append(
           {
             'identity': w.identity,
-            'route': w.points[0].route,
             'segment': number,
             'start_ns': w.points[0].mono_ns,
             'end_ns': w.points[-1].mono_ns,
@@ -198,10 +193,7 @@ def _prior_originals(assessment, group, selected_cohort, root, *, permitted, sta
         )
         if candidates:
           w = candidates[0]
-          profile_spans.append({
-            'identity': w.identity, 'route': w.points[0].route, 'segment': number,
-            'start_ns': w.points[0].mono_ns, 'end_ns': w.points[-1].mono_ns,
-          })
+          profile_spans.append({'identity': w.identity, 'segment': number, 'start_ns': w.points[0].mono_ns, 'end_ns': w.points[-1].mono_ns})
     if stable:
       chosen = _spread(chosen, 12)
       tracking = _spread(tracking, 12)

openpilot/starpilot/flm/offroad_processor.py

@@ -27,8 +27,6 @@ class CompletedAssessment:
   diagnostic_groups: tuple = ()
   stable_groups: tuple = ()
   recorded_tracking: dict | None = None
-  evidence_sources: dict = field(default_factory=dict, repr=False, compare=False)
-  single_drive: object = field(default=None, repr=False, compare=False)
 
   def report(self):
     return {
@@ -42,7 +40,6 @@ class CompletedAssessment:
       'diagnostic_groups': list(self.diagnostic_groups),
       'stable_groups': list(self.stable_groups),
       'recorded_tracking': self.recorded_tracking,
-      'evidence_sources': [asdict(source.receipt) for _, source in sorted(self.evidence_sources.items())],
       'source_role': (
         'Input-selected original drive sections; not a whole-drive assessment or live permission.'
         if type(self.receipt) is SelectedSourceReceipt
@@ -71,39 +68,20 @@ def window_demand(window):
   }
 
 
-def _separate_qualification(report, continuous, *, proposal_windows=None, route_order=None, role_windows=None):
+def _separate_qualification(report, continuous, *, proposal_windows=None):
   qualification = {}
   spans = {} if proposal_windows is None else {w.identity: w for w in proposal_windows}
   for c in report['cohorts']:
     native = [w for w in continuous if cohort(w) == c['cohort']]
     ordered = sorted(native, key=lambda w: (w.points[0].segment, w.points[0].mono_ns))
-    if route_order is not None and role_windows is not None and all(role in c for role in ('training', 'heldout')):
-      owners = {w.identity: w.points[0].route for w in role_windows}
-      routes = {role: {owners[key] for key in c[role]['windows']} for role in ('training', 'heldout')}
-      if routes['training'].isdisjoint(routes['heldout']):
-        c['qualification_roles'] = {}
-        for role in ('training', 'heldout'):
-          chosen = _spread([w for w in ordered if w.points[0].route in routes[role]], MAX_COHORT_WINDOWS // 2)
-          c['qualification_roles'][role] = {'windows': tuple(w.identity for w in chosen)}
-          qualification.update((w.identity, w) for w in chosen)
-        continue
     if proposal_windows is None:
-      from openpilot.starpilot.flm.offroad_history import partition_windows
-
-      train, heldout = partition_windows(ordered, route_order)
-      c['qualification_roles'] = {role: {'windows': tuple(w.identity for w in rows)} for role, rows in (('training', train), ('heldout', heldout))}
-      qualification.update((w.identity, w) for w in (*train, *heldout))
+      chosen = _spread(ordered, MAX_COHORT_WINDOWS)
+      c['qualification_roles'] = {role: {'windows': tuple(w.identity for w in chosen[offset::2])} for role, offset in (('training', 0), ('heldout', 1))}
+      qualification.update((w.identity, w) for w in chosen)
     elif all(role in c for role in ('training', 'heldout')):
       # A native check must stay in the SAME original data role as the
       # corresponding profile. Joining observations never joins holdout data.
       c['qualification_roles'] = {}
-      routes = {role: {spans[key].points[0].route for key in c[role]['windows']} for role in ('training', 'heldout')}
-      if routes['training'].isdisjoint(routes['heldout']):
-        for role in ('training', 'heldout'):
-          chosen = _spread([w for w in ordered if w.points[0].route in routes[role]], MAX_COHORT_WINDOWS // 2)
-          c['qualification_roles'][role] = {'windows': tuple(w.identity for w in chosen)}
-          qualification.update((w.identity, w) for w in chosen)
-        continue
       training_end = max((spans[key].points[-1].segment, spans[key].points[-1].mono_ns) for key in c['training']['windows'])
       heldout_start = min((spans[key].points[0].segment, spans[key].points[0].mono_ns) for key in c['heldout']['windows'])
       if training_end >= heldout_start:
@@ -120,7 +98,6 @@ def _separate_qualification(report, continuous, *, proposal_windows=None, route_
   report['qualification_original_windows'] = [
     {
       'identity': w.identity,
-      'route': w.points[0].route,
       'segment': w.points[0].segment,
       'start_ns': w.points[0].mono_ns,
       'end_ns': w.points[-1].mono_ns,
@@ -156,21 +133,6 @@ def _recorded_tracking(windows):
   }
 
 
-def _recommendation_windows(windows):
-  """Bounded input-only spans, kept separately from automatic trial roles."""
-  groups = {}
-  for window in windows:
-    key = cohort(window)
-    if key is not None:
-      groups.setdefault(key, []).append(window)
-  return [
-    {'identity': w.identity, 'route': w.points[0].route, 'segment': w.points[0].segment,
-     'start_ns': w.points[0].mono_ns, 'end_ns': w.points[-1].mono_ns, 'cohort': key}
-    for key, rows in sorted(groups.items())
-    for w in _spread(sorted(rows, key=lambda w: (w.points[0].route, w.points[0].segment, w.points[0].mono_ns)), 24)
-  ]
-
-
 def _control_source_current(control, stamp):
   return bool(
     control is not None
@@ -476,40 +438,17 @@ def collect_latest_steering_inputs(params, root, *, permitted, selected_route=No
   return route, collected
 
 
-def analyze_latest(params, root, previous, *, permitted, selected_route=None, evidence_bank=None, history_identity=None):
+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)
-  single = _assess_inputs(route, collected, previous, permitted=permitted)
-  if evidence_bank is None:
-    return single
-  if not callable(history_identity):
-    raise ValueError('Current source/settings history identity required')
-  from openpilot.starpilot.flm.offroad_history import combine
-
-  identity = history_identity(collected.state['cp'])
-  history = evidence_bank.retain(collected, route, identity, permitted=permitted)
-  pooled, sources, order = combine(collected, route, history)
-  result = _assess_inputs(route, pooled, previous, permitted=permitted, sources=sources, route_order=order)
-  return replace(result, single_drive=single, recorded_tracking=single.recorded_tracking)
-
-
-def _assess_inputs(route, collected, previous, *, permitted, sources=None, route_order=None):
-  sources = {} if sources is None else sources
-  comparisons = {}
-  if len(sources) > 1:
-    from openpilot.starpilot.flm.offroad_history import statistical_inputs
-
-    collected, comparisons, _ = statistical_inputs(collected)
   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')
   windows, stable_windows, baseline_tracking = collected.windows, collected.stable_windows, collected.baseline_tracking
   bookends, raw_windows, context_metadata = collected.bookends, collected.raw_windows, collected.context_metadata
   counts, quality_stops = collected.counts, collected.quality_stops
-  if sources:
-    counts = dict(counts, evidence_drives=len(sources), evidence_windows=sum(map(len, windows.values())))
   if not permitted():
     raise ValueError('Parked authority lost after analysis')
   from openpilot.starpilot.flm.offroad_baseline_windows import stable_proposal_windows, tracking_proposal_windows
@@ -526,19 +465,17 @@ def _assess_inputs(route, collected, previous, *, permitted, sources=None, route
   for key, values in sorted(proposal_windows.items()):
     if not values:
       continue
-    report = analyze(tuple(values), amplitude_prior, permitted=permitted, route_order=route_order, comparison=comparisons.get(key))
+    report = analyze(tuple(values), amplitude_prior, permitted=permitted)
     if key in proposal_grades:
       report['tracking_source_grade'] = proposal_grades[key]
-      _separate_qualification(report, windows.get(key, ()), route_order=route_order, role_windows=values)
+      _separate_qualification(report, windows.get(key, ()))
     selected_ids = {wid for cohort_row in report['cohorts'] for role in ('training', 'heldout') for wid in cohort_row.get(role, {}).get('windows', ())}
     report.update(
       original_context_sha256=key,
       original_metadata=context_metadata[key],
-      recommendation_original_windows=_recommendation_windows(values),
       selected_original_windows=[
         {
           'identity': w.identity,
-          'route': w.points[0].route,
           'segment': w.points[0].segment,
           'start_ns': w.points[0].mono_ns,
           'end_ns': w.points[-1].mono_ns,
@@ -549,10 +486,6 @@ def _assess_inputs(route, collected, previous, *, permitted, sources=None, route
         if w.identity in selected_ids
       ],
     )
-    if key in comparisons:
-      from openpilot.starpilot.flm.offroad_history import bind_original_contexts
-
-      bind_original_contexts(report, [*values, *windows.get(key, ())], comparisons[key])
     reports.append(report)
   reports = tuple(reports)
   stable = []
@@ -564,7 +497,6 @@ def _assess_inputs(route, collected, previous, *, permitted, sources=None, route
       if cohort(child) == target:
         return {
           'identity': child.identity,
-          'route': child.points[0].route,
           'parent_identity': window.identity,
           'segment': child.points[0].segment,
           'start_ns': child.points[0].mono_ns,
@@ -577,20 +509,18 @@ def _assess_inputs(route, collected, previous, *, permitted, sources=None, route
   for key, values in sorted(stable_windows.items()):
     if not values:
       continue
-    report = analyze_stable_windows(tuple(values), permitted=permitted, route_order=route_order, comparison=comparisons.get(key))
+    report = analyze_stable_windows(tuple(values), permitted=permitted)
     grade = stable_grades.get(key, 'continuous_original_control_metadata')
     report['tracking_source_grade'] = grade
     if grade == 'bookended_zero_route_tracking':
-      _separate_qualification(report, windows.get(key, ()), proposal_windows=values, route_order=route_order)
+      _separate_qualification(report, windows.get(key, ()), proposal_windows=values)
     selected = {wid for c in report['cohorts'] for role in ('training', 'heldout') for wid in c.get(role, {}).get('windows', ())}
     report.update(
       original_context_sha256=key,
       original_metadata=context_metadata[key],
-      recommendation_original_windows=_recommendation_windows(values),
       selected_original_windows=[
         {
           'identity': w.identity,
-          'route': w.points[0].route,
           'segment': w.points[0].segment,
           'start_ns': w.points[0].mono_ns,
           'end_ns': w.points[-1].mono_ns,
@@ -603,10 +533,6 @@ def _assess_inputs(route, collected, previous, *, permitted, sources=None, route
         if w.identity in selected
       ],
     )
-    if key in comparisons:
-      from openpilot.starpilot.flm.offroad_history import bind_original_contexts
-
-      bind_original_contexts(report, [*values, *windows.get(key, ())], comparisons[key])
     stable.append(report)
   counts['stable_qualified_whole_windows'] = sum(map(len, stable_windows.values()))
   diagnostic = []
@@ -623,7 +549,6 @@ def _assess_inputs(route, collected, previous, *, permitted, sources=None, route
     )
   counts['diagnostic_whole_windows'] = sum(map(len, raw_windows.values()))
   return CompletedAssessment(
-    receipt, reports, counts, state['cp'], (), route, tuple(quality_stops), tuple(diagnostic), tuple(stable),
-    _recorded_tracking(proposal_windows), sources,
+    receipt, reports, counts, state['cp'], (), route, tuple(quality_stops), tuple(diagnostic), tuple(stable), _recorded_tracking(proposal_windows)
   )
 

openpilot/starpilot/flm/offroad_residual.py

@@ -77,22 +77,15 @@ def _interval(values):
   return max(0.0, min(min(values), mean - radius)), max(max(values), mean + radius)
 
 
-def estimate_residual(windows, *, cohort, permitted=lambda: True, comparison=None):
+def estimate_residual(windows, *, cohort, permitted=lambda: True):
   if (
     type(windows) is not tuple
     or not 3 <= len(windows) <= 12
     or any(type(w) is not RouteWindow or not w.valid() for w in windows)
     or len({w.identity for w in windows}) != len(windows)
-    or comparison is None and len({(w.domain, w.points[0].context_key) for w in windows}) != 1
+    or len({(w.domain, w.points[0].context_key) for w in windows}) != 1
   ):
     raise ValueError('Distinct complete same-context residual windows required')
-  context = windows[0].points[0].context_key
-  if comparison is not None:
-    from openpilot.starpilot.flm.offroad_history import TrackingComparison
-
-    if type(comparison) is not TrackingComparison:
-      raise ValueError('Original numerical comparison contract required')
-    context = comparison.context(windows)
   seen = set()
   metrics = []
   references = []
@@ -123,7 +116,7 @@ def estimate_residual(windows, *, cohort, permitted=lambda: True, comparison=Non
     means.append(sum(p.desired for p in points) / len(points))
   source = hashlib.sha256(repr((cohort, tuple(w.identity for w in windows))).encode()).hexdigest()
   return ResidualEstimate(
-    context,
+    windows[0].points[0].context_key,
     windows[0].domain,
     cohort,
     tuple(w.identity for w in windows),
@@ -186,11 +179,10 @@ def compare_residual(before, after):
   }
 
 
-def analyze_stable_windows(windows, *, permitted=lambda: True, route_order=None, comparison=None):
+def analyze_stable_windows(windows, *, permitted=lambda: True):
   """Fixed input-only cohorts; calm references need no fictitious gain fit."""
   from dataclasses import asdict
-  from openpilot.starpilot.flm.offroad_analysis import cohort
-  from openpilot.starpilot.flm.offroad_history import partition_windows
+  from openpilot.starpilot.flm.offroad_analysis import cohort, _spread, MAX_COHORT_WINDOWS
 
   if type(windows) is not tuple or any(not w.valid() for w in windows):
     raise ValueError('Original Stable windows required')
@@ -203,10 +195,12 @@ def analyze_stable_windows(windows, *, permitted=lambda: True, route_order=None,
   for key, rows in sorted(groups.items()):
     if not permitted():
       raise ValueError('Parked tracking analysis cancelled')
+    chosen = _spread(sorted(rows, key=lambda w: (w.points[0].segment, w.points[0].mono_ns)), MAX_COHORT_WINDOWS)
     # Keep source-contiguous evidence within each role. Interleaving every
     # six-second slice makes the recovered four-second detector blind at each
     # artificial boundary. The two roles still share no original samples.
-    training, heldout = partition_windows(rows, route_order, contiguous=True)
+    split = len(chosen) // 2
+    training, heldout = tuple(chosen[:split]), tuple(chosen[split:])
     if len(training) < 3 or len(heldout) < 3:
       results.append({'cohort': key, 'whole_windows': len(rows), 'reason': 'more_independent_stable_windows_required'})
       continue
@@ -214,8 +208,8 @@ def analyze_stable_windows(windows, *, permitted=lambda: True, route_order=None,
       {
         'cohort': key,
         'whole_windows': len(rows),
-        'training': asdict(estimate_residual(training, cohort=key, permitted=permitted, comparison=comparison)),
-        'heldout': asdict(estimate_residual(heldout, cohort=key, permitted=permitted, comparison=comparison)),
+        'training': asdict(estimate_residual(training, cohort=key, permitted=permitted)),
+        'heldout': asdict(estimate_residual(heldout, cohort=key, permitted=permitted)),
         'reason': 'original_stable_symptoms_require_replication',
         'proposal': None,
       }

openpilot/starpilot/flm/offroad_stable_bridge.py

@@ -24,7 +24,7 @@ from openpilot.starpilot.flm.offroad_route import RETAIN
 from openpilot.starpilot.flm.offroad_evidence import read_point, MAX_GAP_NS, RouteWindow
 from openpilot.starpilot.flm.offroad_processor import CompletedAssessment
 from openpilot.starpilot.flm.log_decode import decode_full_rlog_segment
-from openpilot.starpilot.galaxy.drive_analysis import _closed_source, _Incomplete
+from openpilot.starpilot.galaxy.drive_analysis import _closed_source
 
 
 def independent_profile_support(training, heldout, previous):
@@ -88,7 +88,6 @@ class StableRouteSelector:
         ids = {
           wid for c in group['cohorts'] if all(role in c for role in ('training', 'heldout')) for role in ('training', 'heldout') for wid in c[role]['windows']
         }
-        ids.update(w['identity'] for w in group.get('recommendation_original_windows', ()))
         groups.append((group, ids))
     elif type(assessment) is PriorOriginals and type(assessment.receipt) is SelectedSourceReceipt and assessment.profile_metadata is not None:
       groups = [
@@ -124,15 +123,12 @@ class StableRouteSelector:
     for group, ids in groups:
       if group['original_metadata']['domain'] != 'lateral_acceleration':
         continue
-      for window in (*group['selected_original_windows'], *group.get('recommendation_original_windows', ())):
+      for window in group['selected_original_windows']:
         if window['identity'] in ids:
-          span = {name: window[name] for name in ('identity', 'route', 'segment', 'start_ns', 'end_ns')}
-          if window['identity'] in self.spans and self.spans[window['identity']] != span:
+          if window['identity'] in self.spans and self.spans[window['identity']] != window:
             raise ValueError('Conflicting original window identity')
-          self.spans[window['identity']] = span
-          context = group.get('original_window_contexts', {}).get(window['identity'])
-          original = group['original_metadata'] if context is None else group['original_contexts'][context]
-          comparison = dict(original)
+          self.spans[window['identity']] = window
+          comparison = dict(group['original_metadata'])
           comparison.pop('route_values')
           comparison.pop('route_revision')
           context_key = hashlib.sha256(repr(comparison).encode()).hexdigest()
@@ -143,23 +139,6 @@ class StableRouteSelector:
       raise ValueError('Original whole-window bound')
 
   def _load(self):
-    if getattr(self.assessment, 'evidence_sources', None):
-      from copy import copy
-
-      self.rows = {}
-      for route, source in self.assessment.evidence_sources.items():
-        reader = copy(self)
-        reader.assessment = source
-        reader.spans = {key: span for key, span in self.spans.items() if span['route'] == route}
-        try:
-          reader._load()
-        except (FileNotFoundError, _Incomplete) as error:
-          if not self.permitted() or isinstance(error, _Incomplete) and str(error) == 'Analysis cancelled':
-            raise
-          reader.rows = dict.fromkeys(reader.spans)
-        assert reader.rows is not None
-        self.rows.update(reader.rows)
-      return
     rows = {key: [] for key in self.spans}
     original_points = {key: [] for key in self.spans}
     invalid = set()
@@ -322,18 +301,6 @@ class StableRouteSelector:
         report['conversion_refusal'] = str(error)
     return mapped, report
 
-  def recommendations(self, group, *, stable=False):
-    from openpilot.starpilot.flm.offroad_behavior import behavior_candidates
-
-    if self.rows is None:
-      self._load()
-    groups = {}
-    for window in group.get('recommendation_original_windows', ()):
-      groups.setdefault(window['cohort'], []).append(window['identity'])
-    return [candidate for key, ids in sorted(groups.items()) for candidate in behavior_candidates(
-      tuple(ids), self.rows, self.contexts, self.previous, key, stable=stable, permitted=self.permitted,
-    )]
-
   def propose(self, group, cohort):
     ids = {role: tuple(cohort[role]['windows']) for role in ('training', 'heldout')}
     if (

openpilot/starpilot/flm/offroad_tracking.py

@@ -75,7 +75,7 @@ def _interval(values):
   return min(min(values), mean - radius), max(max(values), mean + radius)
 
 
-def estimate(windows, *, cohort, permitted=lambda: True, comparison=None):
+def estimate(windows, *, cohort, permitted=lambda: True):
   _allowed(permitted)
   if (
     type(windows) is not tuple
@@ -84,16 +84,9 @@ def estimate(windows, *, cohort, permitted=lambda: True, comparison=None):
     or not 0 < len(cohort) <= 96
     or any(type(w) is not RouteWindow or not w.valid() for w in windows)
     or len({w.identity for w in windows}) != len(windows)
-    or comparison is None and len({(w.domain, w.points[0].context_key) for w in windows}) != 1
+    or len({(w.domain, w.points[0].context_key) for w in windows}) != 1
   ):
     raise ValueError('Whole distinct same-context tracking windows required')
-  context = windows[0].points[0].context_key
-  if comparison is not None:
-    from openpilot.starpilot.flm.offroad_history import TrackingComparison
-
-    if type(comparison) is not TrackingComparison:
-      raise ValueError('Original numerical comparison contract required')
-    context = comparison.context(windows)
   seen = set()
   table = {lag: {} for lag in LAGS_NS}
   rms = []
@@ -158,7 +151,7 @@ def estimate(windows, *, cohort, permitted=lambda: True, comparison=None):
   overshoot = math.sqrt(sum(m['overshoot_rms'] ** 2 for m in metrics) / len(metrics)) / reference_rms
   peak = max(m['worst_absolute_error'] for m in metrics) / reference_rms
   return TrackingEstimate(
-    context,
+    windows[0].points[0].context_key,
     windows[0].domain,
     cohort,
     gain,

openpilot/starpilot/flm/offroad_worker.py

@@ -281,8 +281,7 @@ def recorded_recommendations(assessment, previous, report):
 
   found: dict[str, dict[str, Any]] = {}
 
-  def add(proposal, windows, reason, baseline=None):
-    baseline = previous if baseline is None else baseline
+  def add(proposal, windows, reason):
     if type(proposal) is not dict or type(windows) is not int or windows <= 0:
       return
     values = proposal.get('values')
@@ -294,13 +293,11 @@ def recorded_recommendations(assessment, previous, report):
       delta = proposal.get('delta')
       if name not in NAMES or type(delta) not in (int, float) or not math.isfinite(delta):
         return
-      values = {name: baseline[name] + delta}
+      values = {name: previous[name] + delta}
     for name in NAMES:
       base, target = previous[name], values.get(name, previous[name])
       if type(target) not in (int, float) or not math.isfinite(target):
         continue
-      if target == baseline[name]:
-        continue
       low, high = route_bounds(name)
       if not low <= target <= high or not 1e-12 < abs(target - base) <= MAX_ROUTE_STEP + 1e-12:
         continue
@@ -311,14 +308,10 @@ def recorded_recommendations(assessment, previous, report):
           row.update(target=None, reason='conflicting_recorded_candidates')
       else:
         found[name] = {'parameter': name, 'baseline': base, 'target': target, 'whole_windows': windows, 'reason': reason}
-      if proposal.get('evidence') and found[name]['target'] is not None:
-        found[name]['evidence'] = proposal['evidence']
 
   for group in (*assessment.groups, *assessment.diagnostic_groups):
     for cohort in group['cohorts']:
-      baseline = _recorded_values(group['original_metadata']) if 'original_metadata' in group else previous
-      if baseline is not None:
-        add(cohort.get('proposal'), cohort.get('whole_windows', 0), 'recorded_candidate_requires_automatic_checks', baseline)
+      add(cohort.get('proposal'), cohort.get('whole_windows', 0), 'recorded_candidate_requires_automatic_checks')
   for proof in (*report.get('stable_profile_reports', ()), *report.get('short_turn_profile_reports', ())):
     roles = proof.get('roles', {})
     windows = sum(len(roles.get(role, {}).get('original_windows', ())) for role in ('training', 'heldout'))
@@ -326,8 +319,6 @@ def recorded_recommendations(assessment, previous, report):
     add(proof.get('observed_candidate'), windows, 'recorded_candidate_not_independently_confirmed')
     for proposal in roles.get('training', {}).get('coordinate_candidates', ()):
       add(proposal, windows, 'training_candidate_not_independently_confirmed')
-  for proposal in report.get('behavior_recommendations', ()):
-    add(proposal, proposal['whole_windows'], 'repeated_observed_behavior_requires_automatic_checks')
   return [found[name] for name in NAMES if name in found]
 
 
@@ -364,7 +355,7 @@ def input_coverage(cohort):
 def _input_spans(group, cohort, ids):
   name = 'qualification_original_windows' if 'qualification_roles' in cohort else 'selected_original_windows'
   return {
-    (row.get('route', ''), row['segment'], row['start_ns'], row['end_ns'])
+    (row['segment'], row['start_ns'], row['end_ns'])
     for row in group.get(name, ())
     if row['identity'] in ids and all(key in row for key in ('segment', 'start_ns', 'end_ns'))
   }
@@ -399,22 +390,6 @@ class OffroadWorker:
     self.on_recommendations = on_recommendations
     self.prior_provider = prior_provider
     self.stable_prior_provider = stable_prior_provider
-    self.evidence_bank = None
-    if assessor is analyze_latest:
-      from openpilot.starpilot.flm.offroad_history import EvidenceBank
-
-      self.evidence_bank = EvidenceBank(self.cache.directory)
-
-  def _history_identity(self, cp):
-    settings = []
-    for key in SETTINGS:
-      if key == KEY:
-        continue
-      raw, okay = read_saved(self.params, key, MAX_BYTES)
-      if not okay:
-        raise ValueError('Unreadable evidence history setting')
-      settings.append((key, None if raw is None else hashlib.sha256(raw).hexdigest()))
-    return _digest((self.scope_for(cp, self.params).document(), settings))
 
   def _snapshot(self, route):
     values = []
@@ -484,10 +459,10 @@ class OffroadWorker:
 
     if self.on_analysis is not None:
       self.on_analysis()
-    options: dict[str, Any] = {} if self.evidence_bank is None else {'evidence_bank': self.evidence_bank, 'history_identity': self._history_identity}
-    if route != latest:
-      options['selected_route'] = route
-    assessment = self.assessor(self.params, self.root, prior, permitted=self.permitted, **options)
+    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:
@@ -522,96 +497,6 @@ class OffroadWorker:
     report = assessment.report()
     report.update(policy=POLICY, current_scope=scope.document(), road_improvement_confirmed=False)
     current = dict(ZERO) if row is None else dict(row['values'])
-    recorded['recommendations'] = recorded_recommendations(assessment, current, report)
-
-    choices = []
-    for group in assessment.groups:
-      metadata = group['original_metadata']
-      if _recorded_values(metadata) != current or metadata['controller_mode'] != scope.controller_mode:
-        continue
-      contract = recorded_source_contract(assessment.cp, metadata, scope, self.params)
-      if contract is None:
-        continue
-      report.setdefault('original_source_requalification', []).append(contract)
-      for cohort in group['cohorts']:
-        if 'qualification_roles' in cohort and any(len(cohort['qualification_roles'][role]['windows']) < 3 for role in ('training', 'heldout')):
-          continue
-        proposal = cohort.get('proposal')
-        if proposal is not None:
-          proposal = dict(proposal, values=dict(current, **proposal['values']))
-          choices.append((group, cohort, proposal))
-    # Freeze all proposal sources before any controller/native fit. One
-    # marginal tracking proposal must not hide the Recommended profile.
-    short_groups = tuple(
-      group for group in assessment.groups
-      if any(
-        c['cohort'].rsplit(':', 1)[-1] in ('steady', 'turn_in', 'unwind')
-        and all(3 <= len(c.get(role, {}).get('windows', ())) <= 12 for role in ('training', 'heldout'))
-        for c in group['cohorts']
-      )
-    ) if type(assessment) is CompletedAssessment else ()
-    if (
-      type(assessment) is CompletedAssessment and (
-        assessment.stable_groups or short_groups or any(g.get('recommendation_original_windows') for g in assessment.groups)
-      )
-      and self.checker is qualify and scope.controller_domain == 'torque'
-      and (row is None or row['state'] != 'ready_for_next_drive')
-    ):
-      from openpilot.starpilot.flm.offroad_stable_bridge import StableRouteSelector
-
-      selector = StableRouteSelector(assessment, self.root, current, scope, permitted=self.permitted)
-      report['behavior_recommendations'] = []
-      for stable, groups in ((False, assessment.groups), (True, assessment.stable_groups)):
-        for group in groups:
-          metadata = group['original_metadata']
-          if (group.get('recommendation_original_windows') and _recorded_values(metadata) == current and metadata['controller_mode'] == scope.controller_mode
-              and recorded_source_contract(assessment.cp, metadata, scope, self.params) is not None):
-            report['behavior_recommendations'].extend(selector.recommendations(group, stable=stable))
-      report['stable_profile_reports'] = []
-      for group in assessment.stable_groups:
-        metadata = group['original_metadata']
-        if _recorded_values(metadata) != current or metadata['controller_mode'] != scope.controller_mode:
-          continue
-        contract = recorded_source_contract(assessment.cp, metadata, scope, self.params)
-        if contract is None:
-          continue
-        report.setdefault('original_source_requalification', []).append(contract)
-        for cohort in group['cohorts']:
-          if not all(role in cohort for role in ('training', 'heldout')):
-            continue
-          if 'qualification_roles' in cohort and any(len(cohort['qualification_roles'][role]['windows']) < 3 for role in ('training', 'heldout')):
-            continue
-          proposal, proof = selector.propose(group, cohort)
-          report['stable_profile_reports'].append({'context': group['original_context_sha256'], 'cohort': cohort['cohort'], **proof})
-          if proposal is not None:
-            choices.append((group, {**cohort, 'stable_profile_proof': proof}, proposal))
-          for alternative in proof.get('coordinate_candidates', []):
-            if proposal is None or alternative['values'] != proposal['values']:
-              choices.append((group, {**cohort, 'stable_profile_proof': proof}, alternative))
-      # Turn entry/exit need the existing short controller windows, not a
-      # six-second steady interval. Keep their original partitions and all
-      # native checks; legacy quieting profiles still use stable windows.
-      report['short_turn_profile_reports'] = []
-      for group in short_groups:
-        metadata = group['original_metadata']
-        if _recorded_values(metadata) != current or metadata['controller_mode'] != scope.controller_mode:
-          continue
-        contract = recorded_source_contract(assessment.cp, metadata, scope, self.params)
-        if contract is None:
-          continue
-        for cohort in group['cohorts']:
-          if cohort['cohort'].rsplit(':', 1)[-1] not in ('steady', 'turn_in', 'unwind'):
-            continue
-          if not all(role in cohort for role in ('training', 'heldout')):
-            continue
-          roles = cohort.get('qualification_roles', cohort)
-          if any(not 3 <= len(roles[role]['windows']) <= 12 for role in ('training', 'heldout')):
-            continue
-          _, proof = selector.propose(group, cohort)
-          report['short_turn_profile_reports'].append({'context': group['original_context_sha256'], 'cohort': cohort['cohort'], **proof})
-          for alternative in proof.get('coordinate_candidates', []):
-            choices.append((group, {**cohort, 'stable_profile_proof': proof}, alternative))
-
 
     def recommendations_ready():
       recorded['recommendations'] = recorded_recommendations(assessment, current, report)
@@ -625,22 +510,20 @@ class OffroadWorker:
     offered = None
     profile_suggestions = []
     selected_profile_windows = 0
-    actual_drive = getattr(assessment, 'single_drive', None)
-    actual_drive = actual_drive if type(actual_drive) is CompletedAssessment else assessment
     if row is not None and assessment.receipt.source_sha256 in row['route_sha256s']:
       reason = 'this_completed_drive_already_assessed'
-    elif row is not None and (row['state'] == 'ready_for_next_drive' or quality_withdrawn(actual_drive, scope, row['values'])):
+    elif row is not None and (row['state'] == 'ready_for_next_drive' or quality_withdrawn(assessment, scope, row['values'])):
       trial = trial_document(self.cache, row['analysis_sha256'])
       observed_checks = []
-      if not quality_withdrawn(actual_drive, scope, row['values']):
+      if not quality_withdrawn(assessment, scope, row['values']):
         stable = trial['before']['training'].get('kind') == 'uncommanded_residual_v1'
-        for group in actual_drive.stable_groups if stable else actual_drive.groups:
+        for group in assessment.stable_groups if stable else assessment.groups:
           if group['original_metadata']['route_scope'] != scope.key() or _recorded_values(group['original_metadata']) != row['values']:
             continue
           for cohort in group['cohorts']:
             if cohort['cohort'] == trial['cohort'] and 'training' in cohort and 'heldout' in cohort:
-              actual = self.checker(actual_drive, group, cohort, row['values'], self.root, permitted=self.permitted, require_change=False)
-              contribution = self.checker(actual_drive, group, cohort, row['previous'], self.root, permitted=self.permitted, require_change=True)
+              actual = self.checker(assessment, group, cohort, row['values'], self.root, permitted=self.permitted, require_change=False)
+              contribution = self.checker(assessment, group, cohort, row['previous'], self.root, permitted=self.permitted, require_change=True)
               observed_checks.append(
                 {'qualified': actual['qualified'] and contribution['qualified'], 'actual_recorded_map': actual, 'compared_with_prior_map': contribution}
               )
@@ -648,7 +531,7 @@ 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, actual_drive, scope, trial_started=started, recorded_backlog=route != latest)
+      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
@@ -691,6 +574,84 @@ class OffroadWorker:
         status = 'ready_for_next_drive' if outcome['outcome'] == 'pending' else outcome['outcome']
         reason = outcome['reason']
     else:
+      current = dict(ZERO) if row is None else dict(row['values'])
+      choices = []
+      for group in assessment.groups:
+        metadata = group['original_metadata']
+        if _recorded_values(metadata) != current or metadata['controller_mode'] != scope.controller_mode:
+          continue
+        contract = recorded_source_contract(assessment.cp, metadata, scope, self.params)
+        if contract is None:
+          continue
+        report.setdefault('original_source_requalification', []).append(contract)
+        for cohort in group['cohorts']:
+          if 'qualification_roles' in cohort and any(len(cohort['qualification_roles'][role]['windows']) < 3 for role in ('training', 'heldout')):
+            continue
+          proposal = cohort.get('proposal')
+          if proposal is not None:
+            proposal = dict(proposal, values=dict(current, **proposal['values']))
+            choices.append((group, cohort, proposal))
+      # Freeze all proposal sources before any controller/native fit. One
+      # marginal tracking proposal must not hide the Recommended profile.
+      short_groups = tuple(
+        group for group in assessment.groups
+        if any(
+          c['cohort'].rsplit(':', 1)[-1] in ('steady', 'turn_in', 'unwind')
+          and all(3 <= len(c.get(role, {}).get('windows', ())) <= 12 for role in ('training', 'heldout'))
+          for c in group['cohorts']
+        )
+      ) if type(assessment) is CompletedAssessment else ()
+      if (
+        type(assessment) is CompletedAssessment and (assessment.stable_groups or short_groups)
+        and self.checker is qualify and scope.controller_domain == 'torque'
+      ):
+        from openpilot.starpilot.flm.offroad_stable_bridge import StableRouteSelector
+
+        selector = StableRouteSelector(assessment, self.root, current, scope, permitted=self.permitted)
+        report['stable_profile_reports'] = []
+        for group in assessment.stable_groups:
+          metadata = group['original_metadata']
+          if _recorded_values(metadata) != current or metadata['controller_mode'] != scope.controller_mode:
+            continue
+          contract = recorded_source_contract(assessment.cp, metadata, scope, self.params)
+          if contract is None:
+            continue
+          report.setdefault('original_source_requalification', []).append(contract)
+          for cohort in group['cohorts']:
+            if not all(role in cohort for role in ('training', 'heldout')):
+              continue
+            if 'qualification_roles' in cohort and any(len(cohort['qualification_roles'][role]['windows']) < 3 for role in ('training', 'heldout')):
+              continue
+            proposal, proof = selector.propose(group, cohort)
+            report['stable_profile_reports'].append({'context': group['original_context_sha256'], 'cohort': cohort['cohort'], **proof})
+            if proposal is not None:
+              choices.append((group, {**cohort, 'stable_profile_proof': proof}, proposal))
+            for alternative in proof.get('coordinate_candidates', []):
+              if proposal is None or alternative['values'] != proposal['values']:
+                choices.append((group, {**cohort, 'stable_profile_proof': proof}, alternative))
+        # Turn entry/exit need the existing short controller windows, not a
+        # six-second steady interval. Keep their original partitions and all
+        # native checks; legacy quieting profiles still use stable windows.
+        report['short_turn_profile_reports'] = []
+        for group in short_groups:
+          metadata = group['original_metadata']
+          if _recorded_values(metadata) != current or metadata['controller_mode'] != scope.controller_mode:
+            continue
+          contract = recorded_source_contract(assessment.cp, metadata, scope, self.params)
+          if contract is None:
+            continue
+          for cohort in group['cohorts']:
+            if cohort['cohort'].rsplit(':', 1)[-1] not in ('steady', 'turn_in', 'unwind'):
+              continue
+            if not all(role in cohort for role in ('training', 'heldout')):
+              continue
+            roles = cohort.get('qualification_roles', cohort)
+            if any(not 3 <= len(roles[role]['windows']) <= 12 for role in ('training', 'heldout')):
+              continue
+            _, proof = selector.propose(group, cohort)
+            report['short_turn_profile_reports'].append({'context': group['original_context_sha256'], 'cohort': cohort['cohort'], **proof})
+            for alternative in proof.get('coordinate_candidates', []):
+              choices.append((group, {**cohort, 'stable_profile_proof': proof}, alternative))
       # Freeze input-only ranking; training may select a candidate, validation may not.
       choices.sort(key=lambda v: (-input_coverage(v[1]), -v[1]['whole_windows'], v[1]['cohort'], v[0]['original_context_sha256']))
       recommendations_ready()
@@ -726,7 +687,7 @@ class OffroadWorker:
               validation = tuple(w['identity'] for w in independent.windows)
             validation_spans = _input_spans(group, cohort, validation) if independent is None else ()
             if training_seen.intersection(validation) or any(
-              a[:2] == b[:2] and max(a[2], b[2]) <= min(a[3], b[3]) for a in training_spans for b in validation_spans
+              a[0] == b[0] and max(a[1], b[1]) <= min(a[2], b[2]) for a in training_spans for b in validation_spans
             ):
               reason = 'candidate_validation_already_used_for_training'
               continue
@@ -804,35 +765,20 @@ class OffroadWorker:
               if TrialGuard(scope.key(), proposal['values'], record).stopped is not None:
                 raise ValueError('Original onroad gross-guard baseline unavailable')
               digest = self.cache.put(record)
-              new: dict[str, Any] = {
+              new = {
                 'scope': scope.document(),
                 'values': proposal['values'],
                 'previous': current,
-                'route_sha256s': sorted(
-                  ({w['source_sha256'] for w in checked['windows']} if assessment.evidence_sources else {assessment.receipt.source_sha256})
-                  | (set() if independent_checks is None else {independent_checks['source_partitions']['heldout']})
-                ),
+                'route_sha256s': [assessment.receipt.source_sha256]
+                + ([] if independent_checks is None else [independent_checks['source_partitions']['heldout']]),
                 'source_start_ns': assessment.receipt.original_start_ns,
                 'source_end_ns': assessment.receipt.original_end_ns,
-                'completed_wall_ns': max(
-                  [assessment.receipt.completed_wall_ns] + [source.receipt.completed_wall_ns for source in assessment.evidence_sources.values()]
-                ),
+                'completed_wall_ns': assessment.receipt.completed_wall_ns,
                 'analysis_sha256': digest,
                 'sequence': 1 if row is None else row['sequence'] + 1,
                 'state': 'ready_for_next_drive',
               }
-              def original_sources_current(new=new):
-                if not authorized():
-                  return False
-                from openpilot.starpilot.galaxy.drive_history import DriveHistory
-
-                required = [s.source_inventory for s in assessment.evidence_sources.values() if s.receipt.source_sha256 in new['route_sha256s']]
-                if not required:
-                  return True
-                inventory = DriveHistory(self.root, max_segments_per_route=MAX_SEGMENTS).snapshot()
-                return not inventory['scanIncomplete'] and all(sum(r == source for r in inventory['routes']) == 1 for source in required)
-
-              saved = RouteStorage(self.params).save(new, expected=expected, authorized=original_sources_current)
+              saved = RouteStorage(self.params).save(new, expected=expected, authorized=authorized)
               if not saved:
                 raise ValueError('Next-drive tune compare-and-swap refused')
               status = 'ready_for_next_drive'

openpilot/starpilot/flm/tests/test_offroad_backend.py

@@ -1316,7 +1316,7 @@ class TestOffroadTrackingPolicy(unittest.TestCase):
     assessment = CompletedAssessment(receipt, (group,), {}, cp, (), inventory)
     calls = []
 
-    def assessed(params, root, previous, *, permitted, evidence_bank=None, history_identity=None):
+    def assessed(params, root, previous, *, permitted):
       previous(cp)
       return assessment
 
@@ -1778,8 +1778,7 @@ class TestOffroadTrackingPolicy(unittest.TestCase):
 
     def window(segment, identity):
       points = tuple(
-        SimpleNamespace(route=prior_id, segment=segment, mono_ns=segment * 10_000_000_000 + i * 10_000_000,
-                        context_key=comparison_key, desired=0.2, speed_mps=25.0)
+        SimpleNamespace(segment=segment, mono_ns=segment * 10_000_000_000 + i * 10_000_000, context_key=comparison_key, desired=0.2, speed_mps=25.0)
         for i in range(8)
       )
       return SimpleNamespace(identity=identity, points=points, domain='lateral_acceleration')
@@ -1797,7 +1796,7 @@ class TestOffroadTrackingPolicy(unittest.TestCase):
       stable_windows={metadata_key: profile, other_key: wrong_map},
     )
     route = {'routeId': prior_id, 'segments': [{'files': {'rlog': True}} for _ in range(8)]}
-    assessment = SimpleNamespace(receipt=SimpleNamespace(route_id=current_id, source_sha256='c' * 64, cp_sha256='a' * 64, completed_wall_ns=10))
+    assessment = SimpleNamespace(receipt=SimpleNamespace(route_id=current_id, cp_sha256='a' * 64, completed_wall_ns=10))
     with (
       patch.object(module.DriveHistory, 'snapshot', return_value={'scanIncomplete': False, 'routes': [route]}),
       patch.object(module, 'collect_route_inputs', return_value=collected),

openpilot/starpilot/flm/tests/test_offroad_behavior.py

@@ -1,122 +0,0 @@
-"""Declared original samples test observation counts, never physical improvement."""
-
-from dataclasses import replace
-from pathlib import Path
-import tempfile
-import unittest
-from unittest.mock import patch
-
-from openpilot.starpilot.flm.offroad_behavior import behavior_candidates
-from openpilot.starpilot.flm.offroad_stable_policy import FLMSample
-from openpilot.starpilot.flm.tests.test_offroad_backend import values
-
-
-class TestObservedBehavior(unittest.TestCase):
-  @staticmethod
-  def samples(number, *, bad=True, start=0, route=None):
-    result = []
-    for index in range(201):
-      t = index / 100
-      desired = -1.5 + .3 * t
-      actual = desired - (.25 if bad else 0)
-      result.append(FLMSample(route or f'declared-drive-{number}', 0, start + t + 1, 25., True, False, False,
-                              actual, desired, .3, actual - desired, 0., 0., 0., 0., 0., .1, -20., 0., .1, .1, 0.))
-    return result
-
-  def recommend(self, rows, **kwargs):
-    return behavior_candidates(tuple(rows), rows, dict.fromkeys(rows, 'same-source-context'), values(), '2:right:unwind', **kwargs)
-
-  def test_four_of_five_sections_report_right_unwind_without_claiming_five_turns(self):
-    rows = {str(i): self.samples(i, bad=i != 4) for i in range(5)}
-    candidates = self.recommend(rows)
-    self.assertEqual({r['parameter'] for r in candidates}, {
-      'stable:unwind_threshold_increase_right', 'stable:unwind_taper_right',
-      'stable:curvy_unwind_floor_relief_right', 'stable:curvy_unwind_extra_reduction_right',
-    })
-    for row in candidates:
-      self.assertIn('4 of 5 retained higher-speed right turn-exit sections across 5 drive(s)', row['evidence'])
-      self.assertIn('steering unwound too slowly', row['evidence'])
-      self.assertIn('not a count of distinct turns', row['evidence'])
-      self.assertEqual((row['measured_sections'], row['matched_sections'], row['evidence_drives']), (5, 4, 5))
-      self.assertGreater(row['delta'], 0)
-      self.assertLessEqual(row['delta'], .01)
-
-  def test_adjacent_windows_are_one_section_not_five_votes(self):
-    rows = {str(i): self.samples(i, start=i * 2.01, route='same-route') for i in range(5)}
-    self.assertEqual(self.recommend(rows), [])
-
-  def test_missing_section_cannot_be_removed_from_the_denominator(self):
-    rows = {str(i): self.samples(i) for i in range(5)}
-    rows['4'] = None
-    self.assertEqual(self.recommend(rows), [])
-
-  def test_opposing_behavior_cannot_offer_one_direction(self):
-    rows = {str(i): self.samples(i) for i in range(5)}
-    rows['4'] = [replace(s, actual_la=s.desired_la + .25, error=.25) for s in rows['4']]
-    self.assertEqual(self.recommend(rows), [])
-
-  def test_sparse_or_cancelled_evidence_cannot_offer_a_change(self):
-    rows = {str(i): self.samples(i) for i in range(2)}
-    self.assertEqual(self.recommend(rows), [])
-    with self.assertRaisesRegex(ValueError, 'cancelled'):
-      self.recommend(rows, permitted=lambda: False)
-
-  def test_overlapping_sections_are_refused(self):
-    rows = {str(i): self.samples(i, route='same-route') for i in range(3)}
-    with self.assertRaisesRegex(ValueError, 'repeat or overlap'):
-      self.recommend(rows)
-
-  def test_recommendation_survives_too_few_automatic_windows(self):
-    from openpilot.starpilot.flm.offroad_processor import _assess_inputs
-    from openpilot.starpilot.flm.tests.test_offroad_coordinate_policy import TestCrossDriveEvidence
-
-    collected, inventory = TestCrossDriveEvidence.inputs(1)
-    assessment = _assess_inputs(inventory, collected, values(), permitted=lambda: True)
-    group = assessment.groups[0]
-    self.assertEqual(len(group['recommendation_original_windows']), 2)
-    self.assertEqual(group['selected_original_windows'], [])
-    self.assertNotIn('training', group['cohorts'][0])
-
-  def test_worker_publishes_behavior_without_native_trial_or_settings_write(self):
-    from openpilot.common.params import Params
-    from openpilot.starpilot.flm import offroad_worker as worker_module
-    from openpilot.starpilot.flm.offroad_processor import _assess_inputs
-    from openpilot.starpilot.flm.offroad_store import KEY
-    from openpilot.starpilot.flm.tests.test_offroad_backend import scope
-    from openpilot.starpilot.flm.tests.test_offroad_coordinate_policy import TestCrossDriveEvidence
-
-    collected, inventory = TestCrossDriveEvidence.inputs(1)
-    assessment = _assess_inputs(inventory, collected, values(), permitted=lambda: True)
-    proposed = self.recommend({str(i): self.samples(i, bad=i != 4) for i in range(5)})
-    class Selector:
-      def __init__(self, *args, **kwargs):
-        pass
-
-      def recommendations(self, group, *, stable):
-        return proposed
-
-      def propose(self, *args):
-        raise AssertionError('Observation-only evidence must not create automatic roles')
-
-    def refuse(*args, **kwargs):
-      raise AssertionError('Observation-only recommendations must not enter automatic qualification')
-
-    with tempfile.TemporaryDirectory() as directory:
-      params = Params(directory)
-      with (
-        patch.object(worker_module, 'latest_route', return_value=inventory),
-        patch.object(worker_module, 'pending_routes', return_value=[inventory]),
-        patch.object(worker_module, 'qualify', new=refuse),
-        patch('openpilot.starpilot.flm.offroad_stable_bridge.StableRouteSelector', new=Selector),
-      ):
-        worker = worker_module.OffroadWorker(params, directory, Path(directory) / 'cache', permitted=lambda: True,
-          assessor=lambda *args, **kwargs: assessment, checker=refuse, scope_for=lambda *args: scope())
-        result = worker.run_once()
-      self.assertFalse(result['configuration_changed'])
-      self.assertIsNone(params.get(KEY))
-      self.assertEqual(len(result['recommendations']), 4)
-      self.assertTrue(all('4 of 5' in row['evidence'] for row in result['recommendations']))
-
-
-if __name__ == '__main__':
-  unittest.main()

openpilot/starpilot/flm/tests/test_offroad_coordinate_policy.py

@@ -16,211 +16,6 @@ from openpilot.starpilot.flm.torque_stages import StageFrame
 from openpilot.starpilot.saved_source import read_saved
 
 
-class TestCrossDriveEvidence(unittest.TestCase):
-  """Declared unit observations, never a physical-route or native-output claim."""
-
-  @staticmethod
-  def inputs(number, *, gain=0.85, route_values=None, anchor=()):
-    from dataclasses import replace
-    import hashlib
-    import math
-    from types import SimpleNamespace
-    from openpilot.starpilot.flm.offroad_evidence import RoutePoint, RouteWindow, window_identity
-    from openpilot.starpilot.flm.offroad_processor import RecordedInputs
-    from openpilot.starpilot.flm.offroad_route import RouteReceipt
-
-    route = f'{number:08x}--aaaaaaaaaa'
-    metadata = {
-      'cp_sha256': 'a' * 64, 'controller_mode': 'starpilot', 'controller_revision': 'declared-unit-controller',
-      'baseline_revision': 'b' * 64, 'domain': 'lateral_acceleration', 'baseline_values': (3.0, 0.1),
-      'vehicle_model_anchor': anchor, 'route_scope': scope().key(), 'route_revision': REVISION,
-      'route_values': tuple((values() if route_values is None else route_values).items()),
-    }
-    context = hashlib.sha256(repr(metadata).encode()).hexdigest()
-    comparison = {k: v for k, v in metadata.items() if k not in ('route_values', 'route_revision')}
-    comparison_key = hashlib.sha256(repr(comparison).encode()).hexdigest()
-    windows = []
-    for segment in (0, 1):
-      points = []
-      for index in range(301):
-        stamp = 1_000_000_000 + segment * 4_000_000_000 + index * 10_000_000
-        desired = 0.6 + 0.1 * math.sin(2 * math.pi * index / 100)
-        point = RoutePoint(route, segment, stamp, 'lateral_acceleration', comparison_key, desired, desired, 20.0,
-                           0.1, 0.1, stamp, stamp, True, False, False, True)
-        points.append(replace(point, actual=gain * desired))
-      points = tuple(points)
-      windows.append(RouteWindow(window_identity(points), points))
-    cp = SimpleNamespace(carFingerprint='declared-car', lateralTuning=SimpleNamespace(which=lambda: 'torque'))
-    receipt = RouteReceipt(route, f'{number:064x}', ((route + '--0', 'd' * 64, 1), (route + '--1', 'e' * 64, 1)),
-                           'a' * 64, 1_000_000_000, 8_000_000_000, number * 10_000_000_000, ())
-    inventory = {'routeId': route, 'segments': [{'files': {'rlog': True}}, {'files': {'rlog': True}}]}
-    collected = RecordedInputs(receipt, {'cp': cp}, {context: windows}, {}, {}, {}, {}, {context: metadata}, {}, [])
-    return collected, inventory
-
-  def test_four_sparse_drives_persist_and_support_one_independent_turn_recommendation(self):
-    from openpilot.starpilot.flm.offroad_history import EvidenceBank, combine
-    from openpilot.starpilot.flm.offroad_processor import _assess_inputs
-
-    with tempfile.TemporaryDirectory() as directory:
-      for number in range(1, 5):
-        bank = EvidenceBank(Path(directory) / 'cache')
-        current, inventory = self.inputs(number)
-        single = _assess_inputs(inventory, current, values(), permitted=lambda: True)
-        self.assertFalse(any(c.get('proposal') for g in single.groups for c in g['cohorts']))
-        history = bank.retain(current, inventory, 'same-settings', permitted=lambda: True)
-      pooled, sources, order = combine(current, inventory, history)
-      assessment = _assess_inputs(inventory, pooled, values(), permitted=lambda: True, sources=sources, route_order=order)
-      cohort = assessment.groups[0]['cohorts'][0]
-      self.assertEqual(cohort['proposal']['parameter'], 'ff_gain_left')
-      self.assertGreater(cohort['proposal']['delta'], 0)
-      owners = {w.identity: w.points[0].route for rows in pooled.windows.values() for w in rows}
-      train, heldout = ({owners[key] for key in cohort[role]['windows']} for role in ('training', 'heldout'))
-      self.assertEqual((len(train), len(heldout)), (2, 2))
-      self.assertTrue(train.isdisjoint(heldout))
-      self.assertEqual(assessment.counts['evidence_drives'], 4)
-      self.assertEqual(len(bank.retain(current, inventory, 'same-settings', permitted=lambda: True)), 4)
-      self.assertLess((bank.cache.directory / 'history.z').stat().st_size, 1024 * 1024)
-      self.assertEqual(bank.read('changed-manual-settings'), [])
-
-  def test_cancel_tamper_and_vehicle_change_cannot_create_or_mix_evidence(self):
-    import zlib
-    from dataclasses import replace
-    from openpilot.starpilot.flm.offroad_history import EvidenceBank, combine
-
-    with tempfile.TemporaryDirectory() as directory:
-      bank = EvidenceBank(Path(directory) / 'cache')
-      original, inventory = self.inputs(1)
-      history = bank.retain(original, inventory, 'same', permitted=lambda: True)
-      before = (bank.cache.directory / 'history.z').read_bytes()
-      other, other_inventory = self.inputs(2)
-      with self.assertRaises(ValueError):
-        bank.retain(other, other_inventory, 'same', permitted=lambda: False)
-      self.assertEqual((bank.cache.directory / 'history.z').read_bytes(), before)
-      different_vehicle = replace(other.receipt, cp_sha256='f' * 64)
-      other.receipt, retained_receipt = different_vehicle, other.receipt
-      _, sources, _ = combine(other, other_inventory, history)
-      self.assertEqual(set(sources), {other.receipt.route_id})
-      history[0]['windows'][0]['points'][0][6] += 0.1
-      other.receipt = retained_receipt
-      with self.assertRaisesRegex(ValueError, 'window changed'):
-        combine(other, other_inventory, history)
-      (bank.cache.directory / 'history.z').write_bytes(zlib.compress(b'{"version":9}'))
-      with self.assertRaises(ValueError):
-        bank.read('same')
-
-  def test_observed_vehicle_model_variation_does_not_erase_repeatable_tracking_or_rewrite_points(self):
-    from openpilot.starpilot.flm.offroad_history import EvidenceBank, combine
-    from openpilot.starpilot.flm.offroad_processor import _assess_inputs
-
-    with tempfile.TemporaryDirectory() as directory:
-      bank = EvidenceBank(Path(directory) / 'cache')
-      original_ids = set()
-      for number in range(1, 5):
-        current, inventory = self.inputs(number, anchor=(14.0 + number * 0.1, 180000.0 + number, 190000.0 + number))
-        original_ids.update(w.identity for rows in current.windows.values() for w in rows)
-        history = bank.retain(current, inventory, 'same', permitted=lambda: True)
-      pooled, sources, order = combine(current, inventory, history)
-      assessment = _assess_inputs(inventory, pooled, values(), permitted=lambda: True, sources=sources, route_order=order)
-      self.assertEqual(len(assessment.groups), 1)
-      group = assessment.groups[0]
-      self.assertEqual(group['cohorts'][0]['proposal']['parameter'], 'ff_gain_left')
-      self.assertEqual(set(group['original_window_contexts']), original_ids)
-      self.assertEqual(len(group['original_contexts']), 4)
-      self.assertEqual({w.identity for rows in pooled.windows.values() for w in rows}, original_ids)
-
-  def test_native_dispatch_keeps_each_source_and_does_not_lower_global_role_requirements(self):
-    from types import SimpleNamespace
-    from openpilot.starpilot.flm import offroad_checks as module
-
-    sources = {str(i): SimpleNamespace(receipt=SimpleNamespace(source_sha256=f'{i:064x}')) for i in range(4)}
-    spans = [{'identity': f'{i:064x}', 'route': str(i // 2)} for i in range(8)]
-    group = {'selected_original_windows': spans}
-    roles = {'training': {'windows': tuple(w['identity'] for w in spans[:4])}, 'heldout': {'windows': tuple(w['identity'] for w in spans[4:])}}
-    assessment = SimpleNamespace(evidence_sources=sources)
-    original = module._original_results
-    seen = []
-
-    def dispatch(actual, descriptors, ids, candidate, root, *, permitted, source_batch=False):
-      if actual is assessment:
-        return original(actual, descriptors, ids, candidate, root, permitted=permitted)
-      self.assertTrue(source_batch)
-      self.assertEqual(len(ids), 2)
-      seen.append(actual)
-      return [{'window': key, 'qualified': True, 'numerical': {'qualified': True, 'changed_frames': 1},
-               'registered_encoder': {'qualified': True, 'candidate_changed_emissions': 1}} for key in ids]
-
-    with patch.object(module, '_original_results', new=dispatch):
-      checked = module.qualify(assessment, group, roles, values(), '.', permitted=lambda: True)
-      self.assertTrue(checked['qualified'])
-      self.assertEqual(seen, list(sources.values()))
-      self.assertEqual({w['source_sha256'] for w in checked['windows']}, {s.receipt.source_sha256 for s in sources.values()})
-      with self.assertRaises(ValueError):
-        module.qualify(assessment, group, dict(roles, training={'windows': roles['training']['windows'][:2]}), values(), '.', permitted=lambda: True)
-
-  def test_old_baseline_suggestions_never_recommend_resetting_an_unrelated_saved_adjustment(self):
-    from types import SimpleNamespace
-    from openpilot.starpilot.flm.offroad_worker import recorded_recommendations
-
-    old = values()
-    proposed = dict(old, ff_gain_left=0.01)
-    current = dict(old, highway_center_taper_max=0.001)
-    group = {'original_metadata': {'route_values': tuple(old.items()), 'route_revision': REVISION},
-             'cohorts': [{'proposal': {'values': proposed}, 'whole_windows': 8}]}
-    assessment = SimpleNamespace(groups=(group,), diagnostic_groups=())
-    recommendations = recorded_recommendations(assessment, current, {})
-    self.assertEqual([r['parameter'] for r in recommendations], ['ff_gain_left'])
-    self.assertEqual(recommendations[0]['target'], 0.01)
-
-  def test_different_recorded_tunes_keep_separate_evidence_partitions(self):
-    from openpilot.starpilot.flm.offroad_history import EvidenceBank, combine
-    from openpilot.starpilot.flm.offroad_processor import _assess_inputs
-
-    with tempfile.TemporaryDirectory() as directory:
-      bank = EvidenceBank(Path(directory) / 'cache')
-      for number in range(1, 5):
-        current, inventory = self.inputs(number, route_values=values(0.001) if number == 4 else values())
-        history = bank.retain(current, inventory, 'same-vehicle-settings', permitted=lambda: True)
-      pooled, sources, order = combine(current, inventory, history)
-      assessment = _assess_inputs(inventory, pooled, values(), permitted=lambda: True, sources=sources, route_order=order)
-      self.assertEqual(len(assessment.groups), 2)
-      self.assertFalse(any(c.get('proposal') for g in assessment.groups for c in g['cohorts']))
-
-  def test_production_worker_collects_across_restarts_saves_then_keeps_next_drive_measurement_separate(self):
-    from openpilot.starpilot.flm import offroad_worker as module
-    from openpilot.starpilot.flm.offroad_cache import AssessmentCache
-
-    with tempfile.TemporaryDirectory() as directory:
-      params = Params(directory)
-      cache = AssessmentCache(Path(directory) / 'cache')
-      inventories = []
-      for number in range(1, 6):
-        current, inventory = self.inputs(number)
-        inventories.append(inventory)
-
-        def checks(actual, group, cohort, candidate, root, *, permitted, require_change=True):
-          self.assertTrue(permitted())
-          return {'qualified': True, 'windows': [{'window': key, 'qualified': True,
-                  'source_sha256': f'{int(next(w["route"] for w in group["selected_original_windows"] if w["identity"] == key).split("--")[0], 16):064x}'}
-                  for role in ('training', 'heldout') for key in cohort[role]['windows']]}
-
-        with patch.object(module, 'latest_route', return_value=inventory), patch.object(module, 'pending_routes', return_value=[inventory]), patch(
-          'openpilot.starpilot.flm.offroad_processor.collect_latest_steering_inputs', return_value=(inventory, current)
-        ), patch('openpilot.starpilot.galaxy.drive_history.DriveHistory.snapshot', return_value={'scanIncomplete': False, 'routes': inventories}):
-          worker = module.OffroadWorker(params, directory, cache, permitted=lambda: True, checker=checks, scope_for=lambda cp, p: scope())
-          result = worker.run_once()
-        if number < 4:
-          self.assertFalse(result['configuration_changed'])
-        elif number == 4:
-          self.assertTrue(result['configuration_changed'])
-          self.assertEqual(result['status'], 'ready_for_next_drive')
-          self.assertGreater(result['configured_values']['ff_gain_left'], 0)
-          saved = params.get(KEY)
-        else:
-          self.assertFalse(result['configuration_changed'])
-          self.assertNotEqual(result['status'], 'accepted_after_drive')
-          self.assertEqual(params.get(KEY), saved, 'Old pooled evidence must not count as a new-drive success')
-
-
 class TestCompletedDriveBacklog(unittest.TestCase):
   @staticmethod
   def route(counter):
@@ -500,10 +295,7 @@ class TestShortTurnProposalCoverage(unittest.TestCase):
       return {
         'original_metadata': metadata,
         'cohorts': [{'training': {'windows': windows[:3]}, 'heldout': {'windows': windows[3:]}}],
-        'selected_original_windows': [
-          {'identity': wid, 'route': 'declared-window-reader', 'segment': 0, 'start_ns': i * 10**9, 'end_ns': (i + 1) * 10**9}
-          for i, wid in enumerate(windows)
-        ],
+        'selected_original_windows': [{'identity': wid, 'segment': 0, 'start_ns': i * 10**9, 'end_ns': (i + 1) * 10**9} for i, wid in enumerate(windows)],
       }
 
     short, stable = group('a'), group('b')

openpilot/starpilot/flm/tests/test_offroad_history_safety.py

@@ -1,99 +0,0 @@
-"""Cross-drive source isolation with declared inputs, not physical-drive claims."""
-
-from contextlib import contextmanager
-import hashlib
-from types import SimpleNamespace
-import unittest
-from unittest.mock import patch
-
-from openpilot.starpilot.flm import offroad_checks as checks
-from openpilot.starpilot.flm import offroad_prior_inputs as prior_inputs
-
-
-def receipt(number):
-  return SimpleNamespace(route_id=f'{number:08x}--aaaaaaaaaa', source_sha256=f'{number:064x}',
-                         completed_wall_ns=number, cp_sha256='a' * 64)
-
-
-class TestCrossDriveSourceSafety(unittest.TestCase):
-  def test_single_source_native_results_keep_source_receipt(self):
-    source = receipt(1)
-    raw = b'declared source'
-    source.segments = (('segment', hashlib.sha256(raw).hexdigest(), len(raw)),)
-    assessment = SimpleNamespace(receipt=source, source_inventory={'segments': [{'files': {}}]})
-    original = [{'identity': str(i), 'segment': 0, 'start_ns': i + 10, 'end_ns': i + 11} for i in range(3)]
-
-    @contextmanager
-    def closed(*args, **kwargs):
-      yield raw, 'declared', lambda: True, None
-
-    with patch.object(checks, '_closed_source', closed), patch.object(checks, 'decode_full_rlog_segment', return_value=[]), \
-         patch.object(checks, '_release_unused_analysis_heap'):
-      results = checks._original_results(assessment, original, ('0', '1', '2'), {}, '.', permitted=lambda: True)
-    self.assertEqual(len(results), 3)
-    self.assertTrue(all(row['route'] == source.route_id and row['source_sha256'] == source.source_sha256 for row in results))
-    self.assertTrue(all(not row['qualified'] for row in results))
-
-  def test_unavailable_retained_source_refuses_every_selected_window(self):
-    for error in (FileNotFoundError('evicted'), checks._Incomplete('Route source changed')):
-      with self.subTest(error=type(error).__name__):
-        results = self._dispatch_failure(error)
-        self.assertEqual(len(results), 3)
-        self.assertTrue(all(not row['qualified'] and row['reason'] == 'historical_dense_source_unavailable' for row in results))
-        self.assertTrue(all(row['source_sha256'] == receipt(1).source_sha256 for row in results))
-
-  def test_retained_source_cancellation_and_digest_mismatch_remain_fatal(self):
-    for error in (checks._Incomplete('Analysis cancelled'), ValueError('Original numerical input changed')):
-      with self.subTest(error=str(error)), self.assertRaises(type(error)):
-        self._dispatch_failure(error)
-    with self.assertRaises(FileNotFoundError):
-      self._dispatch_failure(FileNotFoundError('evicted'), cancelled=True)
-
-  def _dispatch_failure(self, error, *, cancelled=False):
-    source = SimpleNamespace(receipt=receipt(1))
-    assessment = SimpleNamespace(evidence_sources={source.receipt.route_id: source})
-    original = [{'identity': str(i), 'route': source.receipt.route_id} for i in range(3)]
-    entry = checks._original_results
-    allowed = True
-
-    def dispatch(actual, *args, **kwargs):
-      nonlocal allowed
-      if actual is assessment:
-        return entry(actual, *args, **kwargs)
-      if cancelled:
-        allowed = False
-      raise error
-
-    with patch.object(checks, '_original_results', side_effect=dispatch):
-      return checks._original_results(assessment, original, ('0', '1', '2'), {}, '.', permitted=lambda: allowed)
-
-  def test_prior_search_skips_every_pooled_source_before_decoding(self):
-    sources = {str(i): SimpleNamespace(receipt=receipt(i)) for i in (1, 2, 3)}
-    assessment = SimpleNamespace(receipt=receipt(4), evidence_sources=sources)
-    routes = [{'routeId': receipt(i).route_id, 'segments': [{'files': {'rlog': True}}]} for i in (1, 2, 3, 4)]
-    with patch.object(prior_inputs.DriveHistory, 'snapshot', return_value={'scanIncomplete': False, 'routes': routes}), \
-         patch.object(prior_inputs, 'collect_route_inputs') as collect:
-      found = prior_inputs.prior_originals(assessment, {'original_metadata': {}}, {}, '.', permitted=lambda: True)
-    self.assertIsNone(found)
-    collect.assert_not_called()
-
-  def test_independent_fit_rejects_pooled_route_even_with_distinct_windows(self):
-    self._assert_source_rejected(receipt(2))
-
-  def test_independent_fit_rejects_pooled_digest_even_with_distinct_route(self):
-    prior = receipt(1)
-    prior.source_sha256 = receipt(2).source_sha256
-    self._assert_source_rejected(prior)
-
-  def _assert_source_rejected(self, prior_receipt):
-    assessment = SimpleNamespace(receipt=receipt(4), evidence_sources={'old': SimpleNamespace(receipt=receipt(2))})
-    cohort = {'training': {'windows': ('a', 'b', 'c')}, 'heldout': {'windows': ('d', 'e', 'f')}}
-    prior = SimpleNamespace(receipt=prior_receipt, windows=tuple({'identity': str(i)} for i in range(6)))
-    with patch.object(checks, '_original_results') as replay:
-      with self.assertRaisesRegex(ValueError, 'Independent original same-vehicle input partitions required'):
-        checks.fit_independent_route(assessment, {'selected_original_windows': []}, cohort, prior, {'values': {}}, {}, '.', permitted=lambda: True)
-    replay.assert_not_called()
-
-
-if __name__ == '__main__':
-  unittest.main()

openpilot/starpilot/galaxy/tests/test_flm_review.mjs

@@ -8,8 +8,6 @@ import { SURFACE_NAMES, FRICTION_NAMES, offroadView, validTrainingResults } from
 
 const idea = (parameter, target = .005) => ({ parameter, baseline: 0, target, delta: target, windows: 14,
   reason: 'heldout_profile_does_not_support_training_step' })
-const evidence = '4 of 5 retained higher-speed right turn-exit sections across 3 drive(s): steering unwound too slowly. These are recorded sections, not a count of distinct turns.'
-assert.equal(ideaExplanation({ ...idea('unwind_taper_right'), evidence }), evidence)
 const route = { suggestions: [idea('highway_center_taper_max'), ...FRICTION_NAMES.slice(1).map(name => idea(name))],
   savedValues: null, saved: false, configurationState: null }
 const review = reviewModel(route)

openpilot/starpilot/galaxy/tests/test_flm_training_results.mjs

@@ -63,12 +63,6 @@ completed.current.details = {
 assert(validTrainingResults(completed))
 assert.equal(offroadView(completed).suggestions.length, 5)
 assert.equal(offroadView(completed).samples, 241573)
-const history = structuredClone(completed)
-history.current.details['Drives in steering evidence history'] = 4
-assert.equal(offroadView(history).evidenceDrives, 4)
-history.current.details['Drives in steering evidence history'] = 9
-assert.equal(offroadView(history).evidenceDrives, null)
-assert.equal(offroadView(completed).evidenceDrives, null, 'Older reports must not imply cross-drive evidence')
 assert.equal(tunerStatus(completed, true, true).title, 'Completed drive analyzed', 'Parked status must not mask completed analysis')
 assert.match(tunerStatus(completed, true, true).description, /none has passed/)
 assert.equal(completed.current.application.applied, false)
@@ -126,11 +120,3 @@ adjustedSlider.current.details['Full surface coordinates'][0][1] = .005
 assert.match(offroadCoordinateEvidence(adjustedSlider, SURFACE_NAMES[0]), /Auto Learn saved: \+0.005 for the next drive/)
 assert(surfaceRows(adjustedSlider, false).every(row => row.applied === null))
 console.log('Saved curve and slider adjustments, unsaved recommendations, absent history and current application reconcile without writes')
-
-const behavioral = structuredClone(completed)
-const observation = '4 of 5 retained right turn-exit sections: steering unwound too slowly. Not a count of distinct turns.'
-behavioral.current.details['Recorded coordinate recommendations'] = [['unwind_taper_right', 0, .005, 5, 'repeated_observed_behavior_requires_automatic_checks', observation]]
-assert(validTrainingResults(behavioral))
-assert.equal(offroadView(behavioral).suggestions[0].evidence, observation)
-behavioral.current.details['Recorded coordinate recommendations'][0][5] = 'x'.repeat(501)
-assert(!validTrainingResults(behavioral), 'Oversized behavioral descriptions are refused')

openpilot/starpilot/galaxy/tests/test_flm_ux_browser.mjs

@@ -11,14 +11,13 @@ import { fileURLToPath } from 'node:url'
 import { SURFACE_NAMES, FRICTION_NAMES, validTrainingResults } from '../web/js/flm-training-results.js'
 
 const root = resolve(dirname(fileURLToPath(import.meta.url)), '../web')
-const output = resolve('ci-results/flm-ux-browser.json')
+const output = resolve('ci-results/flm-offroad.json')
 const screenshots = [], checks = [], requests = [], errors = [], contrastResults = [], fullPageReflow = []
 const targetIndex = SURFACE_NAMES.indexOf('highway_center_taper_max')
 const running = child => child.exitCode === null && child.signalCode === null
 assert.equal(running({ exitCode: null, signalCode: 'SIGTERM' }), false, 'A signal-reaped child must not be awaited for a second exit')
 let scenario = 'ideas', parked = true, saveError = false, reportError = false, pending = null
 const manual = Object.fromEntries(SURFACE_NAMES.map(name => [name, 0]))
-const behaviorEvidence = '4 of 5 retained higher-speed center sections across 3 drive(s): repeated steering oscillation was measured. These are recorded sections, not a count of distinct turns. This is a possible adjustment, not a confirmed cause or improvement.'
 const hints = [['highway_center_taper_max', .005], ['friction_threshold_low', .002],
   ['friction_threshold_mid', .004], ['friction_threshold_fast', .005], ['friction_threshold_highway', .006]]
 function report() {
@@ -34,10 +33,10 @@ function report() {
       application: { applied: application, reason: application ? 'current_controller_application' : 'saved_configuration_is_not_current_application' },
       details: { 'Completed drive analyzed': 'declared-ui-drive', 'Recorded steering observations': 71357,
         'Analysis decision': scenario === 'saved' ? 'bounded_trial_configured_for_next_drive' : 'no_new_native_feasible_training_step',
-        'Drives in steering evidence history': 3, 'Automatic configuration changed': saved, 'Saved route adjustment count': saved ? 1 : 0,
+        'Automatic configuration changed': saved, 'Saved route adjustment count': saved ? 1 : 0,
         'Recorded coordinate recommendations': scenario === 'none' ? [] : hints.map(([name, target], index) => [name, 0,
           scenario === 'conflict' && index === 0 ? null : target, index === 0 ? 148 : 14,
-          scenario === 'conflict' && index === 0 ? 'conflicting_recorded_candidates' : index ? 'heldout_profile_does_not_support_training_step' : 'recorded_candidate_requires_automatic_checks', index === 0 ? behaviorEvidence : '']),
+          scenario === 'conflict' && index === 0 ? 'conflicting_recorded_candidates' : index ? 'heldout_profile_does_not_support_training_step' : 'recorded_candidate_requires_automatic_checks']),
         'Recorded steering response': { version: 1, domain: 'lateral_acceleration', samples: 46231, windows: 234, error_rms: .062,
           route: 'declared-ui-drive', segment: 1, source_start_ns: start, source_end_ns: points.at(-1)[0],
           sampling: 'input_ordered_original_endpoints_no_interpolation', points } } } }
@@ -214,7 +213,6 @@ try {
       assert.equal(await execute('return document.querySelectorAll(".gx-tuner__groups>button").length'), 7)
       assert.equal(await execute('return document.querySelectorAll(".gx-tuner__idea-grid .gx-tuner__idea").length'), 1)
       assert.equal(await execute('return document.querySelectorAll(".gx-tuner__curve-summary [data-coordinate]").length'), 4)
-      assert(await execute('return document.querySelector(".gx-tuner__idea-grid").textContent.includes(arguments[0])', behaviorEvidence))
       await layout(`${width}/${theme}/ideas`)
       if (width === 1280) assert(await execute('const a=document.querySelector(".gx-tuner__idea-grid").getBoundingClientRect(),b=document.querySelector(".gx-tuner__curve-summary").getBoundingClientRect();return Math.abs(a.top-b.top)<1&&a.right<b.left'), 'Desktop groups related findings side by side without overlapping')
       if (width === 320 || width === 416 || width === 1280) {

openpilot/starpilot/galaxy/web/js/flm-review.js

@@ -14,7 +14,6 @@ export function ideaStatus(row) {
 
 export function ideaExplanation(row) {
   if (row.target === null) return 'Different parts of the drive pointed to different settings. There is no single value to use.'
-  if (row.evidence) return row.evidence
   if (row.reason === 'heldout_profile_does_not_support_training_step')
     return 'Some driving supported this idea, but other parts did not. Auto Learn kept it out of your saved tune.'
   if (row.reason === 'independent_support_for_frozen_training_profile')

openpilot/starpilot/galaxy/web/js/flm-training-results.js

@@ -191,7 +191,6 @@ export const SURFACE_GROUPS = Object.freeze([
 ])
 
 const REASONS = {
-  repeated_observed_behavior_requires_automatic_checks: 'Repeated recorded behavior suggests this setting. Automatic saving still requires independent controller and steering-output checks.',
   heldout_profile_does_not_support_training_step: "Part of the drive suggested this setting; separate windows did not confirm it.",
   independent_support_for_frozen_training_profile: "Separate recorded windows support this starting point. Automatic approval is separate.",
   training_candidate_not_independently_confirmed: "Training windows suggested this setting; it is not independently confirmed.",
@@ -251,7 +250,7 @@ export const historyTitle = status => Object.hasOwn(HISTORY_LABELS, status) ? HI
 export function validCoordinateRecommendations(rows) {
   const names = [...SURFACE_NAMES, ...FRICTION_NAMES]
   return Array.isArray(rows) && rows.length <= names.length && new Set(rows.map(row => row?.[0])).size === rows.length &&
-    rows.every(row => Array.isArray(row) && [5, 6].includes(row.length) && (row.length === 5 || text(row[5], 500)) && names.includes(row[0]) &&
+    rows.every(row => Array.isArray(row) && row.length === 5 && names.includes(row[0]) &&
       typeof row[1] === 'number' && Number.isFinite(row[1]) && Math.abs(row[1]) <= 1000 &&
       (row[2] === null && row[4] === 'conflicting_recorded_candidates' ||
         typeof row[2] === 'number' && Number.isFinite(row[2]) && Math.abs(row[2]) <= 1000 &&
@@ -281,14 +280,12 @@ export function offroadView(report) {
   const savedSurface = stored ? fullCoordinates(details['Full surface coordinates']) : null
   const savedFriction = stored ? frictionCoordinates(details['Saved friction-curve adjustments']) : null
   return { route: details['Completed drive analyzed'], samples: details['Recorded steering observations'],
-    evidenceDrives: stamp(details['Drives in steering evidence history']) && details['Drives in steering evidence history'] <= 8 ?
-      details['Drives in steering evidence history'] : null,
     reason: details['Analysis decision'], saved: details['Automatic configuration changed'] === true,
     configurationState, savedValues: savedSurface ? { ...savedSurface, ...(savedFriction || {}) } : null,
     observed, checks, savedCount: stamp(savedCount) && savedCount <= 35 ? savedCount : null,
     retained: details['Earlier analysis shown while updating'] === true,
     suggestions: recommendations === undefined ? rows.map(([parameter, delta, windows, qualification]) => ({ parameter, delta, windows, qualification })) :
-      recommendations.map(([parameter, baseline, target, windows, reason, evidence = '']) => ({ parameter, baseline, target, evidence,
+      recommendations.map(([parameter, baseline, target, windows, reason]) => ({ parameter, baseline, target,
         delta: target === null ? null : target - baseline, windows, reason, qualification: 'controller_native_checks_not_passed' }))
         .filter(row => row.target === null || row.target !== { ...savedSurface, ...savedFriction }[row.parameter]),
     moreSuggestions: details['More recorded suggestions omitted'] === true }

openpilot/starpilot/galaxy/web/js/flm.js

@@ -316,7 +316,7 @@ export const FlmPage = {
       <section v-if="view === 'overview'" id="tuner-panel-overview" role="tabpanel" aria-labelledby="tuner-tab-overview" class="gx-tuner__panel">
         <section class="gx-card gx-tuner__drive" :data-tone="review.tone" aria-label="Drive review">
           <div class="gx-tuner__drive-mark" aria-hidden="true"><i class="bi" :class="review.tone === 'saved' ? 'bi-check2-circle' : review.count ? 'bi-lightbulb' : 'bi-steering-wheel'"></i></div>
-          <div class="gx-tuner__drive-copy"><p v-if="review.historical" class="gx-tuner__eyebrow">{{ offroad?.retained && reportFresh ? 'EARLIER DRIVE · ANALYSIS RUNNING' : review.historical ? 'EARLIER DRIVE REVIEW' : 'YOUR DRIVE REVIEW' }}</p><h3>{{ review.title }}</h3><p>{{ review.description }}</p><p class="gx-tuner__summary"><template v-if="offroad?.evidenceDrives > 1">{{ offroad.evidenceDrives }} drives in your steering history · </template>{{ review.count }} {{ review.count === 1 ? 'finding' : 'findings' }} · <template v-if="offroad">{{ savedSummary }}</template><template v-else-if="trackingValues">{{ amount(acceptedCount) }} accepted · saving not confirmed</template><template v-else>Saved tune not reported</template></p></div>
+          <div class="gx-tuner__drive-copy"><p v-if="review.historical" class="gx-tuner__eyebrow">{{ offroad?.retained && reportFresh ? 'EARLIER DRIVE · ANALYSIS RUNNING' : review.historical ? 'EARLIER DRIVE REVIEW' : 'YOUR DRIVE REVIEW' }}</p><h3>{{ review.title }}</h3><p>{{ review.description }}</p><p class="gx-tuner__summary">{{ review.count }} {{ review.count === 1 ? 'finding' : 'findings' }} · <template v-if="offroad">{{ savedSummary }}</template><template v-else-if="trackingValues">{{ amount(acceptedCount) }} accepted · saving not confirmed</template><template v-else>Saved tune not reported</template></p></div>
           <div class="gx-tuner__drive-actions"><button v-if="review.saved.length" type="button" class="gx-btn gx-btn--tonal" @click="selectView('tune')">{{ review.historical ? 'View earlier saved values' : 'View saved change' }}<i class="bi bi-arrow-right" aria-hidden="true"></i></button><button v-if="review.count" type="button" class="gx-btn gx-btn--tonal gx-tuner__review-button" data-action="view-findings" @click="selectView('tune')">{{ review.saved.length ? 'View other findings' : 'View ' + review.count + ' findings' }}<i class="bi bi-arrow-right" aria-hidden="true"></i></button></div>
         </section>
         <div class="gx-tuner__overview-grid"><section class="gx-card gx-tuner__section gx-tuner__response"><div class="gx-tuner__section-heading"><div><p class="gx-tuner__eyebrow">STEERING RESPONSE</p><h3>How your car followed the steering request</h3></div><span class="gx-tuner__pill">{{ trackingPlot ? 'Completed window' : offroad?.observed ? 'One recorded section · ' + recordedDuration + 's' : offroad ? 'Recorded drive' : 'Waiting for samples' }}</span></div>

openpilot/starpilot/models/tests/test_laboratory_runner.py

@@ -36,6 +36,29 @@ class TestLaboratoryFrame(unittest.TestCase):
     return pair.run_frame(*frame_inputs(pair.lateral), log.ModelDataV2.Action(), .2, .2, 20.,
                           long_smooth_seconds=0., **kwargs)
 
+  def test_actual_modeld_turn_capability_supports_pairs_before_first_frame(self):
+    import ast
+    from types import SimpleNamespace
+    source = Path(__file__).resolve().parents[3] / "selfdrive/modeld/modeld.py"
+    main = next(node for node in ast.parse(source.read_text()).body if isinstance(node, ast.FunctionDef) and node.name == "main")
+    assignments = [node for node in ast.walk(main) if isinstance(node, ast.Assign) and
+                   any(isinstance(target, ast.Name) and target.id in ("turn_models", "turn_supported") for target in node.targets)]
+    caller = compile(ast.fix_missing_locations(ast.Module(body=assignments, type_ignores=[])), str(source), "exec")
+    def supported(model):
+      scope = {"model": model, "LaboratoryPair": LaboratoryPair}
+      exec(caller, scope)
+      return scope["turn_supported"]
+    pair = make_pair()
+    self.assertTrue(supported(pair))
+    self.assertTrue(supported(pair.lateral))
+    self.run_frame(pair)
+    for role in (pair.lateral, pair.longitudinal):
+      original = role.npy["desire"]
+      role.npy["desire"] = np.zeros((1, 5), dtype=np.float32)
+      self.assertFalse(supported(pair))
+      role.npy["desire"] = original
+    self.assertFalse(supported(SimpleNamespace(npy={})))
+
   def test_sequential_frame_one_warp_independent_actions_and_current_frame_pose(self):
     pair = make_pair()
     callback = Mock()

openpilot/starpilot/ui/onroad.py

@@ -85,7 +85,7 @@ def clock_text(now: datetime | None = None, use_24_hour: bool = False) -> str:
 
 def axis_status_color(state: OnroadState) -> rl.Color:
   """Retain StarPilot's existing off, AOL, long-only, and combined edge colors."""
-  lateral = state.lateral_active if state.aol_latched is None else state.aol_latched
+  lateral = state.lateral_active and state.aol_latched is not False
   if state.driver_override_active or state.longitudinal_overridden:
     return rl.Color(145, 155, 149, 255)
   if lateral and state.switchback_mode:

openpilot/starpilot/ui/tests/test_onroad_axes.py

@@ -55,7 +55,7 @@ class TestOnroadAxes(unittest.TestCase):
     intent = IntentState('card-ui', 1, NOW, NOW, NOW + 200_000_000, False, True, False, True, armed)
     self.ui.sm.put('aolIntentWire', encode_intent(replace(intent, **changes)))
 
-  def test_latch_keeps_border_blue_through_standstill_min_speed_and_reverse(self):
+  def test_armed_latch_keeps_intent_but_not_active_color_when_steering_pauses(self):
     self.ui.aol_runtime = True
     self.publish_latch()
     for brand in ('hyundai', 'gm', 'toyota', 'ford', 'honda'):
@@ -67,11 +67,21 @@ class TestOnroadAxes(unittest.TestCase):
           paused = self.state(False, False)
           self.assertTrue(getattr(paused, "aol_latched", False))
           self.assertFalse(paused.lateral_active or paused.longitudinal_active or paused.engaged)
-          self.assertEqual(rgba(axis_status_color(paused)), (10, 186, 181, 255))
+          self.assertEqual(rgba(axis_status_color(paused)), (18, 40, 57, 255))
     self.publish_latch(False)
     self.assertFalse(self.state(False, False).aol_latched)
     self.assertEqual(rgba(axis_status_color(self.state(False, False))), (18, 40, 57, 255))
 
+  def test_missing_model_cannot_show_armed_aol_as_active_steering(self):
+    self.ui.aol_runtime = True
+    self.publish_latch()
+    self.ui.sm.valid["modelV2"] = False
+    current = self.state(False, False)
+    self.assertTrue(current.aol_latched)
+    self.assertFalse(current.lateral_active)
+    self.assertEqual(rgba(axis_status_color(current)), (18, 40, 57, 255))
+    self.assertEqual(rgba(axis_status_color(self.state(True, False))), (10, 186, 181, 255))
+
   def test_cleared_latch_does_not_borrow_previous_active_steering_color(self):
     self.ui.aol_runtime = True
     self.publish_latch()
@@ -280,9 +290,9 @@ class TestOnroadAxes(unittest.TestCase):
       (OnroadAlert(size=AlertSize.FULL), blue), (OnroadAlert(size=AlertSize.SMALL, critical=True), blue))]
     cases = [(*case, None) for case in cases]
     cases += [(False, False, False, alert, color, True) for alert, color in (
-      (OnroadAlert(), (10, 186, 181, 255)),
-      (OnroadAlert(size=AlertSize.SMALL, text1='Steering Paused'), (10, 186, 181, 255)),
-      (OnroadAlert(size=AlertSize.MID), (10, 186, 181, 255)),
+      (OnroadAlert(), (18, 40, 57, 255)),
+      (OnroadAlert(size=AlertSize.SMALL, text1='Steering Paused'), (18, 40, 57, 255)),
+      (OnroadAlert(size=AlertSize.MID), (18, 40, 57, 255)),
       (OnroadAlert(size=AlertSize.FULL), blue), (OnroadAlert(size=AlertSize.SMALL, critical=True), blue))]
     cases = [(*case, False) for case in cases]
     cases += [(True, True, override, alert, color, None, True) for override, alert, color in (
@@ -313,6 +323,7 @@ class TestOnroadAxes(unittest.TestCase):
           view.background_layer = None
           view.projection_viewport = None
           view._corner_cache = Mock()
+          view._fade = Mock()
           view.current_speed = Mock()
           view.steering_wheel = Mock()
           view.compact_hud = Mock()

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "e8a20171bb1859dd229ce41edff2b79205545eefa557b242385f9f1f6b3a9653", "schema_version": 1}
+{"protected_inventory_sha256": "95956b8fbe518cf5d2f9fce5f44218c400ad2fafda771fab3412ffd45be77668", "schema_version": 1}