Accumulate per-drive FLM evidence and explain repeated steering behavior

.github/workflows/flm-offroad.yaml

@@ -64,16 +64,22 @@ 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-offroad.json
+          path: ci-results/flm-*.json
           if-no-files-found: warn

openpilot/selfdrive/modeld/modeld.py

@@ -554,8 +554,7 @@ def main(demo=False):
       meta_extra = meta_main
 
     sm.update(0)
-    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)
+    turn_supported = bool(model.npy.get('desire') is not None and model.npy['desire'].shape[-1] == 8)
     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):
+def analyze(windows, previous, *, permitted=lambda: True, route_order=None, comparison=None):
   """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):
     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 len({(w.domain, w.points[0].context_key) for w in windows}) > 1
+    or comparison is None and 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):
       groups.setdefault(key, []).append(window)
   results = []
   for key, rows in sorted(groups.items()):
-    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]
+    from openpilot.starpilot.flm.offroad_history import partition_windows
+
+    train, heldout = partition_windows(rows, route_order)
     if len(train) < 3 or len(heldout) < 3:
       results.append({'cohort': key, 'whole_windows': len(rows), 'reason': 'more_independent_windows_required'})
       continue
-    first, second = estimate(tuple(train), cohort=key, permitted=permitted), estimate(tuple(heldout), cohort=key, permitted=permitted)
+    first = estimate(tuple(train), cohort=key, permitted=permitted, comparison=comparison)
+    second = estimate(tuple(heldout), cohort=key, permitted=permitted, comparison=comparison)
     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

@@ -0,0 +1,94 @@
+"""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):
+  def _write(self, name, raw, *, maximum=MAX_DOCUMENT):
     self._owned()
-    if not 0 < len(raw) <= MAX_DOCUMENT:
+    if not 0 < len(raw) <= maximum:
       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
+from openpilot.starpilot.galaxy.drive_analysis import _closed_source, _Incomplete
 
 
 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 assessment.receipt.source_sha256 == prior.receipt.source_sha256
-    or assessment.receipt.route_id == prior.receipt.route_id
+    or any(receipt.source_sha256 == prior.receipt.source_sha256 or receipt.route_id == prior.receipt.route_id for receipt in source_receipts)
     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,12 +217,27 @@ def _release_unused_analysis_heap():
       trim(0)
 
 
-def _original_results(assessment, original, ids, values, root, *, permitted):
+def _original_results(assessment, original, ids, values, root, *, permitted, source_batch=False):
   spans = {w['identity']: w for w in original if w['identity'] in ids}
