diff --git a/openpilot/selfdrive/selfdrived/selfdrived.py b/openpilot/selfdrive/selfdrived/selfdrived.py
index dcc59065fa..6c12adabbe 100644
--- a/openpilot/selfdrive/selfdrived/selfdrived.py
+++ b/openpilot/selfdrive/selfdrived/selfdrived.py
@@ -504,9 +504,11 @@ class SelfdriveD:
         self.events.add(EventName.radarFault)
     hardware_events_added = len(self.events) - num_events
 
+    from opendbc.car.hyundai.ev6_aol import qualified as ev6_deadline_qualified
     input_proof = getattr(self, 'aol_step_input_proof', None)
     reconcile_input = bool(input_proof is not None and not REPLAY and self.aol_replay and self.axis_transport_required and
-                           self.CP.openpilotLongitudinalControl and qualified_ioniq6(self.CP) and
+                           self.CP.openpilotLongitudinalControl and
+                           (qualified_ioniq6(self.CP) or ev6_deadline_qualified(self.CP, marked_only=True)) and
                            input_proof[0] == self.aol_session_id and
                            (input_proof[8] or input_proof[9]))
     self.aol_event_input_reconcile = reconcile_input
@@ -806,6 +808,10 @@ class SelfdriveD:
 
     return CS
 
+  def source_deadline_deferral_supported(self):
+    from opendbc.car.hyundai.ev6_aol import qualified as ev6_qualified
+    return qualified_ioniq6(self.CP) or qualified_gm(self.CP) or ev6_qualified(self.CP, marked_only=True)
+
   def held_axis_override_wait(self, CS, proof, now_ns, *, fresh_successor=False):
     """Recognize an already applied override; never admit a new physical edge."""
     if (proof is None or not (qualified_ioniq6(self.CP) or qualified_gm(self.CP)) or not self.sm.seen['carControl'] or
@@ -968,14 +974,24 @@ class SelfdriveD:
 
   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) or qualified_gm(self.CP)):
+    if REPLAY or not self.aol_replay or not self.axis_transport_required or not self.source_deadline_deferral_supported():
       return False
     proof = getattr(self, 'aol_step_input_proof', None)
     if now_ns is None:
       now_ns = time.monotonic_ns()
     source = self.aol_car_state_log_ns
+    source_valid = self.conditional_car_state_valid
+    from opendbc.car.hyundai.ev6_aol import qualified as ev6_qualified
+    if (ev6_qualified(self.CP, marked_only=True) and source == 0 and not source_valid and
+        getattr(self, 'aol_expired_healthy_source', False) and
+        not getattr(self, 'car_state_samples', ()) and not getattr(self, 'car_state_queue_invalid', False) and
+        getattr(self, 'car_state_retained_valid', False)):
+      # A socket timeout cleared the selected source. Validate only the old
+      # publication's receipt; keep the source invalid and its clocks unchanged.
+      source = getattr(self, 'car_state_retained_source_ns', 0)
+      source_valid = True
     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
+        not source_valid or not 0 < source <= now_ns or
         now_ns - source <= INTENT_MAX_AGE_NS or source != proof[1] or
         now_ns >= proof[3] or self.aol_authority_lost or
         getattr(self, 'aol_fault_reason', 'none') != 'none' or
@@ -1443,7 +1459,7 @@ class SelfdriveD:
         final_ns = time.monotonic_ns()
         if ((axis.lateralActive or axis.longitudinalActive) and
             (not 0 < self.aol_car_state_log_ns <= final_ns or final_ns - self.aol_car_state_log_ns > INTENT_MAX_AGE_NS) and
-            (qualified_ioniq6(self.CP) or qualified_gm(self.CP))):
+            self.source_deadline_deferral_supported()):
           if (source_context is not None and source_context[1] and
               self.defer_expired_axis_publication(CS, source_context[0])):
             # Serialization crossed the same source boundary. Retain only the
