Fix early source-expiry handoff without losing restrictive intent bursts

Check bounded retention after event processing before source-expiry handling mutates authority. Preserve only the original unexpired proof for one normal tick; track all intent drains in the step so a later empty drain cannot conceal an intermediate deny.

Preserve fd14 fresh-owner continuity semantics and reconcile five legacy test expectations. Add early handoff and engagement/deny recovery regressions. 239 distinct focused tests and 277 platform tests passed; Ruff, type checks and exact-source import/inventory audits passed.

No native artifacts or lease extensions. This does not establish a fix for the separate measured Card production stall.

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -562,7 +562,7 @@ class SelfdriveD:
     if CS.canValid:
       if self.tesla_stock_consumer.binding is not None:
         intent_companion = getattr(self, 'aol_intent_companion', None)
-        drained = intent_companion is None or intent_companion.drain()
+        drained = intent_companion is None or self.drain_axis_intent(intent_companion)
         now_ns = self.aol_car_state_log_ns if REPLAY and self.aol_car_state_log_ns else time.monotonic_ns()
         companion = current_intent(self.sm, car_state_ns=int(self.sm.logMonoTime['carState']), now_ns=now_ns,
                                    previous=self.aol_last_intent, companion=intent_companion, drain=False) if drained else None
@@ -818,6 +818,13 @@ class SelfdriveD:
           and not sample.buttonEvents for sample, valid in samples))
     return held_brake, held_gas
 
+  def drain_axis_intent(self, companion):
+    drained = companion.drain()
+    # Empty later drains must not erase earlier restrictive burst evidence.
+    self.aol_step_intent_drain_count = (getattr(self, 'aol_step_intent_drain_count', 0) +
+                                       getattr(companion, 'last_drain_count', 0))
+    return drained
+
   def event_input_wait_ns(self, CS, now_ns, *, final_held_override=False):
     """Bound an input wait by an original published lease; grant no new output."""
     proof = getattr(self, 'aol_published_input_proof', None)
@@ -850,7 +857,7 @@ class SelfdriveD:
       return 0
     wait_origin_ns = now_ns
     wait_deadline_ns = min(deadline, wait_origin_ns + int(DT_CTRL * 1e9))
-    if not companion.drain() or companion.integrity_epoch != epoch:
+    if not self.drain_axis_intent(companion) or companion.integrity_epoch != epoch:
       return 0
     # Transport may deliver intent published after the caller's clock sample.
     # Validate it against a post-drain clock without renewing the original wait.
@@ -858,7 +865,7 @@ class SelfdriveD:
     if not wait_origin_ns <= now_ns < wait_deadline_ns:
       return 0
     latest = companion.latest
-    if final_held_override and getattr(companion, 'last_drain_count', 0) > 1:
+    if final_held_override and getattr(self, 'aol_step_intent_drain_count', 0) > 1:
       # A burst could contain an intermediate restriction. Do not infer an
       # unchanged permission from just its final sample during this hold.
       return 0
@@ -1519,6 +1526,7 @@ class SelfdriveD:
     previous_deferred = getattr(self, 'aol_axis_publication_deferred', False)
     self.aol_axis_publication_deferred = False
     self.aol_step_deferral_checked = False
+    self.aol_step_intent_drain_count = 0
     previous_fault_reason = getattr(self, 'aol_fault_reason', 'none')
     self.aol_step_input_proof = getattr(self, 'aol_published_input_proof', None)
     CS = self.data_sample()
@@ -1531,6 +1539,18 @@ class SelfdriveD:
     finally:
       self.event_input_refresh_pending = False
     CS = self.event_car_state
