Recheck live tuning against the accepted baseline

openpilot/starpilot/flm/live_tracking.py

@@ -402,6 +402,18 @@ class TrialState(TypedDict):
   demand: NotRequired[DemandSignature]
   published: NotRequired[bool]
   prior_accepted: NotRequired[TrialState | None]
+  source_rechecked_before: NotRequired[tuple[TrackingEpisode, ...]]
+
+
+@dataclass
+class SourceRecheck:
+  """Inert A/B numbers; never a retained application or permission."""
+  intent: TrialState
+  before: tuple[TrackingEpisode, ...]
+  future: tuple[TrackingEpisode, ...]
+  context: tuple[LiveTrialBinding, ResponseCertificate, str]
+  baseline: list[TrackingEpisode]
+  first_ack_ns: int = 0
 
 
 class DiscoveryState(TypedDict):
@@ -508,6 +520,7 @@ class LiveTrackingOptimizer:
     self.paused_intent: tuple[TrialState, tuple[TrackingEpisode, ...], tuple[TrackingEpisode, ...]] | None = None
     self.paused_demand_context: tuple[LiveTrialBinding, ResponseCertificate, str] | None = None
     self.paused_baseline_ns = 0
+    self.source_recheck: SourceRecheck | None = None
     self.status, self.reason = 'collecting', 'qualified_tracking_episodes_required'
     self.model_status = 'unavailable'
 
@@ -518,6 +531,79 @@ class LiveTrackingOptimizer:
     self.paused_intent = self.paused_demand_context = None
     self.paused_baseline_ns = 0
 
