diff --git a/openpilot/starpilot/flm/steerlearnerd.py b/openpilot/starpilot/flm/steerlearnerd.py index cede463849..c6ad42cf39 100644 --- a/openpilot/starpilot/flm/steerlearnerd.py +++ b/openpilot/starpilot/flm/steerlearnerd.py @@ -599,8 +599,9 @@ 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) - defer_gc = bool(gc_initially_enabled and state is not None and state['phase'] == 'trial' and - not state.get('accepted', False) and clock() <= state['deadline_ns']) + 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 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; @@ -619,8 +620,10 @@ def run( pending_stamp = getattr(pending_poses[0], 'mono_ns', 0) pending_now = clock() pending_fresh = type(pending_stamp) is int and 0 < pending_stamp <= pending_now <= pending_stamp + MAX_GAP_NS + # Keep the genuine tapes current between poses during a bounded trial; + # otherwise each motion arrival starts behind another full FIFO batch. delivered_pose_header = (source_delivery.next(sm, pending_evidence=(control_socket, applied_socket)) - if pending_fresh else source_delivery.next(sm)) + if pending_fresh or pending_trial else source_delivery.next(sm)) source_rejected = False now = clock() # This is an ORIGINAL header from the real poll or an already consumed diff --git a/openpilot/starpilot/flm/tests/test_pending_motion_publication.py b/openpilot/starpilot/flm/tests/test_pending_motion_publication.py index 978ee99eb4..fe5fac2f58 100644 --- a/openpilot/starpilot/flm/tests/test_pending_motion_publication.py +++ b/openpilot/starpilot/flm/tests/test_pending_motion_publication.py @@ -27,7 +27,7 @@ def test_absent_pending_pose_does_not_start_extra_observation(): assert not steerlearnerd._pending_pose_bracket_ready(S(), S(state_ns=BASE), BASE) -@pytest.mark.parametrize('missing', ['state_bracket', 'native_pair', 'delivery_in_batch', 'delivery_received_by_native_refresh']) +@pytest.mark.parametrize('missing', ['state_bracket', 'native_pair', 'delivery_in_batch', 'delivery_received_by_native_refresh', 'pending_trial_ingest']) @pytest.mark.parametrize('retained_header', [False, True]) def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeypatch, missing, retained_header): now = [BASE-25_000_000] @@ -94,6 +94,8 @@ def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeyp def tick(self, *args, **kwargs): self.calls += 1 assert self.calls == 1 + if missing == 'pending_trial_ingest': + self.ingress = S(optimizer=S(offer_state={'phase':'trial', 'accepted':False, 'deadline_ns':old.deadline_ns})) return old def mark_publication(self, value): self.published.append(value.sequence) @@ -104,6 +106,17 @@ def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeyp return True actor = Actor() + control_tape, output_tape = object(), object() + original_next = steerlearnerd._SourceDelivery.next + def observe_wait(delivery, sm, *, pending_evidence=()): + if missing == 'pending_trial_ingest' and actor.calls: + assert not collector._original_source_pose_queue + assert pending_evidence == (control_tape, output_tape) + # The fixture owns the arrival schedule below; actual socket polling is + # separately tested. This assertion observes the real daemon's API call. + return original_next(delivery, sm) + return original_next(delivery, sm, pending_evidence=pending_evidence) + monkeypatch.setattr(steerlearnerd._SourceDelivery, 'next', observe_wait) fitter = S(cancel=lambda *args:None, close=lambda *args:True) history = S(offer=None, requalifier=None, reset=lambda:None, advance=lambda *args, **kwargs:None, observe_response_recovery=lambda *args:False) @@ -158,7 +171,7 @@ def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeyp raise Stop monkeypatch.setattr(steerlearnerd, 'drain_pair', drain) with pytest.raises(Stop): - steerlearnerd.run(S(), SM(), object(), object(), S(send=send), lambda *args, **kwargs:adapter, + steerlearnerd.run(S(), SM(), control_tape, output_tape, S(send=send), lambda *args, **kwargs:adapter, None, source, lambda *args:('objective',), lambda *args:'objective', None, None, fitter, None, clock=lambda:now[0], live_actor=actor) fresh_sends = [stamp for stamp, version, sequence in sent if version == 1 and sequence == 2] diff --git a/prebuilt.json b/prebuilt.json index 877c445273..0c0b0646c6 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "f0549dd2e30990df108d65ab4321b4671fcf80b9b1564d47ed03b62892cb810b", "schema_version": 1} +{"protected_inventory_sha256": "c05bcc6b9f084bfe824655bcd33496f3ef67827c22721fb29e70af3c3e151caa", "schema_version": 1}