+    # Event owners have checked every consumed physical transition. If the
+    # selected source ages out here, decide whether the original output proof
+    # can be retained before source-expiry handling mutates the authority state.
+    # No new input, proof, timestamp, deadline, or enable is generated.
+    if (not previous_deferred and self.defer_expired_axis_publication(CS, initial_state)):
+      self.aol_axis_publication_deferred = True
+      self.update_alerts(CS)
+      self.update_conditional_mode(CS)
+      self.publish_selfdriveState(CS, axis_published=True)
+      self.CS_prev = CS
+      del self.aol_step_input_proof
+      return
     if max(self.car_event_sample_count, len(getattr(self, 'car_state_samples', ()))) > 1 and any(self.events.contains(kind) for kind in (
         ET.USER_DISABLE, ET.SOFT_DISABLE, ET.IMMEDIATE_DISABLE)):
       # A queued enable cannot override a cancellation or pedal edge in this batch.
@@ -1539,7 +1559,7 @@ class SelfdriveD:
     lost_active_aol = bool(self.aol_replay and getattr(self, 'steering_authority_unavailable', False) and
                            (self.aol_axis_decision.lateral_active or self.aol_axis_decision.longitudinal_active))
     intent_companion = getattr(self, 'aol_intent_companion', None)
-    intent_drained = not self.aol_replay or intent_companion is None or intent_companion.drain()
+    intent_drained = not self.aol_replay or intent_companion is None or self.drain_axis_intent(intent_companion)
     if self.aol_replay or getattr(self, 'ordinary_axis_ack_required', False):
       now_ns = self.aol_car_state_log_ns if REPLAY and self.aol_car_state_log_ns else time.monotonic_ns()
       native = current_native(self.sm, self.CP, now_ns=now_ns, axis_session_id=self.aol_session_id)

openpilot/starpilot/aol/tests/test_early_source_handoff.py

@@ -0,0 +1,92 @@
+from collections import deque
+from contextlib import ExitStack
+from unittest import mock
+from dataclasses import replace
+import pytest
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.cereal import messaging
+from openpilot.selfdrive.selfdrived import selfdrived as owner
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests.test_long_only_input_wait import BASE, intent
+from openpilot.starpilot.aol.wire import decode_intent, encode_intent
+
+
+@pytest.mark.parametrize('mode', ['ordinary', 'held_gas', 'lateral_gas', 'held_brake'])
+@pytest.mark.parametrize('condition', ['timely', 'entry_expired', 'deadline', 'fcw', 'deny', 'physical_cancel', 'repeat'])
+def test_early(mode, condition):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as stack:
+    stack.callback(case.doCleanups)
+    with OpenpilotPrefix():
+      lateral = mode in ['lateral_gas', 'held_brake']
+      gas = mode in ['lateral_gas', 'held_gas']
+      brake = mode == 'held_brake'
+      sd, controls, step, card, card_receive = case._ioniq_availability_callers(full_op=not lateral)
+      actual_update_msgs = controls.sm.update_msgs
+
+      def joined_messages(timestamp, messages):
+        event = sd.pm.messages.get('onroadEvents')
+        if event is not None:
+          messages = [sample for sample in messages if sample.which() != 'onroadEvents'] + [event]
+        return actual_update_msgs(timestamp, messages)
+
+      stack.enter_context(mock.patch.object(controls.sm, 'update_msgs', side_effect=joined_messages))
+
+      def wire(stamp, seq, deny=False):
+        m = intent(stamp, seq, pause_lat=False)
+        v = decode_intent(m.aolIntentWire)
+        m.aolIntentWire = encode_intent(replace(v, allowedLatch=lateral, lateralArmed=lateral, pauseLateral=deny))
+        return m
+
+      original, _, cc = step(2, state_stamp=BASE, offset_ns=4_837_661, intent_message=wire(BASE, 3), gas=gas, brake=brake, steering=True, latched=lateral)
+      assert original.lateralActive
+      now = [BASE + (31_625_140 if condition == 'entry_expired' else 29_900_000)]
+      old_update = sd.update_events
+      old_receive = messaging.recv_one_or_none
+      wires = deque()
+      attempt = []
+      old_defer = sd.defer_expired_axis_publication
+
+      def receive(sock):
+        if sock is sd.aol_intent_companion.sock:
+          return wires.popleft() if wires else None
+        return old_receive(sock)
+
+      def events(cs):
+        result = old_update(cs)
+        now[0] = original.validUntilMonoTime + 1 if condition == 'deadline' else BASE + 31_625_140
+        if condition == 'fcw':
+          sd.events.add(owner.EventName.fcw)
+        if condition == 'physical_cancel':
+          sd.events.add(owner.EventName.buttonCancel)
+        if condition == 'deny':
+          wires.append(wire(BASE + 31_000_000, 4, True).as_reader())
+        return result
+
+      def defer(cs, initial):
+        out = old_defer(cs, initial)
+        attempt.append(out)
+        return out
+
+      with (
+        mock.patch.object(owner.time, 'monotonic_ns', side_effect=lambda: now[0]),
+        mock.patch.object(sd, 'update_events', side_effect=events),
+        mock.patch.object(sd, 'defer_expired_axis_publication', side_effect=defer),
+        mock.patch.object(messaging, 'recv_one_or_none', side_effect=receive),
+      ):
+        axis, state, command = step(5, timeout=True, offset_ns=-100_000, intent_message=wire(BASE, 3), gas=gas, brake=brake, steering=True, latched=lateral)
+      if condition in ['timely', 'repeat']:
+        assert axis.to_dict() == original.to_dict() and command.latActive
+        assert command.longActive == (mode == 'ordinary')
+        assert sd.aol_axis_publication_deferred and attempt == [True]
+        if condition == 'repeat':
+          # Next normal iteration gets no new source; the original deadline is
+          # still unexpired and a second deferral may not retain it again.
+          axis, _, command = step(5, timeout=True, offset_ns=2_000_000, intent_message=wire(BASE, 3), gas=gas, brake=brake, steering=True, latched=lateral)
+          assert not sd.aol_axis_publication_deferred and not command.latActive and not command.longActive
+      else:
+        assert not sd.aol_axis_publication_deferred
+        assert axis.faultReason != 'none' or not axis.lateralActive
+        assert not command.latActive and not command.longActive
+        if condition == 'fcw':
+          assert owner.EventName.fcw in sd.events.names