+  def _check_source_recheck(self, now_ns: int) -> None:
+    check = self.source_recheck
+    if check is not None and (
+      type(now_ns) is not int or now_ns > check.intent['deadline_ns']
+      or check.context != (self.binding, self.certificate, self.domain)
+      or check.intent['baseline'] != self.accepted
+    ):
+      self.source_recheck = None
+      self.status, self.reason = 'collecting', 'return_baseline_recheck_expired_or_context_changed'
+
+  @staticmethod
+  def _source_batches_improved(before, future) -> bool:
+    old = [_score(row) for row in before]
+    new = [_score(row) for row in future]
+    mean_reversals = statistics.mean(row.reversal_hz for row in before)
+    reversal_limit = max(1.25 * mean_reversals, mean_reversals + 0.3)
+    return (
+      _feedback_result(old, new) > 0
+      and statistics.mean(row.reversal_hz for row in future) <= reversal_limit
+      and all(_score(row) <= 1.5 * max(statistics.mean(old), 0.03)
+        and row.reversal_hz <= max(reversal_limit, max(old_row.reversal_hz for old_row in before)) for row in future)
+    )
+
+  def _begin_source_recheck(self) -> None:
+    assert self.offer_state is not None
+    state = self.offer_state
+    self.source_recheck = SourceRecheck(state.copy(), self.before, tuple(self.future),
+      (self.binding, self.certificate, self.domain), [])
+    # This is an explicit return-to-baseline measurement, not a rejected
+    # experiment's rollback/retry. Provisional B is neither accepted nor saved.
+    baseline_offer = state.copy()
+    baseline_offer['phase'], baseline_offer['target'] = 'rollback', dict(state['baseline'])
+    self.offer_state = baseline_offer
+    self._drop_paused()
+    self._discard_selector()
+    self.before, self.future, self.settled_ns = (), [], None
+    self.status, self.reason = 'collecting', 'actual_accepted_baseline_ack_before_recheck_required'
+
+  def _observe_source_recheck(self, episode: TrackingEpisode, now_ns: int) -> bool:
+    check, state = self.source_recheck, self.offer_state
+    assert check is not None
+    if (state is None or state['phase'] != 'rollback' or not state['accepted']
+      or not check.first_ack_ns or episode.start_ns <= check.first_ack_ns + self.lag_ns
+      or self.last_ack is None
+      or self.last_ack[:4] != (state['trial_id'], state.get('target_sequence', self.sequence), 'rollback', self._hash(self.accepted))
+      or not 0 <= now_ns - self.last_ack[-1] <= 100_000_000
+      or not _matches_demand(check.intent['demand'], episode)
+      or check.baseline and episode.start_ns <= check.baseline[-1].end_ns):
+      return False
+    check.baseline.append(episode)
+    self.status, self.reason = 'collecting', 'measuring_accepted_baseline_after_trial'
+    if len(check.baseline) < check.intent['required_episodes']:
+      return True
+    self.source_recheck = None
+    if not self._source_batches_improved(check.baseline, check.future):
+      # No second chance on these outcomes. A remains the accepted map.
+      for name in check.intent['names']:
+        direction = 1 if check.intent['target'][name] > self.accepted[name] else -1
+        self.inconclusive_trials.append((name, direction, self._hash(self.accepted), check.intent['demand'], episode.end_ns))
+      self.status, self.reason = 'collecting', 'return_baseline_did_not_confirm_improvement'
+      return True
+    self.sequence += 1
+    trial = check.intent.copy()
+    trial['trial_id'] = hashlib.sha256(repr((trial['trial_id'], 'baseline-rechecked', self.sequence, now_ns)).encode()).hexdigest()
+    trial['target_sequence'], trial['published'], trial['accepted'] = self.sequence, False, False
+    trial['prior_accepted'] = state
+    trial['source_rechecked_before'] = check.before
+    # ORIGINAL start, deadline, fixed quotas and confidence allocation survive.
+    # B2 must earn a fresh full-map ACK, settling and a whole new future batch.
+    self.offer_state, self.before, self.future, self.settled_ns = trial, tuple(check.baseline), [], None
+    self.status, self.reason = 'trial_pending', 'rechecked_target_requires_new_application_and_performance'
+    return True
+
   def _check_paused_demand(self, now_ns: int) -> None:
     if self.paused_demand_context is not None and (
       self.paused_intent is None
@@ -543,6 +629,7 @@ class LiveTrackingOptimizer:
     self.status, self.reason = 'trial_pending', 'fresh_target_application_after_evidence_gap_required'
 
   def interrupt(self, reason: str, *, context_changed: bool = False) -> None:
+    self.source_recheck = None
     # Retain only inert numerical study intent on missing/stale evidence. The
     # command still rolls back; a fresh baseline ACK must precede a NEW wire
     # trial. Completed scores, alpha and original deadline are never reset.
@@ -591,6 +678,7 @@ class LiveTrackingOptimizer:
         self.offer_state = state.get('prior_accepted')
       else:
         state['phase'], state['target'] = 'rollback', dict(state['baseline'])
+        state.pop('source_rechecked_before', None)
         self.offer_state = state
     self.status, self.reason = 'paused', reason
     self.settled_ns = None
@@ -765,6 +853,7 @@ class LiveTrackingOptimizer:
 
   def observe_episode(self, episode: TrackingEpisode, *, now_ns: int) -> bool:
     episode.validate()
+    self._check_source_recheck(now_ns)
     if (self.rmse is None) != (episode.normalization_accel is not None):
       return False  # Never mix legacy force episodes with direct kinematic windows.
     if (
@@ -783,6 +872,8 @@ class LiveTrackingOptimizer:
       self._check_paused_demand(now_ns)
       if episode.profile_sha256 != self._hash(self.accepted):
         return False
+      if self.source_recheck is not None:
+        return self._observe_source_recheck(episode, now_ns)
       # Planning-only evidence retains completed accepted-profile geometry,
       # including discovery and zero-error comparison episodes. It never
       # supplies the scored baseline or any future trial outcome.
@@ -810,19 +901,15 @@ class LiveTrackingOptimizer:
       if self.selector_pending and self._revalidate_selector(episode, now_ns):
         return False
       if self.discovery is not None and self.rmse is None:
-        # A fixed-input source comparison cannot monopolize learning after
-        # that maneuver ends. Three unmatched whole windows release it by
-        # source time alone, without inspecting any tracking outcomes. No
-        # incomplete comparison can confirm or persist a target.
-        # Before launch, source-only discovery follows the current input regime.
-        # Waiting three unmatched windows also suppresses native discovery
-        # throughout a new short curve. No live trial or scored row is accepted.
+        # A short input gap keeps only completed baseline arithmetic. Release
+        # an ended regime after three window durations of original source
+        # time, so one unfinished comparison cannot monopolize discovery.
         last_match = self.comparison[-1].end_ns if self.comparison else self.discovery['selected_end_ns']
-        # A delayed completed window predating selection cannot describe a
-        # newer maneuver change. It remains excluded from the NEW baseline.
+        demand = (self.selector_hint['demand'] if self.selector_hint is not None
+                  else _demand(self.discovery['anchor']))
         if (episode.end_ns > self.discovery['selected_end_ns']
-            and not _matches(self.discovery['anchor'], episode)
-            and (self.rmse is None or episode.end_ns - last_match >= 3 * (episode.end_ns - episode.start_ns))):
+            and not _matches_demand(demand, episode)
+            and episode.end_ns - last_match >= 3 * (episode.end_ns - episode.start_ns)):
           self._discard_selector()
       if self.discovery is not None:
         discovery = self.discovery
@@ -830,8 +917,10 @@ class LiveTrackingOptimizer:
           self._discard_selector()
           self.status, self.reason = 'collecting', 'fresh_comparison_baseline_timeout'
           return False
+        demand = (self.selector_hint['demand'] if self.rmse is None and self.selector_hint is not None
+                  else _demand(discovery['anchor']))
         if (
-          not _matches(discovery['anchor'], episode)
+          not _matches_demand(demand, episode)
           or episode.start_ns <= discovery['selected_end_ns']
           or self.comparison
           and episode.start_ns <= self.comparison[-1].end_ns
@@ -901,10 +990,16 @@ class LiveTrackingOptimizer:
       if self.rmse is None:
         result = _feedback_result(old, new)
         if result > 0:
-          confirmed = state.copy()
-          confirmed['phase'] = 'confirmed'
-          self.offer_state = confirmed
-          self.status, self.reason = 'confirmed', 'consistent_later_tracking_improved'
+          original_before = state.get('source_rechecked_before')
+          if original_before is None:
+            self._begin_source_recheck()
+          elif not self._source_batches_improved(original_before, self.future):
+            self._rollback('original_baseline_improvement_not_repeated')
+          else:
+            confirmed = state.copy()
+            confirmed['phase'] = 'confirmed'
+            self.offer_state = confirmed
+            self.status, self.reason = 'confirmed', 'improved_against_original_and_returned_baselines'
         else:
           demand = state.get('demand', _demand(self.before[0]))
           for name in state['names']:
@@ -1287,6 +1382,7 @@ class LiveTrackingOptimizer:
     )
 
   def _rollback(self, reason: str) -> None:
+    self.source_recheck = None
     self._drop_paused()  # An actual scored rejection is terminal.
     if self.offer_state is None:
       raise ValueError('rollback requires a pending live trial')
@@ -1313,6 +1409,7 @@ class LiveTrackingOptimizer:
         direction = 1 if state['target'][name] > state['baseline'][name] else -1
         self.rejected_directions.append((name, direction, self._hash(state['baseline']), state.get('demand', _demand(self.before[0]))))
     state['phase'], state['target'] = 'rollback', dict(state['baseline'])
+    state.pop('source_rechecked_before', None)
     self.offer_state = state
     self._discard_selector()
     self.status, self.reason = 'rollback', reason
@@ -1333,6 +1430,7 @@ class LiveTrackingOptimizer:
   ) -> bool:
     from openpilot.starpilot.flm.live_tracking_response import validate_response_witness
 
