From 0f57b620ef8dcf11a385cfc2f35b4312cca6c9c2 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Tue, 3 Nov 2026 05:13:06 -0600 Subject: [PATCH] Keep learner confirmation pending through acknowledgement --- openpilot/starpilot/flm/steerlearnerd.py | 13 +- .../flm/tests/test_confirmation_gc.py | 115 ++++++++++++++++++ prebuilt.json | 2 +- 3 files changed, 123 insertions(+), 7 deletions(-) create mode 100644 openpilot/starpilot/flm/tests/test_confirmation_gc.py diff --git a/openpilot/starpilot/flm/steerlearnerd.py b/openpilot/starpilot/flm/steerlearnerd.py index 86d6ad8f55..592de685a9 100644 --- a/openpilot/starpilot/flm/steerlearnerd.py +++ b/openpilot/starpilot/flm/steerlearnerd.py @@ -599,13 +599,14 @@ def run( while True: state = (live_actor.ingress.optimizer.offer_state if live_actor is not None and live_actor.ingress is not None else None) - pending_trial = bool(state is not None and state['phase'] == 'trial' and - not state.get('accepted', False) and clock() <= state['deadline_ns']) - defer_gc = gc_initially_enabled and pending_trial + pending_application = bool(state is not None and state['phase'] in ('trial', 'confirmed') and + not state.get('accepted', False) and clock() <= state['deadline_ns']) + defer_gc = gc_initially_enabled and pending_application if defer_gc and not trial_gc_deferred: # This separate optional process must receive the real 100Hz tape during - # the existing bounded trial. Reference counting continues normally; - # cyclic collection resumes on rollback, completion or the same deadline. + # the bounded trial AND its unacknowledged confirmation. Reference + # counting continues; cyclic GC resumes after actual acceptance, + # rollback, context loss or the unchanged original deadline. gc.disable() trial_gc_deferred = True elif trial_gc_deferred and not defer_gc: @@ -623,7 +624,7 @@ def run( # Bootstrap also needs evidence wakeups; waiting for motion lets the # full-rate tapes spend their original freshness budget before decoding. delivered_pose_header = (source_delivery.next(sm, pending_evidence=(control_socket, applied_socket)) - if pending_fresh or pending_trial or (live_actor is not None and collector is not None and + if pending_fresh or pending_application or (live_actor is not None and collector is not None and getattr(collector.adapter, 'event_collection', False)) else source_delivery.next(sm)) source_rejected = False now = clock() diff --git a/openpilot/starpilot/flm/tests/test_confirmation_gc.py b/openpilot/starpilot/flm/tests/test_confirmation_gc.py new file mode 100644 index 0000000000..8fe4f26113 --- /dev/null +++ b/openpilot/starpilot/flm/tests/test_confirmation_gc.py @@ -0,0 +1,115 @@ +"""Actual daemon-loop GC lifecycle; declared actor state is NOT permission.""" + +import gc +from types import SimpleNamespace + +import pytest + +from openpilot.starpilot.flm import steerlearnerd +from openpilot.starpilot.flm.tests.test_live_tracking_daemon import BASE, source, RecordingActor, NoLegacyFitter, LoopComplete + + +def state(phase, *, accepted=False, expired=False): + return {'phase': phase, 'accepted': accepted, 'deadline_ns': BASE - 1 if expired else BASE + 120_000_000_000} + + +def run_script(monkeypatch, states, *, initially_enabled=True, stop_pending=False, acknowledge_in_drain=False): + prior = gc.isenabled() + (gc.enable if initially_enabled else gc.disable)() + now = [BASE] + observations, polls, persistence_gc = [], [], [] + + class Actor(RecordingActor): + def __init__(self): + super().__init__() + self.index = 0 + self.set_state(states[0]) + + def set_state(self, value): + self.ingress = None if value is None else SimpleNamespace(optimizer=SimpleNamespace(offer_state=value)) + + def tick(self, *args, **kwargs): + super().tick(*args, **kwargs) + self.index += 1 + if self.index < len(states): + self.set_state(states[self.index]) + return None + + def persistence_inputs(self, now_ns): + current = None if self.ingress is None else self.ingress.optimizer.offer_state + if current is not None and current['phase'] == 'confirmed' and current['accepted']: + persistence_gc.append(gc.isenabled()) + return None # Never fabricate a native-authorized save input. + + actor, fitter = Actor(), NoLegacyFitter() + + class SM: + updated = {'deviceMotion': False} + logMonoTime = {'deviceMotion': 0} + sock = {'deviceMotion': object()} + + def update(self, timeout=None): + if actor.index == len(states): + raise LoopComplete + current = None if actor.ingress is None else actor.ingress.optimizer.offer_state + observations.append((current, gc.isenabled())) + if stop_pending: + raise LoopComplete + now[0] += 10_000_000 + + class Poller: + def registerSocket(self, _socket): + pass + + def poll(self, timeout): + polls.append(timeout) + + acknowledged = False + + def drain(*args): + nonlocal acknowledged + current = None if actor.ingress is None else actor.ingress.optimizer.offer_state + if acknowledge_in_drain and current is not None and current['phase'] == 'confirmed' and not acknowledged: + # Only a declared loop-state transition, no application/native evidence. + current['accepted'] = True + acknowledged = True + + monkeypatch.setattr(steerlearnerd.messaging, 'Poller', Poller) + monkeypatch.setattr(steerlearnerd, 'drain_pair', drain) + try: + with pytest.raises(LoopComplete): + steerlearnerd.run(SimpleNamespace(), SM(), object(), object(), SimpleNamespace(send=lambda *args: None), + None, None, lambda stamp: source(stamp), None, None, None, None, fitter, None, + clock=lambda: now[0], live_actor=actor, live_persistence=object()) + assert gc.isenabled() is initially_enabled + assert actor.closed and fitter.closed + finally: + (gc.enable if prior else gc.disable)() + return observations, polls, persistence_gc + + +@pytest.mark.parametrize('initially_enabled', (True, False)) +def test_actual_loop_keeps_confirmation_handoff_pending_then_restores_gc(monkeypatch, initially_enabled): + states = [state('trial'), state('confirmed'), state('confirmed', accepted=True), state('rollback'), None] + observed, polls, _ = run_script(monkeypatch, states, initially_enabled=initially_enabled) + assert [enabled for _, enabled in observed] == [False, False, initially_enabled, initially_enabled, initially_enabled] + assert polls == [10, 10] # Same evidence sockets wake trial AND confirmation. + + +def test_ack_and_same_iteration_persistence_check_precede_gc_reenable(monkeypatch): + observed, _polls, persistence_gc = run_script(monkeypatch, [state('confirmed')], acknowledge_in_drain=True) + assert [enabled for _, enabled in observed] == [False] + assert persistence_gc == [False] + + +@pytest.mark.parametrize('pending', (state('confirmed', expired=True), state('confirmed', accepted=True), state('rollback'), None)) +def test_accepted_rollback_context_loss_and_original_expiry_do_not_defer(monkeypatch, pending): + observed, polls, _ = run_script(monkeypatch, [pending]) + assert [enabled for _, enabled in observed] == [True] + assert not polls + + +@pytest.mark.parametrize('initially_enabled', (True, False)) +def test_shutdown_during_pending_confirmation_restores_original_gc(monkeypatch, initially_enabled): + observed, _polls, _ = run_script(monkeypatch, [state('confirmed')], initially_enabled=initially_enabled, stop_pending=True) + assert [enabled for _, enabled in observed] == [False] diff --git a/prebuilt.json b/prebuilt.json index 8e7d64d940..2d25e27ce0 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "2ba5ea84e77e3acc80ba45d12b36967041d963aa70dfbb57ef1485802da946a4", "schema_version": 1} +{"protected_inventory_sha256": "87f23eb9a7364f77ff02024c2899f34dcf1305ff91a51057a61d8a7602dd91aa", "schema_version": 1}