openpilot/starpilot/aol/tests/test_fresh_axis_engagement.py

@@ -0,0 +1,333 @@
+"""Fresh proofs preserve accepted engagement, while real revocation remains sticky.
+
+These tests use actual SelfdriveD event/state owners and Controls; fixture transport
+and clocks are synthetic. Card checks cover its existing intent-feedback owner.
+"""
+
+from contextlib import ExitStack
+
+from unittest import mock
+
+from dataclasses import replace
+
+from collections import deque
+
+
+import pytest
+
+from openpilot.cereal import log, messaging
+
+from openpilot.common.prefix import OpenpilotPrefix
+
+from openpilot.selfdrive.selfdrived import selfdrived as owner
+
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+
+from openpilot.starpilot.aol.tests.test_carstate_queue import message
+
+from openpilot.starpilot.aol.wire import IntentState, encode_intent, decode_safety, encode_safety
+
+BASE = 10_000_000_000
+
+OLD_SOURCE = BASE + 15_677_616
+
+RENEWAL_TIME = BASE + 51_614_499
+
+FRESH_SOURCE = BASE + 48_648_729
+
+
+def wire(ns, seq, **kw):
+  m = messaging.new_message('aolIntentWire', 0, valid=True, logMonoTime=ns)
+  v = IntentState('authority-card', seq, ns, ns, ns + 200_000_000, False, False, False, True, False)
+  m.aolIntentWire = encode_intent(replace(v, **kw))
+  return m
+
+
+def authority_callers(stack, full=True):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  stack.callback(case.doCleanups)
+  sd, controls, step, card, receive = case._ioniq_availability_callers(full_op=full)
+  axis, _, cmd = step(2, state_stamp=OLD_SOURCE, intent_message=wire(OLD_SOURCE, 3, allowedLatch=not full, lateralArmed=not full), latched=not full)
+  assert axis.lateralActive and cmd.latActive
+  assert axis.validUntilMonoTime == BASE + 50_000_000
+  return sd, controls, step, card, receive, axis
+
+
+@pytest.mark.parametrize(
+  'case',
+  [
+    'healthy',
+    'cancel',
+    'brake',
+    'regen',
+    'gas_cancel',
+    'native_long_deny',
+    'native_lat_deny',
+    'native_invalid',
+    'panda_fault',
+    'relay_fault',
+    'permanent_eps',
+    'can_invalid',
+    'can_timeout',
+    'disabled',
+    'inhibited',
+    'source_zero',
+    'producer',
+    'sequence',
+    'intent_pause',
+    'future_intent',
+    'settings',
+    'model',
+    'critical',
+  ],
+)
+def test_gap_owner_boundaries(case):
+  with ExitStack() as stack, OpenpilotPrefix():
+    sd, controls, step, card, receive, old = authority_callers(stack)
+    state = fixture.car_state()
+    kwargs = {}
+    w = wire(FRESH_SOURCE, 4)
+    if case == 'cancel':
+      state.buttonEvents = [fixture.car.CarState.ButtonEvent(type=fixture.car.CarState.ButtonEvent.Type.cancel, pressed=True)]
+    elif case == 'brake':
+      state.brakePressed = True
+    elif case == 'regen':
+      state.regenBraking = True
+    elif case == 'gas_cancel':
+      kwargs['gas'] = True
+      sd.disengage_on_accelerator = True
+    elif case == 'native_long_deny':
+      kwargs['long_allowed'] = False
+    elif case == 'native_lat_deny':
+      kwargs['allowed'] = False
+    elif case == 'native_invalid':
+      kwargs['native_valid'] = False
+    elif case == 'panda_fault':
+      kwargs['panda_fault'] = True
+    elif case == 'relay_fault':
+      kwargs['relay_fault'] = True
+    elif case == 'permanent_eps':
+      state.steerFaultPermanent = True
+    elif case == 'can_invalid':
+      state.canValid = False
+    elif case == 'can_timeout':
+      state.canTimeout = True
+    elif case == 'disabled':
+      sd.state_machine.state = log.SelfdriveState.OpenpilotState.disabled
+      sd.enabled = sd.active = False
+    elif case == 'inhibited':
+      sd.aol_availability_long_inhibit = True
+    elif case == 'source_zero':
+      with mock.patch.object(owner.time, 'monotonic_ns', return_value=BASE + 49_000_000):
+        sd._expire_axis_source(state, BASE + 49_000_000)
+      assert not sd.enabled
+    elif case == 'producer':
+      w = wire(FRESH_SOURCE, 4, producerSessionId='new-card')
+    elif case == 'sequence':
+      w = wire(FRESH_SOURCE, 3)
+    elif case == 'intent_pause':
+      w = wire(FRESH_SOURCE, 4, pauseLongitudinal=True, pauseLateral=True)
+    elif case == 'future_intent':
+      w = wire(FRESH_SOURCE, 4, observedMonoTime=RENEWAL_TIME + 1)
+    elif case == 'settings':
+      w = wire(FRESH_SOURCE, 4, settingsQualified=False)
+    elif case == 'model':
+      cal = messaging.new_message('extrinsicsCalibration', valid=True, logMonoTime=RENEWAL_TIME)
+      cal.extrinsicsCalibration.calStatus = 'uncalibrated'
+      sd.sm.update_msgs(RENEWAL_TIME / 1e9, [cal.as_reader()])
+    elif case == 'critical':
+      kwargs['critical_event'] = log.OnroadEvent.EventName.controlsMismatch
+    axis, pub, cmd = step(5, state_stamp=FRESH_SOURCE, intent_message=w, offset_ns=1_614_499, latched=False, observed_state=state, **kwargs)
+    healthy = case == 'healthy'
+    assert cmd.longActive == healthy, (case, pub.to_dict(), axis.to_dict())
+    assert axis.longitudinalActive == healthy
+    if case in (
+      'cancel',
+      'brake',
+      'regen',
+      'gas_cancel',
+      'native_long_deny',
+      'native_lat_deny',
+      'native_invalid',
+      'panda_fault',
+      'relay_fault',
+      'permanent_eps',
+      'can_invalid',
+      'can_timeout',
+      'disabled',
+      'inhibited',
+      'source_zero',
+      'producer',
+      'sequence',
+      'future_intent',
+      'settings',
+      'model',
+      'critical',
+    ):
+      again, _, cmd2 = step(6, intent_message=wire(BASE + 60_000_000, 5), latched=False)
+      assert not again.longitudinalActive and not cmd2.longActive, (case, again.to_dict())
+    if healthy:
+      assert not sd.aol_authority_lost and not getattr(sd, 'aol_availability_long_inhibit', False)
+      assert axis.sourceCarStateMonoTime == FRESH_SOURCE and axis.validUntilMonoTime == RENEWAL_TIME + 30_000_000
+      assert old.validUntilMonoTime == BASE + 50_000_000
+
+
+@pytest.mark.parametrize('pulse', ['cancel', 'brake', 'regen', 'gas', 'invalid_envelope', 'permanent_eps'])
+def test_queued_revoke_then_release_enable_cannot_renew(pulse):
+  with ExitStack() as stack, OpenpilotPrefix():
+    sd, controls, step, card, receive, old = authority_callers(stack)
+    first = fixture.car_state()
+    final = fixture.car_state()
+    final.buttonEnable = True
+    if pulse == 'cancel':
+      first.buttonEvents = [fixture.car.CarState.ButtonEvent(type=fixture.car.CarState.ButtonEvent.Type.cancel, pressed=True)]
+    elif pulse == 'brake':
+      first.brakePressed = True
+    elif pulse == 'regen':
+      first.regenBraking = True
+    elif pulse == 'gas':
+      first.gasPressed = True
+      sd.disengage_on_accelerator = True
+    elif pulse == 'permanent_eps':
+      first.steerFaultPermanent = True
+    pending = deque([message(FRESH_SOURCE, final)])
+    receive0 = messaging.recv_one_or_none
+    # Hook data_sample's blocking read, after the fixture prepares its envelope.
+    original = sd.data_sample
+
+    def data():
+      with (
+        mock.patch.object(owner.messaging, 'recv_one', return_value=message(FRESH_SOURCE - 1_000_000, first, valid=pulse != 'invalid_envelope')),
+        mock.patch.object(
+          owner.messaging,
+          'recv_one_or_none',
+          side_effect=lambda sock: (pending.popleft() if pending else None) if sock is sd.car_state_sock else receive0(sock),
+        ),
+      ):
+        return original()
+
+    with mock.patch.object(sd, 'data_sample', side_effect=data):
+      axis, pub, cmd = step(5, state_stamp=FRESH_SOURCE, intent_message=wire(FRESH_SOURCE, 4), offset_ns=1_614_499, latched=False)
+    assert not axis.longitudinalActive and not cmd.longActive
+    assert not sd.events.contains(fixture.ET.ENABLE)
+    again, _, cmd2 = step(6, intent_message=wire(BASE + 60_000_000, 5), latched=False)
+    assert not again.longitudinalActive and not cmd2.longActive
+
+
+@pytest.mark.parametrize('card_latched', [False, True])
+def test_controls_expire_before_fresh_renewal(card_latched):
+  with ExitStack() as stack, OpenpilotPrefix():
+    sd, controls, step, card, receive, old = authority_callers(stack)
+    card.aol_card_intent.allowed_latch = card_latched
+    before = sd.state_machine.state
+    with mock.patch.object(owner.time, 'monotonic_ns', return_value=BASE + 50_000_001):
+      expired, _ = controls.state_control()
+      assert not expired.latActive and not expired.longActive
+    receive(BASE + 50_000_001, state=fixture.car_state())
+    assert sd.state_machine.state == before
+    axis, pub, cmd = step(5, state_stamp=FRESH_SOURCE, intent_message=wire(FRESH_SOURCE, 4), offset_ns=1_614_499, latched=False)
+    assert pub.enabled and axis.longitudinalActive and cmd.longActive
+
+
+@pytest.mark.parametrize('change', ['axis_session', 'native_axis_session', 'native_long_deny', 'native_serial'])
+def test_native_identity_at_renewal(change):
+  with ExitStack() as stack, OpenpilotPrefix():
+    sd, controls, step, card, receive, old = authority_callers(stack)
+    original = sd.data_sample
+
+    def data():
+      result = original()
+      if change == 'axis_session':
+        sd.aol_session_id = 'replacement-session'
+      else:
+        value = decode_safety(sd.sm['aolSafetyWire'])
+        kw = (
+          {'axisSessionId': 'replacement-native-session'}
+          if change == 'native_axis_session'
+          else {'longitudinalAllowed': False}
+          if change == 'native_long_deny'
+          else {
+            'pandaSerial': 'replacement-panda',
+            'pandaInventory': tuple(
+              replace(slot, hardwareSerial='replacement-panda') if slot.hardwareSerial == 'authority-panda' else slot for slot in value.pandaInventory
+            ),
+          }
+        )
+        sample = messaging.new_message('aolSafetyWire', 0, valid=True, logMonoTime=RENEWAL_TIME)
+        sample.aolSafetyWire = encode_safety(replace(value, **kw))
+        sd.sm.update_msgs(RENEWAL_TIME / 1e9, [sample.as_reader()])
+      return result
+
+    with mock.patch.object(sd, 'data_sample', side_effect=data):
+      axis, pub, cmd = step(5, state_stamp=FRESH_SOURCE, intent_message=wire(FRESH_SOURCE, 4), offset_ns=1_614_499, latched=False)
+    assert not axis.longitudinalActive and not cmd.longActive
+
+
+def test_cancel_then_real_accepted_enable():
+  with ExitStack() as stack, OpenpilotPrefix():
+    sd, controls, step, card, receive, old = authority_callers(stack)
+    cs = fixture.car_state()
+    cs.buttonEvents = [fixture.car.CarState.ButtonEvent(type=fixture.car.CarState.ButtonEvent.Type.cancel, pressed=True)]
+    axis, pub, cmd = step(5, state_stamp=FRESH_SOURCE, intent_message=wire(FRESH_SOURCE, 4), offset_ns=1_614_499, latched=False, observed_state=cs)
+    assert not pub.enabled and not cmd.longActive
+    axis, pub, cmd = step(6, intent_message=wire(BASE + 60_000_000, 5), latched=False)
+    assert not pub.enabled and not cmd.longActive
+    enabled = fixture.car_state()
+    enabled.buttonEnable = True
+    axis, pub, cmd = step(7, intent_message=wire(BASE + 70_000_000, 6), latched=False, observed_state=enabled)
+    assert sd.events.contains(fixture.ET.ENABLE)
+    assert pub.enabled and cmd.longActive and axis.longitudinalActive
+
+
+@pytest.mark.parametrize('delivery', ['burst', 'split', 'healthy_fresh'])
+def test_prior_intent_drains_cannot_hide_restrictions_during_early_hold(delivery):
+  with ExitStack() as stack, OpenpilotPrefix():
+    sd, _, step, _, _, old = authority_callers(stack)
+    sd.car_state_poller = mock.Mock()
+    sd.car_state_poller.poll.return_value = []
+    now = [OLD_SOURCE + 29_900_000]
+    original_health = sd.update_process_health
+    original_receive = messaging.recv_one_or_none
+    denied = wire(BASE + 46_000_000, 4, pauseLateral=True, pauseLongitudinal=True).as_reader()
+    permitted = wire(BASE + 46_000_001, 5).as_reader()
+    queue = deque([denied, permitted] if delivery == 'burst' else [denied, None, permitted, None])
+    if delivery == 'healthy_fresh':
+      queue = deque([wire(FRESH_SOURCE - 2, 4).as_reader(), wire(FRESH_SOURCE - 1, 5).as_reader()])
+      now[0] = RENEWAL_TIME
+
+    def health():
+      result = original_health()
+      if delivery != 'healthy_fresh':
+        now[0] = BASE + 46_300_000
+      return result
+
+    def receive(sock):
+      if sock is sd.aol_intent_companion.sock:
+        return queue.popleft() if queue else None
+      return original_receive(sock)
+
+    with (
+      mock.patch.object(owner.time, 'monotonic_ns', side_effect=lambda: now[0]),
+      mock.patch.object(sd, 'update_process_health', side_effect=health),
+      mock.patch.object(owner.messaging, 'recv_one_or_none', side_effect=receive),
+    ):
+      if delivery == 'healthy_fresh':
+        axis, state, command = step(5, state_stamp=FRESH_SOURCE, intent_message=wire(FRESH_SOURCE, 6), offset_ns=1_614_499, latched=False)
+      else:
+        # Retain the already-consumed source while it is still fresh. Ordinary
+        # event work crosses its source cutoff, and refresh drains intent first.
+        axis, state, command = step(4, timeout=True, state_stamp=OLD_SOURCE, intent_message=wire(OLD_SOURCE, 3), offset_ns=6_300_000, latched=False)
+    assert not queue
+    assert not sd.aol_axis_publication_deferred
+    if delivery == 'healthy_fresh':
+      # Recording the count must not reject ordinary complete fresh decisions.
+      assert axis.longitudinalActive and command.longActive and state.enabled
+      assert axis.sourceCarStateMonoTime == FRESH_SOURCE
+      assert sd.aol_step_intent_drain_count == 2
+      following, _, command = step(6, intent_message=wire(BASE + 60_000_000, 7), latched=False)
+      assert following.longitudinalActive and command.longActive
+      assert sd.aol_step_intent_drain_count == 0
+    else:
+      assert not axis.longitudinalActive and not command.longActive
+      assert old.validUntilMonoTime == BASE + 50_000_000
+      assert sd.aol_step_intent_drain_count == 2