+    self._check_source_recheck(now_ns)
     try:
       validate_response_witness(self.response_witness, self.binding, now_ns)
     except (TypeError, ValueError):
@@ -1382,6 +1480,10 @@ class LiveTrackingOptimizer:
         self.offer_state = accepted
         if self.selector_pending:
           self.selector_after_ns = max(self.selector_after_ns, control_ns)
+      if phase == 'rollback' and self.source_recheck is not None:
+        self.source_recheck.first_ack_ns = self.source_recheck.first_ack_ns or control_ns
+        self.status, self.reason = 'collecting', 'measuring_accepted_baseline_after_trial'
+        return True
       self._check_paused_demand(now_ns)
       if phase == 'rollback' and self.paused_demand_context is not None:
         self.paused_baseline_ns = control_ns
@@ -1424,6 +1526,7 @@ class LiveTrackingOptimizer:
     from openpilot.starpilot.flm.live_tracking_response import validate_response_witness
 
     self._check_paused_demand(now_ns)
+    self._check_source_recheck(now_ns)
     try:
       validate_response_witness(self.response_witness, self.binding, now_ns)
     except (TypeError, ValueError):

openpilot/starpilot/flm/tests/test_source_baseline_recheck.py

@@ -0,0 +1,357 @@
+"""Synthetic optimizer protocol tests; no physical/native/application proof."""
+from dataclasses import replace
+import hashlib
+import math
+import pytest
+
+from openpilot.starpilot.flm.canonical_surface_adapter import NAMES
+from openpilot.starpilot.flm.live_tracking import LiveTrackingOptimizer
+from openpilot.starpilot.flm.tests.test_live_tracking import episode
+from openpilot.starpilot.flm.tests.test_tracking_source_context import context, refresh
+
+
+class Drive:
+  """Declared whole episodes and typed ACK operands, never vehicle evidence."""
+  def __init__(self, initial=.0):
+    c = context()
+    self.owner = LiveTrackingOptimizer(c.binding, tuple((n, initial) for n in NAMES),
+      controller_domain='torque', response_certificate=c)
+    self.now = 1_000_000_000
+
+  def ack(self, offer=None, **changes):
+    o = self.owner
+    self.now += 1_000_000
+    refresh(o, self.now)
+    offer = offer or o.offer(self.now)
+    assert offer is not None
+    values = {'trial_id': offer.trial_id, 'target_sequence': offer.target_sequence, 'phase': offer.phase,
+      'target_sha256': offer.target_sha256, 'applied_full30': offer.target, 'binding': offer.binding,
+      'control_ns': self.now, 'now_ns': self.now, 'status': 'applied',
+      'confirmed_target_sha256': offer.target_sha256 if offer.phase == 'confirmed' else ''}
+    values.update(changes)
+    return o.acknowledge_application(**values)
+
+  def row(self, name='ff_gain_left', *, error=.045, goal=.09, profile=None, reversal=1., shape=None, renew=True, kind='subtle'):
+    o = self.owner
+    start = self.now + 1_000_000_000
+    self.now = start + 2_000_000_000
+    row = episode(o, int(start), name, kind='subtle', error=error, reversal=reversal, start=start)
+    row = replace(row, start_ns=start, end_ns=self.now, kind=kind, direction=1 if kind == 'large' else 0, speed_low=26.7, speed_high=26.9,
+      episode_id=hashlib.sha256(repr((start, name)).encode()).hexdigest(),
+      profile_sha256=profile or o._hash(o.accepted), goal_rms=goal, normalization_accel=max(.1, goal),
+      goal_shape=shape or (goal / max(.1, goal),) * 8)
+    refresh(o, self.now)
+    if renew and o.offer_state is not None and o.offer_state['phase'] == 'rollback':
+      assert self.ack()
+    return row, o.observe_episode(row, now_ns=self.now)
+
+  def trial(self, name='ff_gain_left', **row_values):
+    o = self.owner
+    for _ in range(20):
+      self.row(name, **row_values)
+      if o.offer_state is not None and o.offer_state['phase'] == 'trial':
+        offer = o.offer(self.now)
+        assert offer is not None
+        return offer
+    pytest.fail('No bounded source trial selected')
+
+  def batch(self, offer, name='ff_gain_left', **row_values):
+    rows = []
+    for _ in range(3):
+      row, _ = self.row(name, profile=offer.target_sha256, **row_values)
+      rows.append(row)
+    return rows
+
+
+def return_phase(d, name='ff_gain_left', **values):
+  trial = d.trial(name, **values)
+  assert d.ack(trial)
+  d.batch(trial, name, error=.03, goal=values.get('goal', .09), shape=values.get('shape'))
+  o = d.owner
+  assert o.offer_state['phase'] == 'rollback', 'A first positive batch must return to A, not confirm B'
+  assert o.source_recheck is not None
+  assert not o.offer_state['accepted'] and o.source_recheck.first_ack_ns == 0
+  return trial
+
+
+def reapply_phase(d, name='ff_gain_left'):
+  trial = return_phase(d, name)
+  assert d.ack()
+  for _ in range(3):
+    d.row(name, error=.045)
+  o = d.owner
+  reapplied = o.offer(d.now)
+  assert o.source_recheck is None and reapplied is not None and reapplied.phase == 'trial'
+  assert reapplied.trial_id != trial.trial_id and reapplied.target_sequence > trial.target_sequence
+  assert reapplied.started_ns == trial.started_ns and reapplied.deadline_ns == trial.deadline_ns
+  assert reapplied.target == trial.target and reapplied.baseline == trial.baseline
+  return trial, reapplied
+
+
+@pytest.mark.parametrize('name', NAMES)
+def test_each_coordinate_requires_return_baseline_and_fresh_reapplied_performance(name):
+  d = Drive(initial=.002)
+  o = d.owner
+  initial = dict(o.accepted)
+  first, second = reapply_phase(d, name)
+  assert o.accepted == initial and o.settled_ns is None and not o.future
+  assert len(o.before) == 3 and len(o.offer_state['source_rechecked_before']) == 3
+  assert not d.ack(first)  # Old B1 trial/sequence cannot acknowledge B2.
+  early, admitted = d.row(name, profile=second.target_sha256, error=.03)
+  assert not admitted and not o.future
+  assert d.ack(second)
+  for index in range(3):
+    d.row(name, profile=second.target_sha256, error=.03)
+    assert o.offer_state['phase'] == ('confirmed' if index == 2 else 'trial')
+    assert o.accepted == initial
+  confirmed = o.offer(d.now)
+  assert confirmed is not None and confirmed.phase == 'confirmed'
+  assert confirmed.trial_id == second.trial_id and confirmed.deadline_ns == first.deadline_ns
+  assert early.episode_id not in {row.episode_id for row in o.future}
+  assert d.ack(confirmed)
+  assert o.accepted == dict(second.target)
+  assert [n for n in NAMES if o.accepted[n] != initial[n]] == [name]
+
+
+def test_same_optimizer_can_earn_more_than_one_coordinate_without_losing_prior_values():
+  d = Drive()
+  o = d.owner
+  for name in NAMES:
+    previous = dict(o.accepted)
+    _, target = reapply_phase(d, name)
+    assert d.ack(target)
+    d.batch(target, name, error=.03)
+    confirmed = o.offer(d.now)
+    assert confirmed is not None and confirmed.phase == 'confirmed' and d.ack(confirmed)
+    assert all(o.accepted[n] == previous[n] for n in NAMES if n != name)
+    assert o.accepted[name] > previous[name]
+  assert all(o.accepted[n] > 0 for n in NAMES)
+
+
+@pytest.mark.parametrize('case', ['easier_road_within_matching_tolerance', 'substantially_easier_road', 'slower_input_with_unchanged_lag'])
+def test_null_response_cannot_confirm_when_fresh_baseline_has_same_apparent_benefit(case):
+  d = Drive()
+  if case == 'easier_road_within_matching_tolerance':
+    old = {'goal': .087, 'error': .0435}
+    new = {'goal': .075, 'error': .0375}
+  elif case == 'substantially_easier_road':
+    old = {'goal': .09, 'error': .045}
+    new = {'goal': .02, 'error': .01}
+  else:
+    def demand(f):
+      values = [math.sqrt(2)*.09*math.sin(2*math.pi*f*(i+.5)/1000) for i in range(2000)]
+      measured = [math.sqrt(2)*.09*math.sin(2*math.pi*f*((i+.5)/1000-.12)) for i in range(2000)]
+      rms = math.sqrt(sum(v*v for v in values)/len(values))
+      return {'goal': rms, 'error': math.sqrt(sum((a-b)**2 for a,b in zip(values,measured,strict=True))/len(values)),
+        'shape': tuple(sum(values[i:i+250])/250/max(.1,rms) for i in range(0,2000,250))}
+    old,new = demand(.5),demand(.45)
+  trial = d.trial(**old)
+  assert d.ack(trial)
+  d.batch(trial, **new)
+  o = d.owner
+  assert o.offer_state['phase'] == 'rollback', 'An easier input cannot immediately confirm an unchanged-response B'
+  assert o.source_recheck is not None
+  assert not any(o.accepted.values()) and d.ack()
+  for _ in range(3):
+    d.row(**new)
+  assert o.source_recheck is None and o.offer_state['phase'] == 'rollback' and o.offer_state['accepted']
+  assert o.reason == 'return_baseline_did_not_confirm_improvement'
+  assert not any(o.accepted.values())
+
+
+def test_repeated_baseline_ack_does_not_move_first_boundary_or_erase_new_baseline():
+  d = Drive()
+  return_phase(d)
+  o = d.owner
+  assert d.ack()
+  first = o.source_recheck.first_ack_ns
+  row,_ = d.row(error=.045)
+  assert [v.episode_id for v in o.source_recheck.baseline] == [row.episode_id]
+  for _ in range(10):
+    assert d.ack()
+    assert o.source_recheck.first_ack_ns == first
+    assert [v.episode_id for v in o.source_recheck.baseline] == [row.episode_id]
+
+
+@pytest.mark.parametrize('phase', ['baseline_ack_pending', 'baseline_one_row', 'reapplied_ack_pending', 'reapplied_one_row', 'confirmed_ack_pending'])
+def test_original_deadline_ends_every_recheck_transition(phase):
+  d = Drive()
+  o = d.owner
+  if phase.startswith('baseline'):
+    first = return_phase(d)
+    if phase == 'baseline_one_row':
+      assert d.ack()
+      d.row(error=.045)
+  else:
+    first, target = reapply_phase(d)
+    if phase != 'reapplied_ack_pending':
+      assert d.ack(target)
+      d.row(profile=target.target_sha256, error=.03)
+      if phase == 'confirmed_ack_pending':
+        d.row(profile=target.target_sha256, error=.03)
+        d.row(profile=target.target_sha256, error=.03)
+        assert o.offer_state['phase'] == 'confirmed'
+  d.now = first.deadline_ns + 1
+  refresh(o,d.now)
+  offer = o.offer(d.now)
+  assert o.source_recheck is None and not any(o.accepted.values())
+  assert offer is None or offer.phase == 'rollback'
+  assert not d.ack(first)
+
+
+@pytest.mark.parametrize('stage', ['first_trial', 'reapplied_trial'])
+def test_actual_gross_oscillation_rejection_remains_terminal(stage):
+  d = Drive()
+  o = d.owner
+  offer = d.trial() if stage == 'first_trial' else reapply_phase(d)[1]
+  assert d.ack(offer)
+  d.row(profile=offer.target_sha256, error=.03, reversal=10.)
+  assert o.reason == 'trial_tracking_or_oscillation_worse' and o.offer_state['phase'] == 'rollback'
+  assert getattr(o,'source_recheck',None) is None and o.paused_intent is None and not any(o.accepted.values())
+
+
+@pytest.mark.parametrize('reason,changed', [('manual_override',False), ('stale_control',False), ('changed_model',True)])
+def test_interrupted_return_baseline_never_retains_provisional_candidate(reason, changed):
+  d = Drive(initial=.002)
+  return_phase(d)
+  assert d.ack()
+  d.row(error=.045)
+  d.owner.interrupt(reason,context_changed=changed)
+  assert d.owner.source_recheck is None and d.owner.offer_state['phase'] == 'rollback'
+  assert all(value == .002 for value in d.owner.accepted.values())
+
+
+def test_pretrial_numbers_wait_for_original_demand_without_an_offer_or_expiry_renewal():
+  d = Drive()
+  o = d.owner
+  for _ in range(20):
+    d.row()
+    if o.discovery is not None:
+      break
+  assert o.discovery is not None
+  discovery = o.discovery
+  selected,expires = o.selector_hint['selected_ns'],o.selector_hint['expires_ns']
+  d.row()  # First whole independent baseline.
+  ids = tuple(v.episode_id for v in o.comparison)
+  assert len(ids) == 1
+  d.row(goal=.4,kind='large')
+  assert o.discovery is discovery and tuple(v.episode_id for v in o.comparison) == ids
+  assert o.selector_hint['selected_ns'] == selected and o.selector_hint['expires_ns'] == expires
+  assert o.offer(d.now) is None
+  for _ in range(2):
+    d.row()
+  assert o.offer_state is not None and o.offer_state['phase'] == 'trial'
+  assert {v.episode_id for v in o.before}.issuperset(ids)
+
+
+@pytest.mark.parametrize('bad', ['wrong_id', 'old_phase', 'wrong_hash', 'stale'])
+def test_return_baseline_requires_its_own_current_exact_ack(bad):
+  d = Drive()
+  first = return_phase(d)
+  o = d.owner
+  offer = o.offer(d.now)
+  changes = {'wrong_id': {'trial_id': 'f'*64}, 'old_phase': {'phase': 'trial'},
+    'wrong_hash': {'target_sha256': first.target_sha256},
+    'stale': {'control_ns': d.now-100_000_001}}[bad]
+  assert not d.ack(offer, **changes)
+  assert o.source_recheck.first_ack_ns == 0 and not o.offer_state['accepted']
+  row, admitted = d.row(error=.045, renew=False)
+  assert not admitted and not o.source_recheck.baseline
+  assert row.episode_id not in {v.episode_id for v in o.source_recheck.before}
+
+
+def test_return_baseline_excludes_pre_ack_partial_duplicate_and_unmatched_windows():
+  d = Drive()
+  return_phase(d)
+  o = d.owner
+  assert d.ack()
+  boundary = o.source_recheck.first_ack_ns
+  row,_ = d.row(error=.045)
+  assert len(o.source_recheck.baseline) == 1
+  refresh(o,d.now)
+  assert not o.observe_episode(row,now_ns=d.now)
+  assert len(o.source_recheck.baseline) == 1
+  unmatched, admitted = d.row(error=0.,goal=.4,kind='large')
+  assert not admitted and len(o.source_recheck.baseline) == 1
+  assert unmatched.episode_id not in {v.episode_id for v in o.source_recheck.baseline}
+  # A late delivery crossing the first A ACK cannot be repurposed as A2.
+  crossing = replace(row,episode_id='e'*64,start_ns=boundary-100_000_000,end_ns=boundary+1_000_000_000)
+  refresh(o,d.now)
+  assert not o.observe_episode(crossing,now_ns=d.now)
+  assert len(o.source_recheck.baseline) == 1
+
+
+@pytest.mark.parametrize('change', ['binding','certificate','domain','accepted'])
+def test_recheck_numeric_state_does_not_survive_identity_model_or_baseline_change(change):
+  d = Drive()
+  return_phase(d)
+  o = d.owner
+  assert d.ack()
+  if change == 'binding':
+    o.binding = replace(o.binding,source_epoch='f'*64)
+  elif change == 'certificate':
+    o.certificate = replace(o.certificate,observed_ns=o.certificate.observed_ns+1)
+  elif change == 'domain':
+    o.domain = 'torque_pid'
+  else:
+    o.accepted['ff_gain_left'] = .002
+  o.offer(d.now)
+  assert o.source_recheck is None
+  assert o.offer_state['phase'] == 'rollback'
+
+
+def test_zero_error_return_baseline_is_not_censored_to_keep_candidate():
+  d = Drive()
+  return_phase(d)
+  assert d.ack()
+  for _ in range(3):
+    d.row(error=0.)
+  o = d.owner
+  assert o.source_recheck is None and not any(o.accepted.values())
+  assert o.offer_state['phase'] == 'rollback' and o.offer_state['accepted']
+  assert o.reason == 'return_baseline_did_not_confirm_improvement'
+
+
+@pytest.mark.parametrize('stage', ['baseline_ack_pending','baseline_one_row','reapplied_ack_pending','reapplied_one_row','confirmed_ack_pending'])
+def test_missing_current_response_witness_cannot_advance_any_recheck_transition(stage):
+  d = Drive()
+  o = d.owner
+  if stage.startswith('baseline'):
+    return_phase(d)
+    if stage == 'baseline_one_row':
+      assert d.ack()
+      d.row(error=.045)
+  else:
+    _,target = reapply_phase(d)
+    if stage != 'reapplied_ack_pending':
+      assert d.ack(target)
+      d.row(profile=target.target_sha256,error=.03)
+      if stage == 'confirmed_ack_pending':
+        d.row(profile=target.target_sha256,error=.03)
+        d.row(profile=target.target_sha256,error=.03)
+  d.now += 100_000_001
+  assert o.offer(d.now) is None
+  assert o.source_recheck is None and not any(o.accepted.values())
+  assert o.offer_state is None or o.offer_state['phase'] == 'rollback'
+
+
+@pytest.mark.parametrize('cause',['expiry','context'])
+def test_waiting_pretrial_comparison_cannot_renew_or_survive_changed_context(cause):
+  d = Drive()
+  o = d.owner
+  for _ in range(20):
+    d.row()
+    if o.discovery is not None:
+      break
+  previous = o.discovery
+  assert previous is not None
+  d.row()
+  old_ids = {v.episode_id for v in o.comparison}
+  assert old_ids
+  if cause == 'expiry':
+    d.now = o.selector_hint['expires_ns']+1
+    d.row()
+  else:
+    o.interrupt('changed_context',context_changed=True)
+  assert o.discovery is not previous and not o.comparison and o.selector_hint is None
+  assert o.offer_state is None and not any(o.accepted.values())

