diff --git a/openpilot/starpilot/flm/live_tracking.py b/openpilot/starpilot/flm/live_tracking.py index d2a7606f80..471dddff6b 100644 --- a/openpilot/starpilot/flm/live_tracking.py +++ b/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): diff --git a/openpilot/starpilot/flm/tests/test_source_baseline_recheck.py b/openpilot/starpilot/flm/tests/test_source_baseline_recheck.py new file mode 100644 index 0000000000..fab75efd20 --- /dev/null +++ b/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()) diff --git a/openpilot/starpilot/flm/tests/test_tracking_source_context.py b/openpilot/starpilot/flm/tests/test_tracking_source_context.py index 4ef40e73f2..267091edb7 100644 --- a/openpilot/starpilot/flm/tests/test_tracking_source_context.py +++ b/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()) diff --git a/prebuilt.json b/prebuilt.json index ede7a91ac1..9a04cb0a3f 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "fabbd14a4c2dfdaf5c55e515d9b8ba8e9b35f495f28b017bde6ec5df54406c34", "schema_version": 1} +{"protected_inventory_sha256": "7f7cea782c8a2228724072e979f7768b0448776fa88c1dcefab7ceb0ecaf8edf", "schema_version": 1}