diff --git a/openpilot/starpilot/flm/global_runtime.py b/openpilot/starpilot/flm/global_runtime.py index 80a08af1a6..2ad182ce6e 100644 --- a/openpilot/starpilot/flm/global_runtime.py +++ b/openpilot/starpilot/flm/global_runtime.py @@ -82,6 +82,7 @@ class GlobalRuntime: self.stage_effect = False self.reason = 'initializing' self.outcome = None + self._pending_trial_interrupted = False self.snapshot = () self.snapshot_sequence = 0 self.frame_sequence = 0 @@ -140,6 +141,7 @@ class GlobalRuntime: """Bounded current source/admission checks; invoked immediately before LaC.""" self.used, self.stage, self.frame, self.eligible = False, None, None, False self.outcome = None + self._pending_trial_interrupted = False self.controller.automatic_flm_effects = None self.source = None # Capture a fresh V2 generation and calibration interruption before ANY @@ -210,6 +212,7 @@ class GlobalRuntime: self._calibration_pending = None tracker.reject(now_ns=now_ns) self.outcome = None + self._pending_trial_interrupted = False self.centre_frame = None self.centre_raw = () self.centre_result = (0.0, 0.0) @@ -403,8 +406,9 @@ class GlobalRuntime: profile, now_ns=now_ns, active=active, enabled=source.enabled, manual=source.manual, healthy=healthy, replay_validation=validation )[0] self.reason = 'qualified' - except (AttributeError, KeyError, TypeError, ValueError, OverflowError): + except (AttributeError, KeyError, TypeError, ValueError, OverflowError) as error: self._neutral('unavailable') + self._pending_trial_interrupted = str(error) == 'interrupted pending trial cannot resume' def _current_live_trial(self, sm, envelope): from capnp.lib.capnp import _DynamicStructReader @@ -901,8 +905,18 @@ class GlobalRuntime: # preparation stamp and publication stamp are distinct but same-frame. from openpilot.starpilot.flm.live_trial_owner import LiveTrialApplication live_values = () - application = self.outcome live_owner = self.live_trial_owner + if (self._pending_trial_interrupted and live_owner is not None + and self.controller.automatic_flm_effects is None + and active and self.healthy and self.eligible and self.source.enabled + and not (self.source.manual or self.live_manual) + and 0 <= control_ns-self.now_ns <= 30_000_000 + and self.source.observed_ns <= control_ns <= self.source.expires_ns + and (not self.native_required or native_allowed)): + # The real controller refused this pending generation. Report rejection + # after current same-frame evidence, never fabricate an application ACK. + self.outcome = live_owner.rejected_pending(control_ns=control_ns) + application = self.outcome current_offer = None if live_owner is None else (live_owner.restore_offer or live_owner.offer) if ( type(application) is LiveTrialApplication diff --git a/openpilot/starpilot/flm/live_tracking.py b/openpilot/starpilot/flm/live_tracking.py index 9c2df05446..ad84caf1b0 100644 --- a/openpilot/starpilot/flm/live_tracking.py +++ b/openpilot/starpilot/flm/live_tracking.py @@ -498,6 +498,7 @@ class LiveTrackingOptimizer: transient = not context_changed and reason in ( 'acknowledged_trial_profile_withdrawn', 'expired_response_witness', 'stale_control', 'stale_applied_output', 'current_evidence_delayed', 'learning_evidence_delayed', 'consumed_source_mismatch', + 'consumer_pending_trial_interrupted', 'fresh_physical_residual_window_required', ) if not transient: diff --git a/openpilot/starpilot/flm/live_tracking_ingress.py b/openpilot/starpilot/flm/live_tracking_ingress.py index 914d00feb9..5b2b400b29 100644 --- a/openpilot/starpilot/flm/live_tracking_ingress.py +++ b/openpilot/starpilot/flm/live_tracking_ingress.py @@ -180,6 +180,23 @@ class LiveTrackingIngress: self.latest_offer is None or application.version != 1 or not 0 < application.sequence <= self.latest_offer.sequence): return False state = self.optimizer.offer_state + if application.status == 'rejected': + if (application.reason != 'pending_trial_interrupted' or application.binding != self.binding + or state['phase'] not in ('trial', 'confirmed') or state.get('accepted', False) + or application.trial_id != state['trial_id'] + or application.target_sequence != state.get('target_sequence', self.optimizer.sequence) + or application.confirmed_target_sha256 + or not state['started_ns'] < application.control_ns <= now_ns <= application.control_ns+100_000_000): + return False + try: + if full_profile_sha256(self.binding.baseline_revision, application.applied) != application.applied_sha256: + return False + except (TypeError, ValueError): + return False + # Withdraw the wire trial and retain only the existing inert-study rules. + # A fresh baseline ACK and new ID are still required; no lease is renewed. + self.interrupt('consumer_pending_trial_interrupted') + return False return self.optimizer.acknowledge_application(trial_id=application.trial_id, target_sequence=application.target_sequence, phase=state['phase'], target_sha256=application.applied_sha256, applied_full30=application.applied, diff --git a/openpilot/starpilot/flm/live_trial_owner.py b/openpilot/starpilot/flm/live_trial_owner.py index ff5ba04e58..b1a3cc9f0c 100644 --- a/openpilot/starpilot/flm/live_trial_owner.py +++ b/openpilot/starpilot/flm/live_trial_owner.py @@ -133,6 +133,17 @@ class TrackingTrialOwner: ) return True + def rejected_pending(self, *, control_ns): + """Current consumer refusal only; never an applied or confirmed target.""" + offer = self.offer + if (offer is None or offer.phase != 'trial' or not self.interrupted or self.source is not None + or type(control_ns) is not int or control_ns <= 0): + return None + actual = tuple((name, self.applied_full[name]) for name in NAMES) + return LiveTrialApplication(1, self.binding, offer.trial_id, offer.sequence, control_ns, + 'rejected', actual, full_profile_sha256(self.binding.baseline_revision, actual), '', + 'pending_trial_interrupted', offer.target_sequence) + def prepare_restore(self, offer, prior, *, now_ns, permission, binding): from openpilot.starpilot.flm.live_trial_contract import LiveRestoreOffer from openpilot.starpilot.flm.live_trial_store import LiveConfirmedPrior, scope, stable_binding diff --git a/openpilot/starpilot/flm/tests/test_live_trial_global_runtime.py b/openpilot/starpilot/flm/tests/test_live_trial_global_runtime.py index 7570070fdf..4ebf8e2277 100644 --- a/openpilot/starpilot/flm/tests/test_live_trial_global_runtime.py +++ b/openpilot/starpilot/flm/tests/test_live_trial_global_runtime.py @@ -415,3 +415,54 @@ def test_consumed_live_map_without_current_publication_proof_is_absent_not_neutr state_ns=int(fixture.controls.sm.logMonoTime['carState']), active=bool(command.latActive), limited=False) assert evidence is None assert fixture.runtime.outcome.applied == actual # Historical consumption is not relabelled as neutral. + + +def test_interrupted_pending_consumer_reports_rejection_after_actual_current_frame(): + from openpilot.starpilot.flm.application_wire import write_application + from openpilot.starpilot.flm.live_trial_wire import read_application + + with live_controls() as fixture: + proposal = fixture.initial_offer() + for _ in range(21): + fixture.frame(proposal) + for _ in range(11): + fixture.frame(proposal, publish=False) + for _ in range(9): + now, _ = fixture.frame(proposal) + fixture.assert_neutral() + owner = fixture.runtime.live_trial_owner + assert owner.interrupted and owner.offer.phase == 'trial' + evidence = fixture.runtime.capture_evidence(control_ns=now, + state_ns=int(fixture.controls.sm.logMonoTime['carState']), active=bool(fixture.last_command.latActive), limited=False) + assert evidence is not None and evidence.valid and evidence.native_allowed + refusal = fixture.runtime.outcome + assert type(refusal) is LiveTrialApplication and refusal.status == 'rejected' + assert refusal.reason == 'pending_trial_interrupted' and refusal.confirmed_target_sha256 == '' + assert refusal.trial_id == fixture.latest.trial_id and refusal.target_sequence == fixture.latest.target_sequence + assert not any(value for _, value in refusal.applied) and owner.confirmed_full == proposal.baseline + message = messaging.new_message('starpilotLateralState', valid=True, logMonoTime=now) + write_application(message.starpilotLateralState, refusal) + decoded = read_application(messaging.log_from_bytes(message.to_bytes()).starpilotLateralState, + now_ns=now, offer=fixture.latest) + assert decoded == refusal + assert fixture.controls.params.get('AutomaticSteeringLiveTrials') is None + + +@pytest.mark.parametrize('fault', ['driver', 'native', 'source', 'model', 'inactive']) +def test_pending_rejection_never_bypasses_current_consumer_faults(fault): + with live_controls() as fixture: + proposal = fixture.initial_offer() + for _ in range(20): + fixture.frame(proposal) + for _ in range(11): + fixture.frame(proposal, publish=False) + for _ in range(9): + fixture.frame(proposal) + assert fixture.tick == 99 # Next frame delivers real10Hz Panda and20Hz model/source denials. + now, _ = fixture.frame(proposal, driver=fault == 'driver', native_allowed=fault != 'native', + source_valid=fault != 'source', model_age=1_000_000_000 if fault == 'model' else 0, active=fault != 'inactive') + fixture.runtime.capture_evidence(control_ns=now, state_ns=int(fixture.controls.sm.logMonoTime['carState']), + active=bool(fixture.last_command.latActive), limited=False) + outcome = fixture.runtime.outcome + assert type(outcome) is not LiveTrialApplication + assert fixture.controls.params.get('AutomaticSteeringLiveTrials') is None diff --git a/openpilot/starpilot/flm/tests/test_source_study_gaps.py b/openpilot/starpilot/flm/tests/test_source_study_gaps.py index 511ad96000..63a12f8ef7 100644 --- a/openpilot/starpilot/flm/tests/test_source_study_gaps.py +++ b/openpilot/starpilot/flm/tests/test_source_study_gaps.py @@ -160,3 +160,73 @@ class SourceRampFreshness(unittest.TestCase): rows = [pair(o, 1_000_000_000), pair(o, 1_010_000_000)] pose = SimpleNamespace(source_lower_ns=1_005_000_000, source_upper_ns=1_015_000_000) assert o.original_pose_pair(rows, pose) is None + + +# Explicit numerical source fixtures below; real controller/wire rejection is +# independently exercised in test_live_trial_global_runtime. +def pending_refusal(): + from openpilot.starpilot.flm.live_tracking_ingress import LiveTrackingIngress + from openpilot.starpilot.flm.live_trial_owner import TrackingTrialOwner + + o = owner() + for i in range(5): + e = replace(episode(o, i, 'ff_gain_right'), normalization_accel=0.5) + refresh(o, e.end_ns) + o.observe_episode(e, now_ns=e.end_ns) + proposal = o.offer(e.end_ns) + assert proposal is not None and o.settled_ns is None + consumer = TrackingTrialOwner(proposal.binding) + assert consumer.prepare(proposal, now_ns=e.end_ns, permission=True, binding=proposal.binding) + consumer.neutral('unit_source_lapse') + refusal = consumer.rejected_pending(control_ns=e.end_ns+10_000_000) + assert refusal is not None and refusal.status == 'rejected' + ingress = LiveTrackingIngress(o.binding, o.certificate, tuple(o.accepted.items()), controller_domain='torque') + ingress.optimizer = o + ingress.latest_offer = proposal + return ingress, refusal, e.end_ns+10_000_000 + + +def test_actual_pending_refusal_rolls_back_before_target_ack_without_spending_new_deadline(): + from openpilot.starpilot.flm.tests.test_tracking_source_context import source_ack + + ingress, refusal, now = pending_refusal() + o = ingress.optimizer + original = o.offer_state.copy() + before = tuple(o.before) + assert not ingress.observe_application(refusal, now_ns=now) + assert o.offer_state['phase'] == 'rollback' and o.settled_ns is None + assert not any(o.accepted.values()) and ingress.latest_offer is None + assert o.paused_intent is not None and o.paused_intent[1] == before + refresh(o, now+20_000_000) + rollback = o.offer(now+20_000_000) + assert source_ack(o, rollback, now+30_000_000) + resumed = o.offer_state + assert resumed['phase'] == 'trial' and resumed['trial_id'] != original['trial_id'] + assert resumed['target_sequence'] > original['target_sequence'] and o.settled_ns is None + for field in ('started_ns', 'deadline_ns', 'alpha', 'required_episodes', 'target', 'baseline'): + assert resumed[field] == original[field] + assert tuple(o.before) == before and not any(o.accepted.values()) + + +def test_wrong_or_stale_pending_rejection_does_not_withdraw_current_trial(): + for kind in ('binding', 'trial', 'target_sequence', 'reason', 'stale', 'hash', 'confirmed', 'version'): + ingress, refusal, now = pending_refusal() + original = ingress.optimizer.offer_state.copy() + if kind == 'binding': + refusal = replace(refusal, binding=replace(refusal.binding, source_epoch='f'*64)) + elif kind == 'trial': + refusal = replace(refusal, trial_id='f'*64) + elif kind == 'target_sequence': + refusal = replace(refusal, target_sequence=refusal.target_sequence+1) + elif kind == 'reason': + refusal = replace(refusal, reason='unrelated') + elif kind == 'stale': + now += 100_000_001 + elif kind == 'hash': + refusal = replace(refusal, applied_sha256='f'*64) + elif kind == 'confirmed': + refusal = replace(refusal, confirmed_target_sha256='f'*64) + else: + refusal = replace(refusal, version=0) + assert not ingress.observe_application(refusal, now_ns=now), kind + assert ingress.optimizer.offer_state == original and ingress.optimizer.paused_intent is None, kind diff --git a/prebuilt.json b/prebuilt.json index c402a75358..b90645cd92 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "b836753088a9860f10b3d863c0f05352c0a4cdfffa654d9c763257a25df47ca7", "schema_version": 1} +{"protected_inventory_sha256": "ece1c7258d054a0f7290adf9ea477e122bcdd743534dc1b9e825201688b58202", "schema_version": 1}