openpilot/starpilot/flm/tests/test_tracking_source_context.py

@@ -377,6 +377,26 @@ def test_feedback_confirmation_still_needs_actual_target_ack_and_new_current_wit
     )
     refresh(o, e.end_ns)
     assert o.observe_episode(e, now_ns=e.end_ns)
+  returned = o.offer(e.end_ns)
+  assert returned is not None and returned.phase == 'rollback' and not any(o.accepted.values())
+  assert source_ack(o, returned, e.end_ns + 10_000_000)
+  now = e.end_ns
+  for i in range(3):
+    e = replace(episode(o, 200 + i, 'ff_gain_left', error=0.2,
+      start=now + 1_000_000_000 + i * 1_000_000_000), normalization_accel=0.5)
+    refresh(o, e.end_ns)
+    assert source_ack(o, o.offer(e.end_ns), e.end_ns)
+    assert o.observe_episode(e, now_ns=e.end_ns)
+  reapplied = o.offer(e.end_ns)
+  assert reapplied is not None and reapplied.phase == 'trial' and reapplied.trial_id != offer.trial_id
+  assert reapplied.deadline_ns == offer.deadline_ns and not any(o.accepted.values())
+  assert source_ack(o, reapplied, e.end_ns + 10_000_000)
+  now = e.end_ns
+  for i in range(3):
+    e = replace(episode(o, 300 + i, 'ff_gain_left', error=0.15, profile=reapplied.target_sha256,
+      start=now + 1_000_000_000 + i * 1_000_000_000), normalization_accel=0.5)
+    refresh(o, e.end_ns)
+    assert o.observe_episode(e, now_ns=e.end_ns)
   confirmed = o.offer(e.end_ns)
   assert confirmed is not None and confirmed.phase == 'confirmed' and not any(o.accepted.values())
   assert source_ack(o, confirmed, e.end_ns + 10_000_000)