openpilot/starpilot/aol/tests/test_input_wait_clock.py

@@ -72,12 +72,13 @@ def test_last_refresh_intent_wait_uses_post_drain_clock(condition):
           return wires.popleft() if wires else None
         return original_receive(sock)
 
-      def wait(sample, sampled_now):
+      def wait(sample, sampled_now, **kwargs):
         nonlocal in_wait
         in_wait = True
         try:
-          result = original_wait(sample, sampled_now)
+          result = original_wait(sample, sampled_now, **kwargs)
-          waits.append((sampled_now, result, now[0]))
+          if not kwargs.get('final_held_override', False):
+            waits.append((sampled_now, result, now[0]))
           return result
         finally:
           in_wait = False
@@ -128,10 +129,28 @@ def test_last_refresh_intent_wait_uses_post_drain_clock(condition):
       if not denied:
         # Return duration remains anchored to the original caller timestamp.
         assert waits[0][0] + waits[0][1] == deadline
-      if condition in ('timely', 'pause_clear'):
+      if condition == 'missing':
+        # No new source is granted. The early check can retain the unchanged
+        # prior proof for this iteration only; expiry still withdraws output.
+        assert sd.aol_axis_publication_deferred and axis.to_dict() == original.to_dict()
+        with mock.patch('openpilot.selfdrive.controls.controlsd.time.monotonic_ns', return_value=deadline + 1):
+          expired, _ = controls.state_control()
+        assert not expired.latActive and not expired.longActive
+        missing, _, missing_command = step(6, timeout=True, intent_message=intent(BASE, 3),
+          latched=False, pause=True, native_request_lat=False, allowed=False)
+        assert not missing.longitudinalActive and not missing_command.longActive and not sd.enabled
+        recovered, _, recovered_command = step(7, latched=False, pause=True, native_request_lat=False, allowed=False)
+        assert not recovered.longitudinalActive and not recovered_command.longActive and not sd.enabled
+        return
+      # A new valid source renews an existing engagement even if the prior output
+      # expired; the original bounded wait still cannot extend that old lease.
+      renewed = condition in ('late_poll', 'processing_deadline')
+      if condition in ('timely', 'pause_clear') or renewed:
         assert axis.longitudinalActive and command.longActive and sd.enabled and sd.active
         assert not axis.lateralActive and not command.latActive
