Keep manager shutdown nonblocking and preserve paused AOL intent

openpilot/starpilot/aol/tests/test_repeated_warning_join.py

@@ -0,0 +1,123 @@
+from contextlib import ExitStack
+
+import pytest
+
+from opendbc.car.hyundai.values import CAR
+
+from openpilot.cereal import messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+
+
+@pytest.mark.parametrize('profile', [False, CAR.HYUNDAI_IONIQ_5_PE, CAR.KIA_EV9], ids=['ioniq6', 'ioniq5pe', 'ev9'])
+@pytest.mark.parametrize('order', ['event_first', 'axis_first', 'together'])
+def test_repeated_communication_warning_keeps_existing_latch(order, profile):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    sd, _, step, card, receive = case._ioniq_availability_callers(full_op=not profile, stock_angle=profile)
+    for tick in range(2, 65):
+      axis, _, command = step(tick, car_output_valid=False, native_request_long=False)
+      assert not axis.lateralActive and not command.latActive and not command.longActive
+      assert receive(10_000_000_000 + tick * 10_000_000, axis=sd.pm.messages['aolAxisState'],
+                     event=sd.pm.messages['onroadEvents'])
+    assert card.aol_transport_feedback.ordinary_warning_retained
+    axis, _, command = step(65, car_output_valid=False, native_request_long=False)
+    proof = sd.pm.messages['aolAxisState'].as_builder()
+    event = sd.pm.messages['onroadEvents'].as_builder()
+    now = 10_650_000_000
+    event.logMonoTime = now
+    proof.aolAxisState.faultEventMonoTime = now
+    frames = [{'event': event}, {'axis': proof}] if order == 'event_first' else (
+      [{'axis': proof}, {'event': event}] if order == 'axis_first' else [{'axis': proof, 'event': event}])
+    for index, frame in enumerate(frames):
+      assert receive(now + index * 1_000_000, **frame), 'REPEATED_SOFT_WARNING_CLEARED_LATCH'
+    assert not card.aol_transport_feedback.failed
+    assert not command.latActive and not command.longActive and not sd.enabled
+    recovered, _, command = step(66, native_request_long=False)
+    assert recovered.lateralActive and command.latActive
+    assert not command.longActive and not sd.enabled
+    assert receive(10_660_000_000, axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
+
+
+@pytest.mark.parametrize('order', ['axis_first', 'event_first'])
+@pytest.mark.parametrize('repeat', [False, True])
+def test_repeated_warning_missing_partner_keeps_original_deadline(order, repeat):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    sd, _, step, card, receive = case._ioniq_availability_callers()
+    for tick in (2, 3):
+      step(tick, car_output_valid=False, native_request_long=False)
+      assert receive(10_000_000_000 + tick * 10_000_000, axis=sd.pm.messages['aolAxisState'],
+                     event=sd.pm.messages['onroadEvents'])
+    step(4, car_output_valid=False, native_request_long=False)
+    now = 10_040_000_000
+    proof = sd.pm.messages['aolAxisState'].as_builder()
+    event = sd.pm.messages['onroadEvents'].as_builder()
+    proof.aolAxisState.faultEventMonoTime = event.logMonoTime = now
+    half = {'axis': proof} if order == 'axis_first' else {'event': event}
+    assert receive(now, **half), 'REPEATED_SOFT_WARNING_CLEARED_LATCH'
+    original_deadline = card.aol_transport_feedback.pending_event
+    assert original_deadline == now
+    def fresh_half(stamp):
+      updated = messaging.new_message('aolAxisState', valid=True, logMonoTime=stamp)
+      updated.aolAxisState = card.sm['aolAxisState']
+      updated.aolAxisState.sequence += 1
+      updated.aolAxisState.observedMonoTime = updated.aolAxisState.sourceCarStateMonoTime = stamp
+      updated.aolAxisState.validUntilMonoTime = stamp + 30_000_000
+      return {**half, 'axis': updated}
+
+    if repeat:
+      assert receive(now + 10_000_000, **fresh_half(now + 10_000_000))
+      assert card.aol_transport_feedback.pending_event == original_deadline
+    expired = now + 30_000_001
+    assert not receive(expired, **(fresh_half(expired) if repeat else half))
+    assert card.aol_transport_feedback.failed and not card.aol_card_intent.allowed_latch
+
+
+@pytest.mark.parametrize('fault', ['producer', 'epoch', 'session', 'fault_session', 'stale_source',
+                                   'future_source', 'invalid_native', 'can_invalid', 'permanent_eps', 'critical_event'])
+def test_repeated_warning_join_preserves_denials(fault):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    sd, _, step, card, receive = case._ioniq_availability_callers()
+    for tick in (2, 3):
+      step(tick, car_output_valid=False, native_request_long=False)
+      assert receive(10_000_000_000 + tick * 10_000_000, axis=sd.pm.messages['aolAxisState'],
+                     event=sd.pm.messages['onroadEvents'])
+    step(4, car_output_valid=False, native_request_long=False)
+    now = 10_040_000_000
+    proof = sd.pm.messages['aolAxisState'].as_builder()
+    proof.aolAxisState.faultEventMonoTime = now
+    observed = fixture.car_state()
+    frame = {'axis': proof}
+    if fault == 'producer':
+      proof.aolAxisState.intentProducerSessionId = 'other-card'
+    elif fault == 'epoch':
+      proof.aolAxisState.intentIntegrityEpoch += 1
+    elif fault == 'session':
+      proof.aolAxisState.sessionId = 'other-session'
+    elif fault == 'fault_session':
+      proof.aolAxisState.faultSessionId = 'other-session'
+    elif fault == 'stale_source':
+      proof.aolAxisState.sourceCarStateMonoTime = now - 30_000_001
+    elif fault == 'future_source':
+      proof.aolAxisState.sourceCarStateMonoTime = now + 1
+    elif fault == 'invalid_native':
+      sd.sm.valid['aolSafetyWire'] = False
+    elif fault == 'can_invalid':
+      observed.canValid = False
+    elif fault == 'permanent_eps':
+      observed.steerFaultPermanent = True
+    else:
+      event = messaging.new_message('onroadEvents', 1, valid=True, logMonoTime=now)
+      event.onroadEvents[0].name = 'canError'
+      event.onroadEvents[0].immediateDisable = True
+      frame['event'] = event
+    assert not receive(now, state=observed, **frame)
+    assert not card.aol_card_intent.allowed_latch

openpilot/starpilot/aol/transport_pause.py

@@ -440,9 +440,6 @@ class TransportPauseFeedback:
       return False
     warning = any(event.softDisable for event in events)
     recovering = bool(getattr(self, 'ordinary_warning_retained', False) and not warning)
-    if pending and (axis.faultEventMonoTime != self.pending_event or
-        not 0 < self.pending_event <= now_ns or now_ns - self.pending_event > AXIS_MAX_AGE_NS):
-      return False
     stock_user = stock_lateral_user_event(cp)
     # Family button/pedal latch clear: no rearm.
     if any(event.userDisable and event.name != stock_user and not (
@@ -452,6 +449,20 @@ class TransportPauseFeedback:
           now_ns - event_ns <= SAFETY_MAX_AGE_NS)) for event in events):
       return False
     prior = pending_baseline if pending else self.signature
