selfdrived: check bounded proof deferral after intent drain

Use the same post-drain clock for source-expiry deferral and intent
admission, avoiding premature state disable when draining crosses the
source freshness cutoff. Preserve the original proof and deadline.

Add cutoff, follow-through, and restrictive-input regression coverage.
Update the protected source inventory receipt.

Validation: 416 focused aggregate tests and 79 independent boundary and
restriction checks; exact source import audits, Ruff, and ty pass.
This addresses a reproduced admission race, not underlying producer lag.

openpilot/selfdrive/selfdrived/selfdrived.py

@@ -947,12 +947,13 @@ class SelfdriveD:
       CS = latest.carState
     return CS
 
-  def defer_expired_axis_publication(self, CS, initial_state):
+  def defer_expired_axis_publication(self, CS, initial_state, *, now_ns=None):
     """Keep an existing unexpired proof for one normal tick; issue no new lease."""
     if REPLAY or not self.aol_replay or not self.axis_transport_required or not qualified_ioniq6(self.CP):
       return False
     proof = getattr(self, 'aol_step_input_proof', None)
-    now_ns = time.monotonic_ns()
+    if now_ns is None:
+      now_ns = time.monotonic_ns()
     source = self.aol_car_state_log_ns
     if (getattr(self, 'aol_step_deferral_checked', False) or proof is None or proof != getattr(self, 'aol_published_input_proof', None) or
         not self.conditional_car_state_valid or not 0 < source <= now_ns or
@@ -1545,18 +1546,6 @@ 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.
@@ -1568,6 +1557,17 @@ class SelfdriveD:
     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()
+      # Intent draining can cross the source cutoff. Check the original proof
+      # against the same post-drain clock used by admission, before authority
+      # state changes. This retains no new input, proof, deadline, or enable.
+      if not previous_deferred and self.defer_expired_axis_publication(CS, initial_state, now_ns=now_ns):
+        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
       native = current_native(self.sm, self.CP, now_ns=now_ns, axis_session_id=self.aol_session_id)
       if native is None and self.initialized:
         lost_active_aol = self.aol_axis_decision.lateral_active or self.aol_axis_decision.longitudinal_active

openpilot/starpilot/aol/tests/test_early_source_handoff.py

@@ -63,8 +63,8 @@ def test_early(mode, condition):
           wires.append(wire(BASE + 31_000_000, 4, True).as_reader())
         return result
 
-      def defer(cs, initial):
-        out = old_defer(cs, initial)
+      def defer(cs, initial, **kwargs):
+        out = old_defer(cs, initial, **kwargs)
         attempt.append(out)
         return out
 

openpilot/starpilot/aol/tests/test_post_drain_source_cutoff.py

@@ -0,0 +1,123 @@
+from collections import deque
+from contextlib import ExitStack
+from dataclasses import replace
+from unittest import mock
+
+import pytest
+
+from openpilot.cereal import messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests.test_input_deadline_continuity import BASE, intent
+from openpilot.starpilot.aol.wire import decode_intent, encode_intent
+
+
+@pytest.mark.parametrize(('full_op', 'latched'), [(True, False), (True, True), (False, True)])
+@pytest.mark.parametrize('successor', ['fresh_before_deadline', 'fresh_after_deadline', 'no_replacement'])
+def test_source_cutoff_during_empty_drain_retains_only_original_output(full_op, latched, successor):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    with OpenpilotPrefix():
+      sd, controls, step, _, _ = case._ioniq_availability_callers(full_op=full_op)
+      old, _, command = step(2, offset_ns=4_530_833, state_stamp=BASE,
+                              intent_message=intent(BASE, 3, latched=latched), latched=latched)
+      assert old.lateralActive and command.latActive
+      assert old.longitudinalActive == full_op
+      original = sd.pm.messages['aolAxisState'].as_builder().to_bytes()
+      deadline = BASE + 34_530_833
+      assert old.validUntilMonoTime == deadline
+      clock = [BASE + 29_999_000]
+      original_drain = sd.drain_axis_intent
+
+      def drain(companion):
+        result = original_drain(companion)
+        assert companion.last_drain_count == 0
+        clock[0] = BASE + 30_010_236
+        return result
+
+      with mock.patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', side_effect=lambda: clock[0]), \
+           mock.patch.object(sd, 'drain_axis_intent', side_effect=drain), \
+           mock.patch.object(sd.state_machine, 'update', wraps=sd.state_machine.update) as transition:
+        axis, state, command = step(5, timeout=True, offset_ns=-1_000,
+                                   intent_message=intent(BASE, 3, latched=latched), latched=latched)
+      assert sd.aol_axis_publication_deferred and transition.call_count == 0
+      assert sd.pm.messages['aolAxisState'].as_builder().to_bytes() == original
+      assert axis.validUntilMonoTime == deadline and axis.sourceCarStateMonoTime == BASE
+      assert axis.lateralActive and command.latActive and state.enabled == full_op
+      assert axis.longitudinalActive == full_op and command.longActive == full_op
+      assert not sd.aol_authority_lost
+
+      if successor == 'no_replacement':
+        axis, state, command = step(6, timeout=True, latched=latched,
+                                   intent_message=intent(BASE, 3, latched=latched))
+        assert not sd.aol_axis_publication_deferred
+        assert not state.enabled and not axis.lateralActive and not axis.longitudinalActive
+        assert not command.latActive and not command.longActive
+        # Fresh input cannot restore ordinary LONG after a real disable.
+        axis, state, command = step(7, latched=latched, native_request_long=False)
+        assert not state.enabled and not axis.longitudinalActive and not command.longActive
+      else:
+        next_source = BASE + 31_546_896
+        decision = BASE + (32_000_000 if successor == 'fresh_before_deadline' else 36_068_506)
+        if successor == 'fresh_after_deadline':
+          with mock.patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', return_value=deadline + 1):
+            expired, _ = controls.state_control()
+          assert not expired.latActive and not expired.longActive
+        axis, state, command = step(6, offset_ns=decision - (BASE + 40_000_000), state_stamp=next_source,
+                                   intent_message=intent(next_source, 4, latched=latched), latched=latched)
+        assert axis.sourceCarStateMonoTime == next_source and axis.observedMonoTime == decision
+        assert axis.validUntilMonoTime == decision + 30_000_000
+        assert axis.lateralActive and command.latActive and state.enabled == full_op
+        assert axis.longitudinalActive == full_op and command.longActive == full_op
+        assert not sd.aol_authority_lost
+
+
+@pytest.mark.parametrize('restriction', ['pause_lateral', 'pause_longitudinal', 'unlatched', 'unarmed', 'settings',
+    'producer', 'sequence', 'future', 'invalid', 'deny_then_healthy', 'overrun'])
+def test_post_drain_deferral_cannot_hide_new_intent_restriction(restriction):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    with OpenpilotPrefix():
+      sd, _, step, _, _ = case._ioniq_availability_callers(full_op=True)
+      old, _, _ = step(2, offset_ns=4_530_833, state_stamp=BASE, intent_message=intent(BASE, 3))
+      assert old.lateralActive and old.longitudinalActive
+      clock = [BASE + 29_999_000]
+      stamp = BASE + 29_000_000
+      message = intent(stamp, 4)
+      decoded = decode_intent(message.aolIntentWire)
+      changes = {'pause_lateral': {'pauseLateral': True}, 'pause_longitudinal': {'pauseLongitudinal': True},
+                 'unlatched': {'allowedLatch': False}, 'unarmed': {'lateralArmed': False},
+                 'settings': {'settingsQualified': False}, 'producer': {'producerSessionId': 'replacement'},
+                 'sequence': {'sequence': 3}, 'future': {'observedMonoTime': BASE + 31_000_000},
+                 'deny_then_healthy': {'pauseLateral': True}}
+      if restriction in changes:
+        message.aolIntentWire = encode_intent(replace(decoded, **changes[restriction]))
+      if restriction == 'invalid':
+        message.valid = False
+      queue = deque([message.as_reader()])
+      if restriction == 'deny_then_healthy':
+        queue.append(intent(stamp + 1, 5).as_reader())
+      elif restriction == 'overrun':
+        queue = deque(intent(stamp + i, 4 + i).as_reader() for i in range(33))
+      original_receive = messaging.recv_one_or_none
+      original_drain = sd.drain_axis_intent
+
+      def receive(sock):
+        if sock is sd.aol_intent_companion.sock:
+          return queue.popleft() if queue else None
+        return original_receive(sock)
+
+      def drain(companion):
+        result = original_drain(companion)
+        clock[0] = BASE + 30_010_236
+        return result
+
+      with mock.patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', side_effect=lambda: clock[0]), \
+           mock.patch('openpilot.selfdrive.selfdrived.selfdrived.messaging.recv_one_or_none', side_effect=receive), \
+           mock.patch.object(sd, 'drain_axis_intent', side_effect=drain):
+        axis, state, command = step(5, timeout=True, offset_ns=-1_000, intent_message=intent(BASE, 3))
+      assert not sd.aol_axis_publication_deferred
+      assert not state.enabled and not axis.lateralActive and not axis.longitudinalActive
+      assert not command.latActive and not command.longActive

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "4119f1138f47eb02294053424e7a6cd4ec42ec2a34f02ba64dcb8afd721c8988", "schema_version": 1}
+{"protected_inventory_sha256": "29fe455dd74756e18f4a6975a11168c3b749d98c898e7f41f5ea38ad80aa965e", "schema_version": 1}