-        assert axis.sourceCarStateMonoTime == next_stamp and now[0] < deadline
+        assert axis.sourceCarStateMonoTime == next_stamp
+        assert (now[0] > deadline) == renewed
+        assert now[0] - axis.sourceCarStateMonoTime <= 30_000_000
       else:
         assert not axis.longitudinalActive and not command.longActive
         assert not axis.lateralActive and not command.latActive

openpilot/starpilot/aol/tests/test_long_only_input_wait.py

@@ -127,17 +127,35 @@ def test_long_only_wait_keeps_original_deadline_and_restrictions(timing):
         axis, _, command = step(5, timeout=True, offset_ns=338_798, intent_message=latest,
                                 latched=False, pause=True, native_request_lat=False, allowed=False,
                                 long_allowed=timing != 'native_denied', native_valid=timing != 'native_invalid', panda_fault=timing == 'panda_fault')
+      if timing == 'pre_refresh_cutoff':
+        # The source expires after the normal refresh cutoff. One early deferral
+        # retains exactly the old packet, without another state transition.
+        assert sd.aol_axis_publication_deferred and transition.call_count == 0
+        assert axis.to_dict() == original.to_dict() and poller.poll.call_count == 0
+        with mock.patch('openpilot.selfdrive.controls.controlsd.time.monotonic_ns', return_value=deadline + 1):
+          expired, _ = controls.state_control()
+        assert not expired.latActive and not expired.longActive
+        missing, _, missing_command = step(6, timeout=True, intent_message=latest,
+          latched=False, pause=True, native_request_lat=False, allowed=False)
+        assert not missing.longitudinalActive and not missing_command.longActive and not sd.enabled
+        recovered, _, recovered_command = step(7, latched=False, pause=True, native_request_lat=False, allowed=False)
+        assert not recovered.longitudinalActive and not recovered_command.longActive and not sd.enabled
+        return
       assert transition.call_count == 1
       if timing in ('native_denied', 'pause_set', 'pause_long', 'native_invalid', 'panda_fault', 'session_change', 'producer_change', 'pre_refresh_cutoff'):
         assert poller.poll.call_count == 0
       else:
         assert poller.poll.call_count == 1
-      if timing in ('timely', 'temporary_eps', 'pause_clear'):
+      # Expired output inhibits consumers; a fresh owner decision may renew an
+      # engagement that was never revoked. Source/native/physical negatives stay off.
+      renewed = timing in ('late_poll', 'processing_crosses_deadline', 'construction_crosses_deadline')
+      if timing in ('timely', 'temporary_eps', 'pause_clear') or renewed:
         assert poller.poll.call_count == 1 and not queued
         assert axis.sourceCarStateMonoTime == next_stamp
         assert not axis.lateralActive and not command.latActive
         assert axis.longitudinalActive and command.longActive and sd.enabled and sd.active
-        assert now[0] < deadline
+        assert (now[0] > deadline) == renewed
+        assert now[0] - axis.sourceCarStateMonoTime <= 30_000_000
         if timing == 'temporary_eps':
           assert fixture.log.OnroadEvent.EventName.steerTempUnavailableSilent in sd.events.names
       else:

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "3d88425721c1d189604572dbca407dbfbd033287cec60650329a1d893d606d22", "schema_version": 1}
+{"protected_inventory_sha256": "15b3e4616dfc9d157556c2bc33ffdc172cf3a29b677d48340a0f2c15915bee73", "schema_version": 1}