+    repeated_join = bool(warning and self.ordinary_warning_retained and prior is not None and
+      prior[5] == 'availabilityPause' and not any((prior[9], prior[10], prior[12])) and
+      prior[14:16] == (producer, self.availability_epoch) and
+      str(axis.faultReason) == 'availabilityPause' and not axis.lateralActive and
+      not axis.longitudinalActive and not axis.desiredLongitudinal and
+      (axis.desiredLateral, axis.nativeAcknowledged) == (prior[11], prior[13]) and
+      0 < axis.faultEventMonoTime <= stamp and
+      0 < prior[6] == min(axis.faultEventMonoTime, event_ns) < max(axis.faultEventMonoTime, event_ns) <= now_ns and
+      now_ns - max(axis.faultEventMonoTime, event_ns) <= AXIS_MAX_AGE_NS)
+    if pending and (not 0 < self.pending_event <= now_ns or
+        now_ns - self.pending_event > AXIS_MAX_AGE_NS or
+        (max(axis.faultEventMonoTime, event_ns) != self.pending_event if repeated_join else
+         axis.faultEventMonoTime != self.pending_event)):
+      return False
     # Pre-event SOFT: intent only, fixed original 30ms join; no newer/duplicate withdrawal reanchoring.
     axis_first = bool(str(axis.faultReason) == 'availabilityPause' and
       not (axis.lateralActive or axis.longitudinalActive or axis.desiredLateral or
@@ -471,21 +482,21 @@ class TransportPauseFeedback:
       not axis.desiredLongitudinal and
       (warning and str(axis.faultReason) == 'availabilityPause' and axis.faultEventMonoTime == event_ns <= stamp or
        recovering and str(axis.faultReason) == 'none'))
-    if not active_join and not withdrawn and not axis_first:
+    if not active_join and not withdrawn and not axis_first and not repeated_join:
       return False
-    if pending and not axis_first and not (withdrawn and warning and event_ns == self.pending_event):
+    if pending and not axis_first and not repeated_join and not (withdrawn and warning and event_ns == self.pending_event):
       return False
     native = _configured_transport_native(sm, cp, self.session, now_ns,
-      allow_restricted_lateral=withdrawn or axis_first or bool(cs.steerFaultTemporary),
-      require_longitudinal=not (withdrawn or axis_first))
+      allow_restricted_lateral=withdrawn or axis_first or repeated_join or bool(cs.steerFaultTemporary),
+      require_longitudinal=not (withdrawn or axis_first or repeated_join))
     if (native is None or not native.pandaInventory or
         (native.pandaSerial, tuple(slot.hardwareSerial for slot in native.pandaInventory)) != self.lateral_inventory or
         native.observedMonoTime < self.lateral_native_ns or
         active_join and not (native.requestedLateral and native.lateralAllowed)):
       return False