-  if not 3 <= len(ids) <= 24 or len(set(ids)) != len(ids) or set(spans) != set(ids):
+  if not (1 if source_batch else 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
@@ -301,7 +316,7 @@ def _original_results(assessment, original, ids, values, root, *, permitted):
       controls = []
       outputs = []
       missing = []
-  return results
+  return [dict(row, route=receipt.route_id, source_sha256=receipt.source_sha256) for row in results]
 
 
 def qualified_originals(results, cohort, *, require_change):

openpilot/starpilot/flm/offroad_daemon.py

@@ -103,6 +103,8 @@ 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))
@@ -120,7 +122,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['parameter'], s['baseline'], s['target'], s['whole_windows'], s['reason'], s.get('evidence', '')]
           for s in recommendations
         ]
         details['Recorded route suggestions'] = [

openpilot/starpilot/flm/offroad_history.py

@@ -0,0 +1,267 @@
+"""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,9 +118,11 @@ 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 LOCAL_ID.fullmatch(row['routeId']):
+    if row['routeId'] not in used_routes and 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))
@@ -139,6 +141,8 @@ 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 = []
@@ -172,6 +176,7 @@ 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,
@@ -193,7 +198,10 @@ def _prior_originals(assessment, group, selected_cohort, root, *, permitted, sta
         )
         if candidates:
           w = candidates[0]
-          profile_spans.append({'identity': w.identity, 'segment': number, 'start_ns': w.points[0].mono_ns, 'end_ns': w.points[-1].mono_ns})
+          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,
+          })
     if stable:
       chosen = _spread(chosen, 12)
       tracking = _spread(tracking, 12)

openpilot/starpilot/flm/offroad_processor.py

@@ -27,6 +27,8 @@ 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 {
@@ -40,6 +42,7 @@ 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
@@ -68,20 +71,39 @@ def window_demand(window):
   }
 
 
-def _separate_qualification(report, continuous, *, proposal_windows=None):
+def _separate_qualification(report, continuous, *, proposal_windows=None, route_order=None, role_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:
-      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)
+      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))
     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:
@@ -98,6 +120,7 @@ def _separate_qualification(report, continuous, *, proposal_windows=None):
   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,
@@ -133,6 +156,21 @@ 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
@@ -438,17 +476,40 @@ def collect_latest_steering_inputs(params, root, *, permitted, selected_route=No
   return route, collected
 
 
-def analyze_latest(params, root, previous, *, permitted, selected_route=None):
+def analyze_latest(params, root, previous, *, permitted, selected_route=None, evidence_bank=None, history_identity=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
@@ -465,17 +526,19 @@ def analyze_latest(params, root, previous, *, permitted, selected_route=None):
   for key, values in sorted(proposal_windows.items()):
     if not values:
       continue
-    report = analyze(tuple(values), amplitude_prior, permitted=permitted)
+    report = analyze(tuple(values), amplitude_prior, permitted=permitted, route_order=route_order, comparison=comparisons.get(key))
     if key in proposal_grades:
       report['tracking_source_grade'] = proposal_grades[key]
-      _separate_qualification(report, windows.get(key, ()))
+      _separate_qualification(report, windows.get(key, ()), route_order=route_order, role_windows=values)
     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,
@@ -486,6 +549,10 @@ def analyze_latest(params, root, previous, *, permitted, selected_route=None):
         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 = []
@@ -497,6 +564,7 @@ def analyze_latest(params, root, previous, *, permitted, selected_route=None):
       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,
@@ -509,18 +577,20 @@ def analyze_latest(params, root, previous, *, permitted, selected_route=None):
   for key, values in sorted(stable_windows.items()):
     if not values:
       continue
-    report = analyze_stable_windows(tuple(values), permitted=permitted)
+    report = analyze_stable_windows(tuple(values), permitted=permitted, route_order=route_order, comparison=comparisons.get(key))
     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)
+      _separate_qualification(report, windows.get(key, ()), proposal_windows=values, route_order=route_order)
     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,
@@ -533,6 +603,10 @@ def analyze_latest(params, root, previous, *, permitted, selected_route=None):
         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 = []
@@ -549,6 +623,7 @@ def analyze_latest(params, root, previous, *, permitted, selected_route=None):
     )
   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)
+    receipt, reports, counts, state['cp'], (), route, tuple(quality_stops), tuple(diagnostic), tuple(stable),
+    _recorded_tracking(proposal_windows), sources,
   )
 

openpilot/starpilot/flm/offroad_residual.py

@@ -77,15 +77,22 @@ 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):
+def estimate_residual(windows, *, cohort, permitted=lambda: True, comparison=None):
   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 len({(w.domain, w.points[0].context_key) for w in windows}) != 1
+    or comparison is None and 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 = []
@@ -116,7 +123,7 @@ def estimate_residual(windows, *, cohort, permitted=lambda: True):
     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(
-    windows[0].points[0].context_key,
+    context,
     windows[0].domain,
     cohort,
     tuple(w.identity for w in windows),
@@ -179,10 +186,11 @@ def compare_residual(before, after):
   }
 
 
-def analyze_stable_windows(windows, *, permitted=lambda: True):
+def analyze_stable_windows(windows, *, permitted=lambda: True, route_order=None, comparison=None):
   """Fixed input-only cohorts; calm references need no fictitious gain fit."""
   from dataclasses import asdict
-  from openpilot.starpilot.flm.offroad_analysis import cohort, _spread, MAX_COHORT_WINDOWS
+  from openpilot.starpilot.flm.offroad_analysis import cohort
+  from openpilot.starpilot.flm.offroad_history import partition_windows
 
   if type(windows) is not tuple or any(not w.valid() for w in windows):
     raise ValueError('Original Stable windows required')
@@ -195,12 +203,10 @@ def analyze_stable_windows(windows, *, permitted=lambda: True):
   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.
-    split = len(chosen) // 2
-    training, heldout = tuple(chosen[:split]), tuple(chosen[split:])
+    training, heldout = partition_windows(rows, route_order, contiguous=True)
     if len(training) < 3 or len(heldout) < 3:
       results.append({'cohort': key, 'whole_windows': len(rows), 'reason': 'more_independent_stable_windows_required'})
       continue
@@ -208,8 +214,8 @@ def analyze_stable_windows(windows, *, permitted=lambda: True):
       {
         'cohort': key,
         'whole_windows': len(rows),
-        'training': asdict(estimate_residual(training, cohort=key, permitted=permitted)),
-        'heldout': asdict(estimate_residual(heldout, cohort=key, permitted=permitted)),
+        'training': asdict(estimate_residual(training, cohort=key, permitted=permitted, comparison=comparison)),
+        'heldout': asdict(estimate_residual(heldout, cohort=key, permitted=permitted, comparison=comparison)),
         '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
+from openpilot.starpilot.galaxy.drive_analysis import _closed_source, _Incomplete
 
 
 def independent_profile_support(training, heldout, previous):
@@ -88,6 +88,7 @@ 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 = [
@@ -123,12 +124,15 @@ class StableRouteSelector:
     for group, ids in groups:
       if group['original_metadata']['domain'] != 'lateral_acceleration':
         continue
-      for window in group['selected_original_windows']:
+      for window in (*group['selected_original_windows'], *group.get('recommendation_original_windows', ())):
         if window['identity'] in ids:
-          if window['identity'] in self.spans and self.spans[window['identity']] != window:
+          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:
             raise ValueError('Conflicting original window identity')
-          self.spans[window['identity']] = window
-          comparison = dict(group['original_metadata'])
+          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)
           comparison.pop('route_values')
           comparison.pop('route_revision')
           context_key = hashlib.sha256(repr(comparison).encode()).hexdigest()
@@ -139,6 +143,23 @@ 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()
@@ -301,6 +322,18 @@ 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):
+def estimate(windows, *, cohort, permitted=lambda: True, comparison=None):
   _allowed(permitted)
   if (
     type(windows) is not tuple
@@ -84,9 +84,16 @@ def estimate(windows, *, cohort, permitted=lambda: True):
     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 len({(w.domain, w.points[0].context_key) for w in windows}) != 1
+    or comparison is None and 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 = []
@@ -151,7 +158,7 @@ def estimate(windows, *, cohort, permitted=lambda: True):
   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(
-    windows[0].points[0].context_key,
+    context,
     windows[0].domain,
     cohort,
     gain,

openpilot/starpilot/flm/offroad_worker.py

@@ -281,7 +281,8 @@ def recorded_recommendations(assessment, previous, report):
 
   found: dict[str, dict[str, Any]] = {}
 
-  def add(proposal, windows, reason):
+  def add(proposal, windows, reason, baseline=None):
+    baseline = previous if baseline is None else baseline
     if type(proposal) is not dict or type(windows) is not int or windows <= 0:
       return
     values = proposal.get('values')
@@ -293,11 +294,13 @@ 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: previous[name] + delta}
+      values = {name: baseline[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
@@ -308,10 +311,14 @@ 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']:
-      add(cohort.get('proposal'), cohort.get('whole_windows', 0), 'recorded_candidate_requires_automatic_checks')
+      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)
   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'))
@@ -319,6 +326,8 @@ 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]
 
 
@@ -355,7 +364,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['segment'], row['start_ns'], row['end_ns'])
+    (row.get('route', ''), 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'))
   }
@@ -390,6 +399,22 @@ 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 = []
@@ -459,10 +484,10 @@ class OffroadWorker:
 
     if self.on_analysis is not None:
       self.on_analysis()
-    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)
+    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 self.on_assessment is not None:
       self.on_assessment(assessment)
     if not state:
@@ -497,6 +522,96 @@ 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)
@@ -510,20 +625,22 @@ 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(assessment, scope, row['values'])):
+    elif row is not None and (row['state'] == 'ready_for_next_drive' or quality_withdrawn(actual_drive, scope, row['values'])):
       trial = trial_document(self.cache, row['analysis_sha256'])
       observed_checks = []
-      if not quality_withdrawn(assessment, scope, row['values']):
+      if not quality_withdrawn(actual_drive, scope, row['values']):
         stable = trial['before']['training'].get('kind') == 'uncommanded_residual_v1'
-        for group in assessment.stable_groups if stable else assessment.groups:
+        for group in actual_drive.stable_groups if stable else actual_drive.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(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)
+              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)
               observed_checks.append(
                 {'qualified': actual['qualified'] and contribution['qualified'], 'actual_recorded_map': actual, 'compared_with_prior_map': contribution}
               )
@@ -531,7 +648,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, assessment, scope, trial_started=started, recorded_backlog=route != latest)
+      outcome = assess_next_drive(trial, actual_drive, 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
@@ -574,84 +691,6 @@ 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()
@@ -687,7 +726,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[0] == b[0] and max(a[1], b[1]) <= min(a[2], b[2]) for a in training_spans for b in validation_spans
+              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
             ):
               reason = 'candidate_validation_already_used_for_training'
               continue
@@ -765,20 +804,35 @@ 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 = {
+              new: dict[str, Any] = {
                 'scope': scope.document(),
                 'values': proposal['values'],
                 'previous': current,
-                'route_sha256s': [assessment.receipt.source_sha256]
-                + ([] if independent_checks is None else [independent_checks['source_partitions']['heldout']]),
+                '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']})
+                ),
                 'source_start_ns': assessment.receipt.original_start_ns,
                 'source_end_ns': assessment.receipt.original_end_ns,
-                'completed_wall_ns': assessment.receipt.completed_wall_ns,
+                'completed_wall_ns': max(
+                  [assessment.receipt.completed_wall_ns] + [source.receipt.completed_wall_ns for source in assessment.evidence_sources.values()]
+                ),
                 'analysis_sha256': digest,
                 'sequence': 1 if row is None else row['sequence'] + 1,
                 'state': 'ready_for_next_drive',
               }
-              saved = RouteStorage(self.params).save(new, expected=expected, authorized=authorized)
+              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)
               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):
+    def assessed(params, root, previous, *, permitted, evidence_bank=None, history_identity=None):
       previous(cp)
       return assessment
 
@@ -1778,7 +1778,8 @@ class TestOffroadTrackingPolicy(unittest.TestCase):
 
     def window(segment, identity):
       points = tuple(
-        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)
+        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)
         for i in range(8)
       )
       return SimpleNamespace(identity=identity, points=points, domain='lateral_acceleration')
@@ -1796,7 +1797,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, cp_sha256='a' * 64, completed_wall_ns=10))
+    assessment = SimpleNamespace(receipt=SimpleNamespace(route_id=current_id, source_sha256='c' * 64, 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

@@ -0,0 +1,122 @@
+"""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,6 +16,211 @@ 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):
@@ -295,7 +500,10 @@ class TestShortTurnProposalCoverage(unittest.TestCase):
       return {
         'original_metadata': metadata,
         'cohorts': [{'training': {'windows': windows[:3]}, 'heldout': {'windows': windows[3:]}}],
-        'selected_original_windows': [{'identity': wid, 'segment': 0, 'start_ns': i * 10**9, 'end_ns': (i + 1) * 10**9} for i, wid in enumerate(windows)],
+        '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)
+        ],
       }
 
     short, stable = group('a'), group('b')