@@ -453,7 +473,7 @@ def test_fitted_plant_does_not_reverse_identified_direction_at_boundary():
 
 
 @pytest.mark.parametrize('source_only', [False, True])
-def test_new_input_regime_retires_only_source_discovery_before_launch(source_only):
+def test_new_input_regime_does_not_erase_inert_pretrial_baseline(source_only):
   from openpilot.starpilot.flm.tests.test_live_tracking import optimizer, refresh as fitted_refresh
 
   o = owner() if source_only else optimizer()
@@ -473,12 +493,10 @@ def test_new_input_regime_retires_only_source_discovery_before_launch(source_onl
     changed = replace(changed, normalization_accel=max(0.1, changed.goal_rms))
   update(o, changed.end_ns)
   o.observe_episode(changed, now_ns=changed.end_ns)
-  assert (o.discovery is None) is source_only
+  assert o.discovery is previous and not o.comparison
   if source_only:
-    assert o.selector_hint is None and not o.comparison
-    assert o.baseline[-1].episode_id == changed.episode_id
-  else:
-    assert o.discovery == previous
+    assert o.selector_hint is not None
+    assert o.selector_hint['expires_ns'] == previous['expires_ns']
   assert o.offer_state is None and not any(o.accepted.values())
 
 

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "fabbd14a4c2dfdaf5c55e515d9b8ba8e9b35f495f28b017bde6ec5df54406c34", "schema_version": 1}
+{"protected_inventory_sha256": "7f7cea782c8a2228724072e979f7768b0448776fa88c1dcefab7ceb0ecaf8edf", "schema_version": 1}