-    if axis_first:
+    if axis_first or repeated_join:
       if not pending:
-        self.pending_event = int(axis.faultEventMonoTime)
+        self.pending_event = max(int(axis.faultEventMonoTime), event_ns)
         self.ordinary_warning_pending_baseline = prior
       self.awaiting = True
       return True

openpilot/starpilot/models/recovery_process.py

@@ -114,7 +114,7 @@ class ModeldProcess(PythonProcess):
     if not self.context_ready or self.recovery.phase in ('sigint', 'sigkill', 'failed'):
       return
     if self.shutting_down:
-      self.stop()
+      self.stop(block=False)
     if self.proc is not None:
       return
     self.proc = Process(name=self.name, target=modeld_launcher,

openpilot/system/manager/process.py

@@ -67,6 +67,7 @@ class ManagerProcess(ABC):
   enabled = True
   name = ""
   shutting_down = False
+  stop_requested_at: float | None = None
 
   @abstractmethod
   def start(self) -> None:
@@ -83,9 +84,12 @@ class ManagerProcess(ABC):
           sig = signal.SIGKILL if self.sigkill else signal.SIGINT
         self.signal(sig)
         self.shutting_down = True
+        self.stop_requested_at = time.monotonic()
 
-        if not block:
-          return None
+      if not block:
+        if retry and self.stop_requested_at is not None and time.monotonic() - self.stop_requested_at >= 5:
+          self.signal(signal.SIGKILL)
+        return None
 
       join_process(self.proc, 5)
 
@@ -100,6 +104,7 @@ class ManagerProcess(ABC):
 
     if self.proc.exitcode is not None:
       self.shutting_down = False
+      self.stop_requested_at = None
       self.proc = None
 
     return ret
@@ -143,7 +148,7 @@ class NativeProcess(ManagerProcess):
   def start(self) -> None:
     # In case we only tried a non blocking stop we need to stop it before restarting
     if self.shutting_down:
-      self.stop()
+      self.stop(block=False)
 
     if self.proc is not None:
       return
@@ -195,7 +200,7 @@ class PythonProcess(ManagerProcess):
   def start(self) -> None:
     # In case we only tried a non blocking stop we need to stop it before restarting
     if self.shutting_down:
-      self.stop()
+      self.stop(block=False)
 
     if self.proc is not None:
       return

openpilot/system/manager/test/test_nonblocking_stop.py

@@ -0,0 +1,76 @@
+import signal
+from unittest.mock import Mock, call, patch
+
+import pytest
+
+from openpilot.starpilot.models.recovery_process import ModeldProcess
+from openpilot.system.manager import process
+
+
+def owner(kind):
+  if kind == 'native':
+    return process.NativeProcess('worker', '.', ['unused'], lambda *_: True)
+  if kind == 'modeld':
+    result = ModeldProcess('modeld', 'unused', lambda *_: True)
+    result.context_ready = True
+    return result
+  return process.PythonProcess('starportd', 'unused', lambda *_: True)
+
+
+@pytest.mark.parametrize('kind', ['python', 'native', 'modeld'])
+@pytest.mark.parametrize('restart', [False, True])
+def test_polling_stop_and_restart_never_wait_for_a_live_child(kind, restart):
+  p = owner(kind)
+  child = Mock(pid=123, exitcode=None)
+  p.proc = child
+  now = [100.]
+  with patch.object(process.time, 'monotonic', side_effect=lambda: now[0]), \
+       patch.object(process.os, 'kill') as kill, \
+       patch.object(process, 'join_process', side_effect=AssertionError('NONBLOCKING_STOP_WAITED')), \
+       patch.object(process, 'Process', side_effect=AssertionError('OVERLAPPING_CHILD')):
+    p.stop(block=False)
+    for elapsed in (0.5, 1., 4.99):
+      now[0] = 100. + elapsed
+      if restart:
+        p.start()
+      else:
+        p.stop(block=False)
+      assert p.proc is child and p.shutting_down
+      assert p.stop_requested_at == 100.
+      assert kill.call_args_list == [call(123, signal.SIGINT)]
+    now[0] = 105.
+    p.stop(block=False)
+    assert kill.call_args_list == [call(123, signal.SIGINT), call(123, signal.SIGKILL)]
+    child.join.assert_not_called()
+    child.exitcode = -signal.SIGKILL
+    assert p.stop(block=False) == -signal.SIGKILL
+    assert p.proc is None and not p.shutting_down and p.stop_requested_at is None
+
+
+def test_nonblocking_stop_retry_false_preserves_original_escalation_deadline():
+  p = owner('python')
+  p.proc = Mock(pid=123, exitcode=None)
+  now = [100.]
+  with patch.object(process.time, 'monotonic', side_effect=lambda: now[0]), \
+       patch.object(process.os, 'kill') as kill, \
+       patch.object(process, 'join_process', side_effect=AssertionError('NONBLOCKING_STOP_WAITED')):
+    p.stop(retry=False, block=False)
+    now[0] = 106.
+    p.stop(retry=False, block=False)
+    assert kill.call_count == 1
+    p.stop(block=False)
+    assert kill.call_args.args == (123, signal.SIGKILL)
+
+
+@pytest.mark.parametrize('retry', [False, True])
+def test_explicit_blocking_shutdown_keeps_wait_and_kill_behavior(retry):
+  p = owner('python')
+  child = Mock(pid=123, exitcode=None)
+  p.proc = child
+  child.join.side_effect = lambda: setattr(child, 'exitcode', -signal.SIGKILL)
+  with patch.object(process.os, 'kill') as kill, patch.object(process, 'join_process') as wait:
+    result = p.stop(retry=retry)
+    wait.assert_called_once_with(child, 5)
+    assert kill.call_args.args == (123, signal.SIGKILL if retry else signal.SIGINT)
+    assert child.join.call_count == int(retry)
+    assert result == (-signal.SIGKILL if retry else None)

openpilot/system/manager/test/test_process_recovery.py

@@ -173,6 +173,11 @@ class TestProcessRecovery(unittest.TestCase):
         self.assertTrue(old.joined)
       self.step(p, started=False)
       self.assertTrue(p.shutting_down)
+      stopping = p.proc
+      self.step(p, started=False)
+      self.assertIs(p.proc, stopping)
+      assert isinstance(stopping, Child)
+      stopping.exitcode = 0
       self.step(p, started=False)
       self.assertIsNone(p.proc)
       self.now += 60

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "8b02ec20dd62df5e56337f2378dc6ee87aba456cc486b14ad99f4b25f0992034", "schema_version": 1}
+{"protected_inventory_sha256": "6ced2806b5125a6965c378b25c736e83d0187cd1fa2c0770baeed339d5ac0887", "schema_version": 1}