Report interrupted pending trials to the learner

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

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:

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,

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

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

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

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "b836753088a9860f10b3d863c0f05352c0a4cdfffa654d9c763257a25df47ca7", "schema_version": 1}
+{"protected_inventory_sha256": "ece1c7258d054a0f7290adf9ea477e122bcdd743534dc1b9e825201688b58202", "schema_version": 1}