openpilot/starpilot/flm/tests/test_offroad_history_safety.py

@@ -0,0 +1,99 @@
+"""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,6 +8,8 @@ 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,6 +63,12 @@ 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)
@@ -120,3 +126,11 @@ 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,13 +11,14 @@ 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-offroad.json')
+const output = resolve('ci-results/flm-ux-browser.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() {
@@ -33,10 +34,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',
-        'Automatic configuration changed': saved, 'Saved route adjustment count': saved ? 1 : 0,
+        'Drives in steering evidence history': 3, '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']),
+          scenario === 'conflict' && index === 0 ? 'conflicting_recorded_candidates' : index ? 'heldout_profile_does_not_support_training_step' : 'recorded_candidate_requires_automatic_checks', index === 0 ? behaviorEvidence : '']),
         '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 } } } }
@@ -213,6 +214,7 @@ 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,6 +14,7 @@ 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,6 +191,7 @@ 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.",
@@ -250,7 +251,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) && row.length === 5 && names.includes(row[0]) &&
+    rows.every(row => Array.isArray(row) && [5, 6].includes(row.length) && (row.length === 5 || text(row[5], 500)) && 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 &&
@@ -280,12 +281,14 @@ 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]) => ({ parameter, baseline, target,
+      recommendations.map(([parameter, baseline, target, windows, reason, evidence = '']) => ({ parameter, baseline, target, evidence,
         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">{{ 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"><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-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,29 +36,6 @@ 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 and state.aol_latched is not False
+  lateral = state.lateral_active if state.aol_latched is None else state.aol_latched
   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_armed_latch_keeps_intent_but_not_active_color_when_steering_pauses(self):
+  def test_latch_keeps_border_blue_through_standstill_min_speed_and_reverse(self):
     self.ui.aol_runtime = True
     self.publish_latch()
     for brand in ('hyundai', 'gm', 'toyota', 'ford', 'honda'):
@@ -67,21 +67,11 @@ 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)), (18, 40, 57, 255))
+          self.assertEqual(rgba(axis_status_color(paused)), (10, 186, 181, 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()
@@ -290,9 +280,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(), (18, 40, 57, 255)),
-      (OnroadAlert(size=AlertSize.SMALL, text1='Steering Paused'), (18, 40, 57, 255)),
-      (OnroadAlert(size=AlertSize.MID), (18, 40, 57, 255)),
+      (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(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 (
@@ -323,7 +313,6 @@ 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": "95956b8fbe518cf5d2f9fce5f44218c400ad2fafda771fab3412ffd45be77668", "schema_version": 1}
+{"protected_inventory_sha256": "e8a20171bb1859dd229ce41edff2b79205545eefa557b242385f9f1f6b3a9653", "schema_version": 1}