diff --git a/openpilot/starpilot/aol/tests/test_ev6_source_deadline.py b/openpilot/starpilot/aol/tests/test_ev6_source_deadline.py
new file mode 100644
index 0000000000..3056884d4d
--- /dev/null
+++ b/openpilot/starpilot/aol/tests/test_ev6_source_deadline.py
@@ -0,0 +1,142 @@
+"""Recorded EV6 8f2e source cutoff keeps only its original unexpired proof."""
+from contextlib import ExitStack
+from unittest import mock
+
+import pytest
+from openpilot.cereal import log, messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests.test_ev6_physical_authorization import ev6_params
+from openpilot.starpilot.aol.tests.test_ordinary_source_availability import settings
+
+SOURCE = 10_020_000_000
+
+
+@pytest.mark.parametrize('socket_timeout', [False, True])
+@pytest.mark.parametrize('age_ns,published_offset_ns', [
+  (30_960_101, 4_945_854),
+  (31_244_572, 5_262_889),
+  (31_383_633, 5_039_179),
+  (30_890_079, 4_798_596),
+  (31_008_968, 5_325_976),
+  (31_351_205, 5_392_434),
+  (31_544_229, 5_502_541),
+  (32_161_269, 6_033_539),
+  (31_684_749, 5_741_086),
+  (31_896_523, 5_808_274),
+  (32_254_551, 5_671_607),
+  (31_565_476, 5_393_216),
+  (30_879_839, 5_223_629),
+])
+def test_ev6_recorded_source_cutoff_retains_only_original_unexpired_output(age_ns, published_offset_ns, socket_timeout):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    cp = ev6_params()
+    sd, controls, drive = case._authority_callers(True, cp=cp, genuine_events=True)
+    assert sd.CP.carFingerprint == 'KIA_EV6'
+    assert sd.CP.safetyConfigs[0].safetyParam == 0x815 and sd.CP.alternativeExperience == 32
+    for tick in (0, 1):
+      drive(tick, **settings('combined', 10_000_000_000 + tick * 10_000_000, tick + 1))
+    original, state, command = drive(2, state_stamp=SOURCE, offset_ns=published_offset_ns, **settings('combined', SOURCE, 3))
+    assert original.lateralActive and original.longitudinalActive and command.latActive and command.longActive
+    deadline = original.validUntilMonoTime
+    assert deadline == SOURCE + published_offset_ns + 30_000_000
+    old = original.to_dict()
+    timer = sd.state_machine.soft_disable_timer
+    sd.car_state_poller = mock.Mock()
+    sd.car_state_poller.poll.return_value = []
+    axis, state, command = drive(5, timeout=socket_timeout, state_stamp=SOURCE, offset_ns=age_ns - 30_000_000,
+                                 **settings('combined', SOURCE, 3))
+    assert sd.aol_axis_publication_deferred
+    if socket_timeout:
+      assert sd.aol_car_state_log_ns == 0 and not sd.conditional_car_state_valid
+    else:
+      assert sd.aol_car_state_log_ns == SOURCE and sd.conditional_car_state_valid
+    assert axis.to_dict() == old and axis.validUntilMonoTime == deadline
+    assert state.enabled and sd.state_machine.soft_disable_timer == timer
+    assert log.OnroadEvent.EventName.commIssue not in sd.events.names
+    # The consumer stops at the original cutoff; retaining a proof never renews it.
+    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
+    # A second stale loop cannot repeat the deferral even within that deadline.
+    next_axis, _, next_command = drive(5, timeout=True, state_stamp=SOURCE, offset_ns=age_ns - 30_000_000 + 100_000,
+                                       **settings('combined', SOURCE, 3))
+    assert not sd.aol_axis_publication_deferred
+    assert not next_axis.lateralActive and not next_axis.longitudinalActive
+    assert not next_command.latActive and not next_command.longActive
+
+
+@pytest.mark.parametrize('fault', ['can', 'epsTemporary', 'epsPermanent', 'acc', 'brake', 'gas',
+                                 'pandaRx', 'relay', 'nativeInvalid', 'uncalibrated'])
+def test_ev6_unexpired_output_never_defers_a_new_fault(fault):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    sd, _, drive = case._authority_callers(True, cp=ev6_params(), genuine_events=True)
+    for tick in (0, 1, 2):
+      stamp = 10_000_000_000 + tick * 10_000_000
+      axis, _, command = drive(tick, state_stamp=stamp, offset_ns=4_945_854 if tick == 2 else 0,
+                               **settings('combined', stamp, tick + 1))
+    assert axis.lateralActive and axis.longitudinalActive and command.latActive and command.longActive
+    state = fixture.car_state()
+    changes = {}
+    if fault == 'can':
+      state.canValid = False
+    elif fault == 'epsTemporary':
+      state.steerFaultTemporary = True
+    elif fault == 'epsPermanent':
+      state.steerFaultPermanent = True
+    elif fault == 'acc':
+      state.accFaulted = True
+    elif fault == 'brake':
+      changes['brake'] = True
+    elif fault == 'gas':
+      changes['gas'] = True
+    elif fault == 'pandaRx':
+      changes['panda_fault'] = True
+    elif fault == 'relay':
+      changes['relay_fault'] = True
+    elif fault == 'nativeInvalid':
+      changes['native_valid'] = False
+    elif fault == 'uncalibrated':
+      # Calibration publishes at 1 Hz; the normal fixture cadence correctly
+      # skips a replacement only 30 ms later. Deliver the new envelope itself.
+      stamp = SOURCE + 30_960_101
+      calibration = messaging.new_message('extrinsicsCalibration', valid=True, logMonoTime=stamp)
+      calibration.extrinsicsCalibration.calStatus = 'uncalibrated'
+      sd.sm.update_msgs(stamp / 1e9, [calibration.as_reader()])
+    if fault in ('can', 'epsTemporary', 'epsPermanent', 'acc'):
+      changes['observed_state'] = state
+    axis, _, command = drive(5, state_stamp=SOURCE, offset_ns=960_101,
+                             **settings('combined', SOURCE, 3), **changes)
+    if fault == 'uncalibrated':
+      assert sd.sm['extrinsicsCalibration'].calStatus == log.ExtrinsicsCalibration.Status.uncalibrated
+    assert not sd.aol_axis_publication_deferred
+    # The existing fault/pedal rules select the remaining axes; the previous
+    # combined proof cannot survive the new restriction.
+    assert not (axis.lateralActive and axis.longitudinalActive)
+    assert not (command.latActive and command.longActive)
+
+
+@pytest.mark.parametrize('timing', ['timely', 'late_poll', 'processing_crosses_deadline'])
+def test_ev6_fresh_pair_reconciles_without_extending_the_old_deadline(timing):
+  from openpilot.starpilot.aol.tests import test_input_deadline_continuity as continuity
+
+  def ev6_callers(case, full_op=True, *, stock_angle=False):
+    assert full_op and not stock_angle
+    sd, controls, drive = case._authority_callers(True, cp=ev6_params(), genuine_events=True)
+    assert sd.CP.carFingerprint == 'KIA_EV6'
+    assert sd.CP.safetyConfigs[0].safetyParam == 0x815 and sd.CP.alternativeExperience == 32
+    for tick in (0, 1):
+      stamp = 10_000_000_000 + tick * 10_000_000
+      drive(tick, **settings('combined', stamp, tick + 1))
+    return sd, controls, drive, None, None
+
+  # Reuse the existing timing stimulus with the captured EV6 configuration;
+  # production event, source, native and consumer checks remain real.
+  with mock.patch.object(fixture.IoniqAvailabilityTests, '_ioniq_availability_callers', ev6_callers):
+    continuity.test_fresh_source_replaces_expired_output_without_restoring_disengagement('ordinary', timing)
diff --git a/prebuilt.json b/prebuilt.json
index 65ba3a5607..c268ff6b36 100644
--- a/prebuilt.json
+++ b/prebuilt.json
@@ -1 +1 @@
-{"protected_inventory_sha256": "dc8ad8a50e217c02cb820e05312df6d418b5c0853da58033fd068049e19737fe", "schema_version": 1}
+{"protected_inventory_sha256": "f519a895eeac1226e4ab79722e1227633eee2369587b661939cb6e5f10c4ae2d", "schema_version": 1}
diff --git a/tools/ci/run_vehicle_tests.py b/tools/ci/run_vehicle_tests.py
index bcdeab0048..e3196550eb 100644
--- a/tools/ci/run_vehicle_tests.py
+++ b/tools/ci/run_vehicle_tests.py
@@ -24,6 +24,7 @@ from tools.ci import vehicle_pytest
 # Representative production callers plus their source/native refusal cases.
 # The emitted selected_test_ids record exact parametrized configurations.
 AOL_LIFECYCLE_TARGETS = (
+  'openpilot/starpilot/aol/tests/test_ev6_source_deadline.py',
   "openpilot/starpilot/car/rivian/tests/test_aol_policy.py",
   "openpilot/starpilot/car/rivian/tests/test_aol_native.py",
   "openpilot/starpilot/car/rivian/tests/test_aol_lifecycle.py",
@@ -114,6 +115,7 @@ DEBUG_SAFETY_CLASS_TARGETS = {
 
 
 RELEASE_TARGETS = (
+  'openpilot/starpilot/aol/tests/test_ev6_source_deadline.py',
   "openpilot/starpilot/aol/tests/test_ev9_main_withdrawal_guards.py",
   "opendbc_repo/opendbc/safety/tests/test_ev9_combined_aol.py::TestEV9CombinedReleaseDenied",
   "opendbc_repo/opendbc/car/hyundai/tests/test_ev9_combined_aol.py::TestEV9CombinedProfile",
diff --git a/tools/ci/tests/test_vehicle_ci.py b/tools/ci/tests/test_vehicle_ci.py
index 171dbf2851..6aaa6a564d 100644
--- a/tools/ci/tests/test_vehicle_ci.py
+++ b/tools/ci/tests/test_vehicle_ci.py
@@ -46,6 +46,11 @@ class TestVehicleCollection(unittest.TestCase):
     }
     self.assertLessEqual(expected, set(targets_for('safety-release')))
 
+  def test_ev6_source_deadline_refusals_run_in_lifecycle_and_release_ci(self):
+    target = 'openpilot/starpilot/aol/tests/test_ev6_source_deadline.py'
+    self.assertIn(target, targets_for('aol-lifecycle'))
+    self.assertIn(target, targets_for('safety-release'))
+
   def test_ev6_release_ci_exercises_real_alpha_startup_and_physical_owner(self):
     expected = {
       'openpilot/starpilot/tests/test_ev6_startup.py',